117 lines
12 KiB
TypeScript
117 lines
12 KiB
TypeScript
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, callAiModelProtocol } from './ai-model-protocol'
|
||
import { AI_MODEL_STAGES, type AiModelStage } from './ai-model-types'
|
||
import { AI_MODEL_VERIFICATION_COLUMNS, type AiModelVerificationRow } from './ai-model-verification-state'
|
||
import { prisma } from './prisma'
|
||
import { AI_MODEL_VERIFICATION_RULE_VERSION, createVerificationChallenge, readVerificationChallenge, validateAiModelVerificationOutput, verificationReferenceImage } 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'
|
||
|
||
const requestSchema = z.object({ stage: z.enum(AI_MODEL_STAGES), revision: z.number().int().positive(), idempotencyKey: z.string().uuid() }).strict()
|
||
function safeId(value: unknown, secret: string): string | null { return typeof value === 'string' && /^[\w.-]{1,160}$/.test(value) && !value.includes(secret) ? value : null }
|
||
async function assertExecutionSlot(db: AiModelDatabase, actorId: string, resumeId: string | null = null): Promise<void> {
|
||
await detailTransactionLock(db, 'detail-generator-execution')
|
||
const rows = await db.$queryRawUnsafe<Array<{ count: number; own: number }>>(`SELECT COUNT(*)::int AS count,COUNT(*) FILTER (WHERE actor_id=$1)::int AS own FROM (
|
||
SELECT actor_id FROM ai_model_verifications WHERE ($2::text IS NULL OR id<>$2) AND status IN ('running','unknown')
|
||
UNION ALL SELECT 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')
|
||
}
|
||
async function executeVerification(row: AiModelVerificationRow, userId: string, providerId: string, resume: boolean): Promise<void> {
|
||
let sent = false, resultReceived = false, apiKey = '', knownTaskId = row.taskId
|
||
const beforeRequest = async () => {
|
||
await assertAiModelAdmin(userId)
|
||
const current = await resolveAiModelVerification(providerId, row.modelKey, row.stage)
|
||
if (current.contextHash !== row.contextHash) throw new AiModelError('连接或路由已改变，测试已停止', 409, 'verification_context_changed')
|
||
const pending = await prisma.$queryRawUnsafe<Array<{ id: string }>>("SELECT id FROM ai_model_verifications WHERE id=$1 AND status='running' AND deadline_at>NOW()", row.id)
|
||
if (!pending[0]) throw new AiModelError('原请求已暂停或核查结束，不能继续调用', 409, 'verification_closed')
|
||
sent = true
|
||
}
|
||
try {
|
||
const context = await resolveAiModelVerification(providerId, row.modelKey, row.stage)
|
||
apiKey = context.apiKey
|
||
if (context.contextHash !== row.contextHash) throw new AiModelError('验证对应的连接或路由已改变，不能继续此任务', 409, 'verification_context_changed')
|
||
const challenge = readVerificationChallenge(row.evidence?.challenge)
|
||
const prompt = 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: [{ buffer: await verificationReferenceImage(challenge), mimeType: 'image/png' }] } : {}), maxOutputTokens: row.stage === 'copy' ? 1200 : 1024,
|
||
idempotencyKey: row.idempotencyKey, beforeRequest, ...(resume ? { resumeTaskId: row.taskId! } : {}),
|
||
onTaskSubmitted: async taskId => {
|
||
if (!safeId(taskId, context.apiKey)) throw new AiModelError('上游任务编号无效', 502, 'verification_task_invalid')
|
||
knownTaskId = taskId
|
||
const updated = await prisma.$executeRawUnsafe(`UPDATE ai_model_verifications SET task_id=$2 WHERE id=$1 AND status='running' AND (task_id IS NULL OR task_id=$2)`, row.id, taskId)
|
||
if (updated !== 1) throw new AiModelError('任务检查点未保存，已暂停后续请求', 409, 'verification_checkpoint_failed')
|
||
} })
|
||
resultReceived = true // The native protocol has returned a terminal, decoded result.
|
||
const evidence = await validateAiModelVerificationOutput(row.stage, result, challenge)
|
||
await assertAiModelAdmin(userId)
|
||
await aiModelTransaction(async db => {
|
||
await assertAiModelAdminForWrite(userId, db)
|
||
const current = await resolveAiModelVerification(providerId, row.modelKey, row.stage)
|
||
if (current.contextHash !== row.contextHash) throw new AiModelError('验证期间连接或路由已改变', 409, 'verification_context_changed')
|
||
const updated = await db.$executeRawUnsafe(`UPDATE ai_model_verifications SET status='passed',request_id=$2,message='实际接口测试通过',evidence=evidence || $3::jsonb,finished_at=NOW() WHERE id=$1 AND status='running' AND deadline_at>NOW()`, row.id, safeId(result.requestId, context.apiKey), JSON.stringify({ ...evidence, ruleVersion: AI_MODEL_VERIFICATION_RULE_VERSION, adapterVersion: AI_MODEL_ADAPTER_VERSION, protocol: context.protocol.id, scopeId: context.protocol.scopeId, routeId: context.protocol.routeId, operationId: context.protocol.operationId }))
|
||
if (updated === 1) await aiModelAudit(db, userId, 'model.verification-passed', providerId, { verificationId: row.id, modelKey: row.modelKey, stage: row.stage, evidence })
|
||
})
|
||
} catch (error) {
|
||
const outcome = error as { outcomeUnknown?: boolean; taskId?: string; requestId?: string }
|
||
const unknown = !resultReceived && (typeof outcome.outcomeUnknown === 'boolean' ? (sent || Boolean(knownTaskId)) && outcome.outcomeUnknown : Boolean(knownTaskId) || sent && !(error instanceof AiModelError))
|
||
const message = unknown ? '请求结果尚未确认，请核查上游记录；禁止重复提交' : error instanceof AiModelError ? error.message : '模型测试未通过'
|
||
await aiModelTransaction(async db => {
|
||
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'`, row.id, unknown ? 'unknown' : 'failed', safeId(outcome.requestId, apiKey), safeId(outcome.taskId || knownTaskId, apiKey), message)
|
||
if (updated === 1) await aiModelAudit(db, userId, unknown ? 'model.verification-unknown' : 'model.verification-failed', providerId, { verificationId: row.id, modelKey: row.modelKey, stage: row.stage })
|
||
})
|
||
}
|
||
}
|
||
|
||
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, context = await resolveAiModelVerification(providerId, modelKey, input.stage, input.revision)
|
||
const created = await aiModelTransaction(async db => {
|
||
await assertAiModelAdminForWrite(userId, db)
|
||
const prior = await db.$queryRawUnsafe<AiModelVerificationRow[]>(`SELECT ${AI_MODEL_VERIFICATION_COLUMNS} FROM ai_model_verifications WHERE model_key=$1 AND stage=$2 AND idempotency_key=$3`, modelKey, input.stage, input.idempotencyKey)
|
||
if (prior[0]) return null // A repeated HTTP request never repeats a model call.
|
||
const closed = await db.$queryRawUnsafe<Array<{ id: string }>>(`SELECT id FROM ai_model_audit WHERE action='model.verification-intake-cancelled' AND details->>'modelKey'=$1 AND details->>'stage'=$2 AND details->>'idempotencyKey'=$3`, modelKey, input.stage, 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<Array<{ count: number }>>(`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 rows = await db.$queryRawUnsafe<AiModelVerificationRow[]>(`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, input.stage, context.contextHash, input.idempotencyKey, userId, JSON.stringify({ challenge: createVerificationChallenge() }))
|
||
await aiModelAudit(db, userId, 'model.verification-start', providerId, { verificationId: rows[0].id, modelKey, stage: input.stage })
|
||
return rows[0]
|
||
})
|
||
if (created) await executeVerification(created, userId, providerId, false)
|
||
return getAiModelAdminState()
|
||
}
|
||
|
||
/** Only fetch an already-created upstream task. This endpoint can never submit a new generation. */
|
||
export async function refreshAiModelVerification(userId: string, providerId: string, verificationId: string) {
|
||
if (!z.string().uuid().safeParse(verificationId).success) throw new AiModelError('测试编号无效')
|
||
await assertAiModelAdmin(userId)
|
||
const row = await aiModelTransaction(async db => {
|
||
await assertAiModelAdminForWrite(userId, db)
|
||
const rows = await db.$queryRawUnsafe<AiModelVerificationRow[]>(`SELECT ${AI_MODEL_VERIFICATION_COLUMNS} FROM ai_model_verifications WHERE id=$1 FOR UPDATE`, verificationId)
|
||
const current = rows[0]
|
||
if (!current?.taskId || !['unknown','running'].includes(current.status)) throw new AiModelError('没有可以查询的上游任务', 409, 'verification_not_resumable')
|
||
if (current.status === 'running' && current.deadlineAt.getTime() > Date.now()) throw new AiModelError('测试仍在执行，请等待', 409, 'verification_busy')
|
||
const context = await resolveAiModelVerification(providerId, current.modelKey, current.stage)
|
||
if (context.contextHash !== current.contextHash) throw new AiModelError('原连接或路由已改变，请在上游核查原任务', 409, 'verification_context_changed')
|
||
await assertExecutionSlot(db, userId, current.id)
|
||
await db.$executeRawUnsafe(`UPDATE ai_model_verifications SET status='running',actor_id=$2,deadline_at=NOW()+INTERVAL '10 minutes',finished_at=NULL WHERE id=$1`, current.id, userId)
|
||
await aiModelAudit(db, userId, 'model.verification-query', providerId, { verificationId: current.id, modelKey: current.modelKey, stage: current.stage })
|
||
return current
|
||
})
|
||
await executeVerification(row, userId, providerId, true)
|
||
return getAiModelAdminState()
|
||
}
|