import { randomUUID } from 'node:crypto' import { z } from 'zod' import { AiModelError, assertAiModelAdmin, assertAiModelAdminForWrite } from './ai-model-access' import { aiModelAudit, aiModelTransaction } from './ai-model-repository' import { getAiModelAdminState, resolveAiModelVerification } from './ai-model-service' import { AI_MODEL_ADAPTER_VERSION, AiModelProtocolError, callAiModelProtocol, validateStoredAiModelProtocols, type AiModelProtocol } from './ai-model-protocol' import { AI_MODEL_TEST_PURPOSES, aiModelTestStages } from './ai-model-types' import { AI_MODEL_VERIFICATION_COLUMNS, lockAiModelVerificationExecution, verificationExecutionId, type AiModelVerificationRow } from './ai-model-verification-state' import { prisma } from './prisma' import { AI_MODEL_VERIFICATION_RULE_VERSION, createVerificationChallenge, readVerificationChallenge, validateAiModelCombinedVerificationOutput, validateAiModelVerificationOutput, verificationReferenceImage, type AiModelTextCheck } from './ai-model-verification-output' export { validateAiModelVerificationOutput } from './ai-model-verification-output' import { detailTransactionLock } from './detail-generator/repository' import type { AiModelDatabase } from './ai-model-repository' import { createMultiReferenceChallenge, multiReferenceImages, multiReferencePrompt, readMultiReferenceChallenge, validateMultiReferenceOutput } from './ai-model-multireference' const requestSchema = z.object({ stage: z.enum(AI_MODEL_TEST_PURPOSES), revision: z.number().int().positive(), idempotencyKey: z.string().uuid(), referenceImageCount: z.number().int().min(1).max(8).optional() }).strict() function safeId(value: unknown, secret: string): string | null { return typeof value === 'string' && /^[\w.-]{1,160}$/.test(value) && !value.includes(secret) ? value : null } /** Persist only bounded adapter facts; model output cannot supply or override validation evidence. */ function imageTransportEvidence(value: unknown, protocol: AiModelProtocol, secret: string): Record { if (!value || typeof value !== 'object' || Array.isArray(value)) return {} const summary = value as Record, evidence: Record = {} if ((summary.referenceTransport === 'multipart' || summary.referenceTransport === 'json-inline') && summary.referenceTransport === protocol.referenceTransport) { evidence.referenceTransport = summary.referenceTransport const actual = validateStoredAiModelProtocols([{ ...protocol, requestParameters: summary.requestParameters }])[0] const expected = validateStoredAiModelProtocols([protocol])[0] // Use the adapter's existing parameter allowlist, and require the actual // values to match the route whose context was locked for this execution. if (actual?.requestParameters && expected?.requestParameters && JSON.stringify(actual.requestParameters) === JSON.stringify(expected.requestParameters) && (!secret || !JSON.stringify(actual.requestParameters).includes(secret))) evidence.requestParameters = actual.requestParameters } for (const [widthKey, heightKey, maximum] of [['generatedWidth', 'generatedHeight', 40_000_000], ['storedWidth', 'storedHeight', 1440]] as const) { const width = summary[widthKey], height = summary[heightKey] if (typeof width === 'number' && typeof height === 'number' && Number.isInteger(width) && Number.isInteger(height) && width > 0 && height > 0 && width <= maximum && height <= maximum && width * height <= 40_000_000) { evidence[widthKey] = width; evidence[heightKey] = height } } return evidence } async function assertExecutionSlot(db: AiModelDatabase, actorId: string, resumeId: string | null = null): Promise { await detailTransactionLock(db, 'detail-generator-execution') const rows = await db.$queryRawUnsafe>(`SELECT COUNT(*)::int AS count,COUNT(*) FILTER (WHERE actor_id=$1)::int AS own FROM ( SELECT DISTINCT COALESCE(evidence->>'executionGroupId',id) AS execution_id,actor_id FROM ai_model_verifications WHERE ($2::text IS NULL OR COALESCE(evidence->>'executionGroupId',id)<>$2) AND status IN ('running','unknown') UNION ALL SELECT id AS execution_id,owner_id AS actor_id FROM detail_generator_jobs WHERE status='running' AND lease_expires_at>NOW() OR status='queued' AND checkpoint->>'dispatchPending'='true' AND COALESCE(checkpoint #>> '{upstreamTask,id}', '') ~ '^[a-zA-Z0-9][a-zA-Z0-9_.:-]{0,199}$' ) AS active_executions`, actorId, resumeId) if (rows[0]?.count >= 2 || rows[0]?.own >= 1) throw new AiModelError('已有生成或模型测试正在执行,请等待完成', 409, 'verification_busy') } type VerificationContext = Awaited> function assertCombinedContexts(contexts: VerificationContext[]): void { const transport = (context: VerificationContext) => JSON.stringify({ ...context.protocol, supportsImageInput: undefined }) if (contexts.length !== 2 || !contexts[0].protocol.supportsImageInput || transport(contexts[0]) !== transport(contexts[1])) throw new AiModelError('当前模型不能在同一路由联合验证分析和文案', 400, 'verification_combined_unsupported') } async function executeVerification(rows: AiModelVerificationRow[], userId: string, providerId: string, resume: boolean, retainedRequestId?: string): Promise { const row = rows[0], combined = rows.length === 2 let sent = false, resultReceived = false, apiKey = '', knownTaskId = row.taskId, knownRequestId: string | null = null const beforeRequest = async () => { await assertAiModelAdmin(userId) const current = await Promise.all(rows.map(async item => { const context = await resolveAiModelVerification(providerId, item.modelKey, item.stage) if (context.contextHash !== item.contextHash) throw new AiModelError('连接或路由已改变,测试已停止', 409, 'verification_context_changed') const pending = await prisma.$queryRawUnsafe>("SELECT id FROM ai_model_verifications WHERE id=$1 AND status='running' AND deadline_at>NOW()", item.id) if (!pending[0]) throw new AiModelError('原请求已暂停或核查结束,不能继续调用', 409, 'verification_closed') return context })) if (combined) assertCombinedContexts(current) sent = true } try { const contexts = await Promise.all(rows.map(async item => { const context = await resolveAiModelVerification(providerId, item.modelKey, item.stage) if (context.contextHash !== item.contextHash) throw new AiModelError('验证对应的连接或路由已改变,不能继续此任务', 409, 'verification_context_changed') return context })) if (combined) assertCombinedContexts(contexts) const context = contexts[0] apiKey = context.apiKey knownRequestId = safeId(row.requestId, apiKey) if (context.contextHash !== row.contextHash) throw new AiModelError('验证对应的连接或路由已改变,不能继续此任务', 409, 'verification_context_changed') const challenge = readVerificationChallenge(row.evidence?.challenge) const multi = row.evidence?.multiReferenceChallenge ? readMultiReferenceChallenge(row.evidence.multiReferenceChallenge) : null if (multi && (row.stage !== 'image' || multi.references.length > (context.protocol.maxImages ?? 8))) throw new AiModelError('多图测试数量与当前协议不一致', 409, 'verification_checkpoint_invalid') const prompt = multi ? multiReferencePrompt(multi) : combined ? '一次完成两项独立任务。analysis:必须读取参考图,识别左上与右下方块的颜色,以英文小写填写。copy:真实产品资料为型号TEST、颜色蓝色,图片只是读图测试图,不是该产品外观;只依据这两条真实资料写一屏中文详情页,不编造材质、尺寸、性能或认证。严格只输出一个JSON对象:{"analysis":{"topLeft":"英文小写颜色","bottomRight":"英文小写颜色"},"copy":{"sellingPoints":["卖点"],"warnings":[],"screens":[{"title":"标题","body":"正文","imagePrompt":"无文字的场景描述"}]}}。' : row.stage === 'analysis' ? '观察图片:左上方方块与右下方方块分别是什么颜色?只输出 JSON:{"topLeft":"英文小写颜色","bottomRight":"英文小写颜色"}。不要根据文字猜测,必须读取图片。' : row.stage === 'copy' ? '真实产品资料:型号TEST,颜色蓝色。不要编造其他规格。写一屏中文详情页,严格只输出 JSON:{"sellingPoints":["卖点"],"warnings":[],"screens":[{"title":"标题","body":"正文","imagePrompt":"无文字的场景描述"}]}。' : '将参考图重新绘制为干净的平面图,严格保留两个方块各自的原始颜色、形状和相对位置,纯白背景;不要添加文字、标志、阴影、纹理、渐变、光照或其它内容。' const result = await callAiModelProtocol({ baseUrl: context.provider.baseUrl, apiKey: context.apiKey, modelId: context.model.id, protocol: context.protocol, prompt, ...(row.stage !== 'copy' ? { images: multi ? await multiReferenceImages(multi) : [{ buffer: await verificationReferenceImage(challenge), mimeType: 'image/png' }] } : {}), maxOutputTokens: combined ? 1800 : row.stage === 'copy' ? 1200 : 1024, idempotencyKey: row.idempotencyKey, beforeRequest, ...(retainedRequestId ? { resumeRequestId: retainedRequestId } : resume ? { resumeTaskId: row.taskId! } : {}), onTaskSubmitted: async taskId => { if (!safeId(taskId, context.apiKey)) throw new AiModelError('上游任务编号无效', 502, 'verification_task_invalid') knownTaskId = taskId await aiModelTransaction(async db => { for (const item of rows) { const updated = await db.$executeRawUnsafe(`UPDATE ai_model_verifications SET task_id=$2 WHERE id=$1 AND status='running' AND (task_id IS NULL OR task_id=$2)`, item.id, taskId) if (updated !== 1) throw new AiModelError('任务检查点未保存,已暂停后续请求', 409, 'verification_checkpoint_failed') } }) } }) resultReceived = true // The native protocol has returned a terminal, decoded result. knownRequestId = safeId(result.requestId, context.apiKey) || knownRequestId const checks: Partial> = combined ? await validateAiModelCombinedVerificationOutput(result, challenge) : { [row.stage]: { status: 'passed', message: multi ? `${multi.references.length} 张参考图逐张内容校验通过` : '实际接口测试通过', evidence: { ...(row.stage === 'image' ? imageTransportEvidence(result.summary, context.protocol, context.apiKey) : {}), ...(retainedRequestId ? { retainedResponse: true } : {}), ...(multi ? await validateMultiReferenceOutput(result.image, multi) : await validateAiModelVerificationOutput(row.stage, result, challenge)), } } } await assertAiModelAdmin(userId) await aiModelTransaction(async db => { await assertAiModelAdminForWrite(userId, db) for (const item of rows) { const current = await resolveAiModelVerification(providerId, item.modelKey, item.stage) if (current.contextHash !== item.contextHash) throw new AiModelError('验证期间连接或路由已改变', 409, 'verification_context_changed') } const changes: number[] = [] for (let index = 0; index < rows.length; index++) { const item = rows[index] const check = checks[item.stage]!, protocol = contexts[index].protocol changes.push(await db.$executeRawUnsafe(`UPDATE ai_model_verifications SET status='${check.status}',request_id=$2,message=$4,evidence=evidence || $3::jsonb,finished_at=NOW() WHERE id=$1 AND status='running' AND deadline_at>NOW()`, item.id, safeId(result.requestId, context.apiKey), JSON.stringify({ ...check.evidence, ruleVersion: AI_MODEL_VERIFICATION_RULE_VERSION, adapterVersion: AI_MODEL_ADAPTER_VERSION, protocol: protocol.id, scopeId: protocol.scopeId, routeId: protocol.routeId, operationId: protocol.operationId }), check.message)) } if (combined && changes.some(value => value !== changes[0])) throw new AiModelError('联合测试记录已改变,已停止写入', 409, 'verification_checkpoint_failed') for (let index = 0; index < rows.length; index++) if (changes[index] === 1) { const item = rows[index] const check = checks[item.stage]! await aiModelAudit(db, userId, `model.verification-${check.status}`, providerId, { verificationId: item.id, modelKey: item.modelKey, stage: item.stage, evidence: check.evidence, ...(combined ? { executionGroupId: verificationExecutionId(item) } : {}) }) } }) } catch (error) { const outcome = error as { outcomeUnknown?: boolean; taskId?: string; requestId?: string } const unknown = !resultReceived && (retainedRequestId ? true : typeof outcome.outcomeUnknown === 'boolean' ? (sent || Boolean(knownTaskId)) && outcome.outcomeUnknown : Boolean(knownTaskId) || sent && !(error instanceof AiModelError)) const safeProtocolMessage = error instanceof AiModelProtocolError && (!apiKey || !error.message.includes(apiKey)) ? error.message : null const message = safeProtocolMessage || (unknown ? '请求结果尚未确认,请核查上游记录;禁止重复提交' : error instanceof AiModelError ? error.message : '模型测试未通过') await aiModelTransaction(async db => { for (const item of rows) { const updated = await db.$executeRawUnsafe(`UPDATE ai_model_verifications SET status=$2,request_id=$3,task_id=COALESCE(task_id,$4),message=$5,finished_at=NOW() WHERE id=$1 AND status='running'`, item.id, unknown ? 'unknown' : 'failed', safeId(outcome.requestId, apiKey) || knownRequestId, safeId(outcome.taskId || knownTaskId, apiKey), message) if (updated === 1) await aiModelAudit(db, userId, unknown ? 'model.verification-unknown' : 'model.verification-failed', providerId, { verificationId: item.id, modelKey: item.modelKey, stage: item.stage, ...(combined ? { executionGroupId: verificationExecutionId(item) } : {}) }) } }) } } export async function verifyAiModel(userId: string, providerId: string, modelKey: string, body: unknown) { const parsed = requestSchema.safeParse(body) if (!parsed.success) throw new AiModelError('测试参数无效') await assertAiModelAdmin(userId) const input = parsed.data, stages = aiModelTestStages(input.stage) const contexts = await Promise.all(stages.map(stage => resolveAiModelVerification(providerId, modelKey, stage, input.revision))) if (input.stage === 'analysis-copy') assertCombinedContexts(contexts) const referenceImageCount = input.stage === 'image' ? input.referenceImageCount ?? (contexts[0].protocol.referenceContract === 'probe-required' ? contexts[0].protocol.maxImages ?? 1 : 1) : 1 if (input.referenceImageCount !== undefined && input.stage !== 'image' || referenceImageCount > (contexts[0].protocol.maxImages ?? 8)) throw new AiModelError('参考图测试数量超出当前用途或协议范围', 400) const created = await aiModelTransaction(async db => { await assertAiModelAdminForWrite(userId, db) const prior = await db.$queryRawUnsafe(`SELECT ${AI_MODEL_VERIFICATION_COLUMNS} FROM ai_model_verifications WHERE model_key=$1 AND idempotency_key=$2`, modelKey, input.idempotencyKey) if (prior[0]) { if ((prior[0].evidence?.purpose || prior[0].stage) !== input.stage) throw new AiModelError('该请求编号已用于其他测试,请创建新请求', 409, 'verification_idempotency_conflict') if (Number(prior[0].evidence?.referenceImageCount ?? 1) !== referenceImageCount) throw new AiModelError('该请求编号已用于不同的参考图数量,不能重复提交', 409, 'verification_idempotency_conflict') return null // A repeated HTTP request never repeats a model call. } const closed = await db.$queryRawUnsafe>(`SELECT id FROM ai_model_audit WHERE action='model.verification-intake-cancelled' AND details->>'modelKey'=$1 AND details->>'idempotencyKey'=$2`, modelKey, input.idempotencyKey) if (closed[0]) throw new AiModelError('原提交已关闭,请重新确认测试并创建新请求', 409, 'verification_submission_closed') await db.$executeRawUnsafe(`UPDATE ai_model_verifications SET status='unknown',message='进程或请求中断,结果需核查',finished_at=NOW() WHERE status='running' AND deadline_at<=NOW()`) const active = await db.$queryRawUnsafe>(`SELECT COUNT(*)::int AS count FROM ai_model_verifications WHERE model_key=$1 AND status IN ('running','unknown')`, modelKey) if (active[0]?.count) throw new AiModelError('该模型存在进行中或结果未知的请求,不能重复测试;请先核查原请求', 409, 'verification_unresolved') await assertExecutionSlot(db, userId) const evidence = { challenge: createVerificationChallenge(), purpose: input.stage, ...(input.stage === 'image' ? { referenceImageCount, ...(referenceImageCount > 1 ? { multiReferenceChallenge: createMultiReferenceChallenge(referenceImageCount) } : {}) } : {}), ...(input.stage === 'analysis-copy' ? { executionGroupId: randomUUID() } : {}) } const rows: AiModelVerificationRow[] = [] for (let index = 0; index < stages.length; index++) { const stage = stages[index] const inserted = await db.$queryRawUnsafe(`INSERT INTO ai_model_verifications (id,model_key,stage,context_hash,idempotency_key,actor_id,status,evidence) VALUES ($1,$2,$3,$4,$5,$6,'running',$7::jsonb) RETURNING ${AI_MODEL_VERIFICATION_COLUMNS}`, randomUUID(), modelKey, stage, contexts[index].contextHash, input.idempotencyKey, userId, JSON.stringify(evidence)) rows.push(inserted[0]) } await aiModelAudit(db, userId, 'model.verification-start', providerId, { verificationId: rows[0].id, modelKey, stage: input.stage, verificationIds: rows.map(item => item.id), ...(input.stage === 'analysis-copy' ? { executionGroupId: verificationExecutionId(rows[0]) } : {}) }) return rows }) if (created) await executeVerification(created, userId, providerId, false) return getAiModelAdminState() } /** Fetch an existing task or retained image response. This endpoint can never submit a new generation. */ export async function refreshAiModelVerification(userId: string, providerId: string, verificationId: string, body: unknown = {}) { const parsed = z.object({ requestId: z.string().uuid().optional() }).strict().safeParse(body) if (!parsed.success) throw new AiModelError('请输入有效的客户请求编号') if (!z.string().uuid().safeParse(verificationId).success) throw new AiModelError('测试编号无效') await assertAiModelAdmin(userId) let retainedRequestId: string | undefined const row = await aiModelTransaction(async db => { await assertAiModelAdminForWrite(userId, db) const rows = await db.$queryRawUnsafe(`SELECT ${AI_MODEL_VERIFICATION_COLUMNS} FROM ai_model_verifications WHERE id=$1 FOR UPDATE`, verificationId) const current = rows[0] const requestId = parsed.data.requestId || current?.requestId if (!current || !['unknown','running'].includes(current.status) || !current.taskId && !(current.stage === 'image' && requestId)) throw new AiModelError('没有可以查询的上游任务或客户请求编号', 409, 'verification_not_resumable') if (parsed.data.requestId && (current.taskId || current.requestId && current.requestId !== parsed.data.requestId)) throw new AiModelError('原请求编号已固定,不能替换为其他请求', 409, 'verification_request_conflict') if (current.status === 'running' && current.deadlineAt.getTime() > Date.now()) throw new AiModelError('测试仍在执行,请等待', 409, 'verification_busy') const execution = await lockAiModelVerificationExecution(db, current) const contexts = await Promise.all(execution.map(async item => { if (!['unknown', 'running'].includes(item.status) || item.status === 'running' && item.deadlineAt.getTime() > Date.now()) throw new AiModelError('联合测试仍在执行或已结束,请刷新状态', 409, 'verification_busy') const context = await resolveAiModelVerification(providerId, item.modelKey, item.stage) if (context.contextHash !== item.contextHash) throw new AiModelError('原连接或路由已改变,请在上游核查原任务', 409, 'verification_context_changed') return context })) if (execution.length === 2) assertCombinedContexts(contexts) if (!current.taskId) { if (!z.string().uuid().safeParse(requestId).success) throw new AiModelError('请输入有效的客户请求编号') if (!['openai-image-generate', 'openai-image-edit'].includes(contexts[0].protocol.id)) throw new AiModelError('该协议尚未适配原图片响应取回', 400, 'protocol_unsupported') retainedRequestId = requestId! } await assertExecutionSlot(db, userId, verificationExecutionId(current)) for (const item of execution) { await db.$executeRawUnsafe(`UPDATE ai_model_verifications SET status='running',actor_id=$2,deadline_at=NOW()+INTERVAL '10 minutes',finished_at=NULL,request_id=COALESCE(request_id,$3) WHERE id=$1`, item.id, userId, retainedRequestId || null) if (retainedRequestId) item.requestId = retainedRequestId } await aiModelAudit(db, userId, 'model.verification-query', providerId, { verificationId: current.id, modelKey: current.modelKey, stage: current.stage, ...(retainedRequestId ? { requestId: retainedRequestId, retainedResponse: true } : {}) }) return execution }) await executeVerification(row, userId, providerId, true, retainedRequestId) return getAiModelAdminState() }