222 lines
22 KiB
TypeScript
222 lines
22 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, 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 { aiModelExecutionOccupiesSlot } from './ai-model-execution-slot'
|
||
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<string, unknown> {
|
||
if (!value || typeof value !== 'object' || Array.isArray(value)) return {}
|
||
const summary = value as Record<string, unknown>, evidence: Record<string, unknown> = {}
|
||
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<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 DISTINCT COALESCE(v.evidence->>'executionGroupId',v.id) AS execution_id,v.actor_id FROM ai_model_verifications v WHERE ($2::text IS NULL OR COALESCE(v.evidence->>'executionGroupId',v.id)<>$2) AND ${aiModelExecutionOccupiesSlot('v')}
|
||
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<ReturnType<typeof resolveAiModelVerification>>
|
||
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<void> {
|
||
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<Array<{ id: string }>>("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<Record<AiModelVerificationRow['stage'], AiModelTextCheck>> = 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, ...(retainedRequestId ? { retainedResponse: true } : {}), 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 && (typeof outcome.outcomeUnknown === 'boolean' ? (sent || Boolean(knownTaskId) || Boolean(retainedRequestId)) && outcome.outcomeUnknown : Boolean(retainedRequestId) || 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<AiModelVerificationRow[]>(`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<Array<{ id: string }>>(`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<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 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<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, 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 the original response or existing task. 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<AiModelVerificationRow[]>(`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 && !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-chat', 'anthropic-messages', 'openai-image-generate', 'openai-image-edit'].includes(contexts[0].protocol.id)) throw new AiModelError('该协议尚未适配原响应取回', 400, 'protocol_unsupported')
|
||
retainedRequestId = requestId!
|
||
}
|
||
// Reading an already-submitted request does not reserve another generation slot.
|
||
await detailTransactionLock(db, 'detail-generator-execution')
|
||
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()
|
||
}
|