root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(models): secure catalog and pin model versions

Browse Source
This commit is contained in:
Jesse_Chen
2026-08-06 20:15:08 +08:00
parent 7040fd998e
commit 28e04857fa
13 changed files with 1426 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/app/api/consult/route.ts
+82 -97
View File
@@ -7,17 +7,20 @@ import {
runConsultationWorkflow,
toAgentConsultationContext,
} from "@/mastra";
import {
languageModelConfigurationMessage,
resolveLanguageModel,
} from "@/mastra/model";
import { languageModelConfigurationMessage } from "@/mastra/model";
import { blocksPromptExtraction } from "@/lib/consult-safety";
import {
consultationEntrypointSchema,
resolveConsultationQuestion,
} from "@/lib/consultation-entrypoint";
import { CreditRpcError, runCreditRpc } from "@/lib/consultation-billing";
import {
authorizeUsage,
completeUsage,
CreditRpcError,
releaseUsage,
} from "@/lib/consultation-billing";
import { reserveConsultationModel } from "@/lib/consultation-model-selection";
import { resolveSessionLanguageModel } from "@/lib/model-catalog";
import { createAdminSupabaseClient } from "@/lib/supabase/admin";
import { createServerSupabaseClient } from "@/lib/supabase/server";
import { streamTextResponse } from "@/lib/stream-text-response";
@@ -45,6 +48,7 @@ export const maxDuration = 60;
const chatRequestMetadataSchema = z.object({
requestId: z.string().uuid(),
sessionId: z.string().uuid(),
modelId: z.string().trim().min(1).max(64),
name: z.string().trim().max(80).optional().default(""),
history: z
@@ -166,38 +170,6 @@ function rangeBoundaryWorkflowContext(
};
}
async function recordModelUsage(
accounting: ReturnType<typeof createAdminSupabaseClient>,
userId: string,
requestId: string,
modelId: string,
usage: Promise<{ inputTokens?: number; outputTokens?: number }>,
) {
try {
const resolved = await usage;
const { error } = await accounting
.from("credit_transactions")
.update({
model: modelId,
input_tokens: Math.max(0, Math.trunc(resolved.inputTokens ?? 0)),
output_tokens: Math.max(0, Math.trunc(resolved.outputTokens ?? 0)),
})
.eq("user_id", userId)
.eq("transaction_type", "reserve")
.eq("request_id", requestId);
if (error)
console.warn(
`[billing] unable to record model usage request=${requestId} model=${modelId}`,
);
} catch (error) {
const reason = error instanceof Error ? error.name : "UnknownError";
console.warn(
`[billing] unable to read model usage request=${requestId} model=${modelId} reason=${reason}`,
);
}
}
export async function POST(request: Request) {
let supabase: Awaited<ReturnType<typeof createServerSupabaseClient>>;
let accounting: ReturnType<typeof createAdminSupabaseClient>;
@@ -232,6 +204,35 @@ export async function POST(request: Request) {
);
}
const { data: chatSession, error: chatSessionError } = await supabase
.from("chat_sessions")
.select("id,model_id,model_config_version,session_type")
.eq("id", parsed.data.sessionId)
.eq("user_id", user.id)
.maybeSingle();
if (chatSessionError) {
return NextResponse.json(
{ error: "暂时无法读取咨询会话", message: "请稍后重试。" },
{ status: 503 },
);
}
if (!chatSession || chatSession.session_type !== "consultation") {
return NextResponse.json(
{ error: "咨询会话不存在", message: "请重新进入咨询。" },
{ status: 404 },
);
}
if (!chatSession.model_id || chatSession.model_id !== parsed.data.modelId) {
return NextResponse.json(
{ error: "会话模型已经变化", message: "请刷新会话后重新发送，本次不会扣点。" },
{ status: 409 },
);
}
const sessionModel = await resolveSessionLanguageModel(
chatSession.model_id,
chatSession.model_config_version,
);
if (parsed.data.entrypoint === "birth_time_rectification") {
return NextResponse.json(
{
@@ -383,20 +384,25 @@ export async function POST(request: Request) {
return data;
},
reserve: () => reserveConsultationModel(
parsed.data.modelId,
resolveLanguageModel,
() => handoffExecution?.billingReused
chatSession.model_id,
(modelId) => sessionModel?.id === modelId ? sessionModel : null,
(model) => handoffExecution?.billingReused
? Promise.resolve({
success: true,
credits: handoffExecution.credits ?? null,
error_code: null,
})
: runCreditRpc(
accounting,
"begin_consultation_credit",
: 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,
})),
),
});
} catch (error) {
@@ -499,12 +505,7 @@ export async function POST(request: Request) {
return;
}
try {
await runCreditRpc(
accounting,
"cancel_consultation_credit",
userId,
requestId,
);
await releaseUsage(accounting, userId, requestId, "consultation_cancelled");
} catch (error) {
const reason = error instanceof Error ? error.name : "UnknownError";
console.error(
@@ -513,17 +514,28 @@ export async function POST(request: Request) {
}
}
async function complete() {
const usageStartedAt = Date.now();
async function complete(usage: Promise<{ inputTokens?: number; outputTokens?: number }>) {
if (handoffExecution?.status === "ready") {
await settleHandoff(true);
return;
}
const result = await runCreditRpc(
accounting,
"complete_consultation_credit",
userId,
requestId,
);
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, {
eventKey: requestId,
actualModelId: selectedModel.id,
modelConfigVersion: selectedModel.configVersion,
inputTokens,
outputTokens,
costMicrousd,
durationMs: Date.now() - usageStartedAt,
});
if (!result.success)
throw new CreditRpcError(result.error_code || "completion_rejected");
}
@@ -552,18 +564,9 @@ export async function POST(request: Request) {
].filter(Boolean).join("\n"),
},
]);
const completeAndRecordUsage = async () => {
await complete();
void recordModelUsage(
accounting,
userId,
requestId,
modelSelection.usageModelId,
result.totalUsage,
);
};
const completeWithUsage = () => complete(result.totalUsage);
const settleInterrupted = (emitted: boolean) =>
settle(emitted ? completeAndRecordUsage : cancel);
settle(emitted ? completeWithUsage : cancel);
return streamTextResponse(result.textStream, {
transformText: createBirthTimeModeOutputGuard(consultationMode, false),
mode: "mastra",
@@ -576,8 +579,8 @@ export async function POST(request: Request) {
"x-jyotish-missing-layers": "birth-minute",
"x-jyotish-birth-time-mode": consultationMode,
},
...(handoff ? { onFirstOutput: () => settle(completeAndRecordUsage) } : {}),
onComplete: () => settle(completeAndRecordUsage),
...(handoff ? { onFirstOutput: () => settle(completeWithUsage) } : {}),
onComplete: () => settle(completeWithUsage),
onError: (_error, emitted) => settleInterrupted(emitted),
onCancel: settleInterrupted,
});
@@ -626,18 +629,9 @@ export async function POST(request: Request) {
].filter(Boolean).join("\n"),
},
]);
const completeAndRecordUsage = async () => {
await complete();
void recordModelUsage(
accounting,
userId,
requestId,
modelSelection.usageModelId,
result.totalUsage,
);
};
const completeWithUsage = () => complete(result.totalUsage);
const settleInterrupted = (emitted: boolean) =>
settle(emitted ? completeAndRecordUsage : cancel);
settle(emitted ? completeWithUsage : cancel);
return streamTextResponse(result.textStream, {
transformText: createBirthTimeModeOutputGuard("unverified_birth_time", false),
mode: "mastra",
@@ -650,8 +644,8 @@ export async function POST(request: Request) {
"x-jyotish-missing-layers": workflowReceipt.missingLayers,
"x-jyotish-birth-time-mode": "unverified_birth_time",
},
onFirstOutput: () => settle(completeAndRecordUsage),
onComplete: () => settle(completeAndRecordUsage),
onFirstOutput: () => settle(completeWithUsage),
onComplete: () => settle(completeWithUsage),
onError: (_error, emitted) => settleInterrupted(emitted),
onCancel: settleInterrupted,
});
@@ -685,18 +679,9 @@ export async function POST(request: Request) {
].filter(Boolean).join("\n"),
},
]);
const completeAndRecordUsage = async () => {
await complete();
void recordModelUsage(
accounting,
userId,
requestId,
modelSelection.usageModelId,
result.totalUsage,
);
};
const completeWithUsage = () => complete(result.totalUsage);
const settleInterrupted = (emitted: boolean) =>
settle(emitted ? completeAndRecordUsage : cancel);
settle(emitted ? completeWithUsage : cancel);
return streamTextResponse(result.textStream, {
transformText: createBirthTimeModeOutputGuard(
consultationMode,
@@ -712,8 +697,8 @@ export async function POST(request: Request) {
"x-jyotish-missing-layers": workflowReceipt.missingLayers,
"x-jyotish-birth-time-mode": consultationMode,
},
...(handoff ? { onFirstOutput: () => settle(completeAndRecordUsage) } : {}),
onComplete: () => settle(completeAndRecordUsage),
...(handoff ? { onFirstOutput: () => settle(completeWithUsage) } : {}),
onComplete: () => settle(completeWithUsage),
onError: (_error, emitted) => settleInterrupted(emitted),
onCancel: settleInterrupted,
});
@@ -721,7 +706,7 @@ export async function POST(request: Request) {
await cancel();
const reason = error instanceof Error ? error.name : "UnknownError";
console.error(
`[consult] generation failed request=${requestId} model=${modelSelection.usageModelId} reason=${reason}`,
`[consult] generation failed request=${requestId} model=${selectedModel.id} reason=${reason}`,
);
return NextResponse.json(
{