8987598eb8
23 files (+4676/-10): - Contract: m-ai-05-feynman-migration-contract.md (737 lines) - Gate audit: m-ai-05-gate-audit.md (318 lines) - Job Definition + Snapshot Builder + Registration - Executor + BusinessValidator + ReferenceValidator - Projector (atomic transaction + 3-layer idempotency) - ExecutionRouter (FeatureFlag + idempotencyKey) - ObservabilityService (structured logging + counters) - Engine: feynman_evaluation execution branch - AiJobCreationService: feynman_evaluation safety branch - Controller/Module: Router injection - CI: path detection for m-ai-05 - E2E: 8 HTTP-layer scenarios (14 total) - Unit tests: 104 new tests (5 spec files) Co-Authored-By: Claude <noreply@anthropic.com>
491 lines
19 KiB
TypeScript
491 lines
19 KiB
TypeScript
import { Injectable, Logger } from '@nestjs/common';
|
||
import * as crypto from 'crypto';
|
||
import { AiGatewayService } from '../ai/gateway/ai-gateway.service';
|
||
import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository';
|
||
import { JobDefinitionRegistry } from './job-definition-registry';
|
||
import { AiJobStateMachine } from './ai-job-state-machine';
|
||
import { PrismaService } from '../../infrastructure/database/prisma.service';
|
||
import { ProjectionExecutor } from './projection-executor.service';
|
||
import { ActiveRecallExecutor } from './active-recall-executor';
|
||
import { ActiveRecallObservabilityService } from './active-recall-observability.service';
|
||
import { FeynmanExecutor } from './feynman-executor';
|
||
import { FeynmanObservabilityService } from './feynman-observability.service';
|
||
import { FeynmanBusinessValidator, FeynmanReferenceValidator } from './feynman-validator';
|
||
import type { ActiveRecallSnapshot } from './active-recall-snapshot-builder';
|
||
import type { FeynmanSnapshot } from './feynman-snapshot-builder';
|
||
import type { FeynmanEvaluationResult } from '../ai/prompts/schemas/feynman-evaluation.schema';
|
||
import {
|
||
AiJobExecutionEngine,
|
||
EngineJobContext,
|
||
} from './ai-job-execution-engine.interface';
|
||
import {
|
||
JobLockConflictError,
|
||
JobAlreadyTerminalError,
|
||
} from './ai-job.errors';
|
||
|
||
// ═══════════════════════════════════════════════════════════
|
||
// 错误分类（ADR-003 §5.2）
|
||
// ═══════════════════════════════════════════════════════════
|
||
|
||
interface ClassifiedError {
|
||
errorCode: string;
|
||
publicMessage: string;
|
||
retryable: boolean;
|
||
}
|
||
|
||
function classifyError(err: any): ClassifiedError {
|
||
const msg = err?.message || '';
|
||
|
||
if (err?.code === 'JOB_CANCELLED' || msg.includes('cancelled')) {
|
||
return { errorCode: 'cancelled', publicMessage: 'Job was cancelled', retryable: false };
|
||
}
|
||
if (msg.includes('429') || msg.includes('rate') || msg.includes('Rate')) {
|
||
return { errorCode: 'provider_rate_limited', publicMessage: 'AI 服务繁忙，请稍后重试', retryable: true };
|
||
}
|
||
if (msg.includes('timeout') || msg.includes('ETIMEDOUT') || msg.includes('AbortError')) {
|
||
return { errorCode: 'provider_timeout', publicMessage: 'AI 服务响应超时，请重试', retryable: true };
|
||
}
|
||
if (msg.includes('5xx') || msg.includes('503') || msg.includes('502') || msg.includes('unavailable')) {
|
||
return { errorCode: 'provider_unavailable', publicMessage: 'AI 服务暂不可用', retryable: true };
|
||
}
|
||
if (msg.includes('schema') || msg.includes('validation') || msg.includes('invalid')) {
|
||
return { errorCode: 'schema_validation_failed', publicMessage: 'AI 输出格式异常', retryable: false };
|
||
}
|
||
if (msg.includes('business') || msg.includes('Business')) {
|
||
return { errorCode: 'business_validation_failed', publicMessage: 'AI 输出不符合业务规则', retryable: false };
|
||
}
|
||
if (msg.includes('reference') || msg.includes('Reference')) {
|
||
return { errorCode: 'reference_validation_failed', publicMessage: 'AI 输出引用无效', retryable: false };
|
||
}
|
||
if (err instanceof JobLockConflictError || err instanceof JobAlreadyTerminalError) {
|
||
return { errorCode: 'internal_error', publicMessage: err.message, retryable: false };
|
||
}
|
||
|
||
return { errorCode: 'internal_error', publicMessage: '内部错误', retryable: true };
|
||
}
|
||
|
||
/**
|
||
* M-AI-03-08: AiJobExecutionEngine
|
||
*
|
||
* 统一执行管线（ADR-003 §5.1）：
|
||
* PREPARE → RESOLVE → EXECUTE → PROJECT → COMPLETE
|
||
*
|
||
* 供 AiInteractiveJobWorker / AiBackgroundJobWorker 调用。
|
||
* 替代 #288 中的 no-op 占位。
|
||
*/
|
||
@Injectable()
|
||
export class AiJobExecutionEngineImpl implements AiJobExecutionEngine {
|
||
private readonly logger = new Logger(AiJobExecutionEngineImpl.name);
|
||
|
||
constructor(
|
||
private readonly prisma: PrismaService,
|
||
private readonly lifecycleRepo: AiJobLifecycleRepository,
|
||
private readonly registry: JobDefinitionRegistry,
|
||
private readonly stateMachine: AiJobStateMachine,
|
||
private readonly aiGateway: AiGatewayService,
|
||
private readonly projectionExecutor: ProjectionExecutor,
|
||
private readonly activeRecallExecutor: ActiveRecallExecutor,
|
||
private readonly feynmanExecutor: FeynmanExecutor,
|
||
private readonly feynmanBusinessValidator: FeynmanBusinessValidator,
|
||
private readonly feynmanReferenceValidator: FeynmanReferenceValidator,
|
||
private readonly observability: ActiveRecallObservabilityService,
|
||
private readonly feynmanObs: FeynmanObservabilityService,
|
||
) {}
|
||
|
||
async execute(aiJobId: string, context: EngineJobContext): Promise<void> {
|
||
// ── PREPARE ──
|
||
// 1. 读取 AiJob
|
||
const job = await this.prisma.aiJob.findUnique({ where: { id: aiJobId } });
|
||
if (!job) {
|
||
this.logger.error(`Job ${aiJobId} not found`);
|
||
return;
|
||
}
|
||
|
||
// 2. 检查是否终态
|
||
const currentStatus = this.stateMachine.parse(job.lifecycleStatus);
|
||
if (this.stateMachine.isTerminal(currentStatus)) {
|
||
this.logger.warn(`Job ${aiJobId} already in terminal status "${currentStatus}" — skipping`);
|
||
return;
|
||
}
|
||
|
||
// 3. 检查取消请求（queued 状态直接取消，running 状态设置信号）
|
||
if (job.cancelRequestedAt) {
|
||
await this.lifecycleRepo.markCancelled(aiJobId);
|
||
return;
|
||
}
|
||
|
||
// 4. Registry 获取 Definition
|
||
const def = this.registry.get(job.jobType);
|
||
|
||
// 5. CAS 抢锁 (queued → running)，递增 attemptCount
|
||
// lockJob 内部：如果 AttemptCount > maxAttempts 或已经是终态，会抛错
|
||
let lockedJob: any;
|
||
try {
|
||
lockedJob = await this.lifecycleRepo.lockJob(aiJobId, context.jobId || 'engine');
|
||
} catch (err: any) {
|
||
// 已被其他 Worker 抢走 → 静默退出，不重试
|
||
if (err instanceof JobLockConflictError) {
|
||
this.logger.warn(`Job ${aiJobId} already locked by another worker — skipping`);
|
||
return;
|
||
}
|
||
if (err instanceof JobAlreadyTerminalError) {
|
||
this.logger.warn(`Job ${aiJobId} already terminal — skipping`);
|
||
return;
|
||
}
|
||
throw err;
|
||
}
|
||
|
||
await context.updateProgress(10);
|
||
|
||
// ── RESOLVE ──
|
||
let snapshot: any;
|
||
try {
|
||
// 6. 加载 Snapshot
|
||
const snap = await this.prisma.aiJobSnapshot.findUnique({
|
||
where: { jobId: aiJobId },
|
||
});
|
||
if (snap) {
|
||
// 验证 schemaVersion 兼容性
|
||
if (snap.schemaVersion !== def.input.schemaVersion) {
|
||
await this.lifecycleRepo.markFailed(aiJobId, {
|
||
errorCode: 'schema_validation_failed',
|
||
publicErrorMessage: 'Snapshot schema version mismatch',
|
||
internalErrorMessage: `Expected ${def.input.schemaVersion}, got ${snap.schemaVersion}`,
|
||
});
|
||
return;
|
||
}
|
||
snapshot = snap.content;
|
||
}
|
||
|
||
// 7. 取消检查（Provider 调用前）
|
||
const currentJob = await this.prisma.aiJob.findUnique({
|
||
where: { id: aiJobId },
|
||
select: { cancelRequestedAt: true },
|
||
});
|
||
if (currentJob?.cancelRequestedAt) {
|
||
await this.lifecycleRepo.markCancelled(aiJobId);
|
||
return;
|
||
}
|
||
|
||
await context.updateProgress(30);
|
||
|
||
// ── EXECUTE ──
|
||
// 按 jobType 分派执行策略：active_recall / feynman_evaluation → Executor, 其他 → AiGateway
|
||
const timeoutMs = def.execution.timeoutMs || 30000;
|
||
try {
|
||
let parsedOutput: Record<string, any>;
|
||
let response: any;
|
||
|
||
if (job.jobType === 'active_recall' && snapshot) {
|
||
// M-AI-04-05: ActiveRecall Executor 处理消息构造 + AiGateway 调用
|
||
const activeRecallSnapshot = snapshot as unknown as ActiveRecallSnapshot;
|
||
response = await this.activeRecallExecutor.execute(
|
||
activeRecallSnapshot,
|
||
timeoutMs,
|
||
);
|
||
parsedOutput = response.parsed;
|
||
this.logger.log(
|
||
`ActiveRecall Executor completed: job=${aiJobId} ` +
|
||
`score=${(parsedOutput as any)?.score}`,
|
||
);
|
||
} else if (job.jobType === 'feynman_evaluation' && snapshot) {
|
||
// M-AI-05-03: Feynman Executor 处理消息构造 + AiGateway 调用
|
||
const feynmanSnapshot = snapshot as unknown as FeynmanSnapshot;
|
||
response = await this.feynmanExecutor.execute(
|
||
feynmanSnapshot,
|
||
timeoutMs,
|
||
);
|
||
parsedOutput = response.parsed;
|
||
|
||
this.logger.log(
|
||
`Feynman Executor completed: job=${aiJobId} ` +
|
||
`score=${(parsedOutput as any)?.score}`,
|
||
);
|
||
|
||
// ── M-AI-05-03: 结构化输出验证 ──
|
||
try {
|
||
this.feynmanBusinessValidator.validate(parsedOutput as FeynmanEvaluationResult);
|
||
this.feynmanReferenceValidator.validate(parsedOutput as FeynmanEvaluationResult);
|
||
} catch (validationErr: any) {
|
||
this.logger.warn(
|
||
`Feynman validation failed for job=${aiJobId}: ${validationErr.message}`,
|
||
);
|
||
throw validationErr; // classifyError → markFailed
|
||
}
|
||
} else {
|
||
// 默认路径：直接调用 AiGateway（synthetic_job 等）
|
||
response = await this.aiGateway.generate(
|
||
{
|
||
userId: job.userId,
|
||
feature: job.jobType,
|
||
tier: def.model.modelTier as any,
|
||
promptKey: def.prompt.promptKey,
|
||
promptVersion: def.prompt.promptVersion,
|
||
messages: [],
|
||
maxTokens: def.model.maxTokens,
|
||
outputSchema: undefined,
|
||
},
|
||
timeoutMs,
|
||
);
|
||
parsedOutput = response.parsed;
|
||
}
|
||
|
||
await context.updateProgress(70);
|
||
|
||
// 取消检查（Projector 前）
|
||
const jobAfterExec = await this.prisma.aiJob.findUnique({
|
||
where: { id: aiJobId },
|
||
select: { cancelRequestedAt: true },
|
||
});
|
||
if (jobAfterExec?.cancelRequestedAt) {
|
||
await this.lifecycleRepo.markCancelled(aiJobId);
|
||
return;
|
||
}
|
||
|
||
// ── PROJECT ──
|
||
// 调用 ProjectionExecutor（事务内：Projector + Artifact + markSucceeded）
|
||
// active_recall → ActiveRecallProjector (key: 'active_recall_projector')
|
||
// synthetic_job → SyntheticResultProjector (key: 'synthetic_projector')
|
||
const outputHash = this.computeHash(JSON.stringify(parsedOutput));
|
||
await this.prisma.aiJob.update({
|
||
where: { id: aiJobId },
|
||
data: { validatedOutput: parsedOutput as any, outputHash },
|
||
});
|
||
|
||
let artifacts: any[];
|
||
try {
|
||
artifacts = await this.projectionExecutor.execute(
|
||
def.projectorKey,
|
||
{
|
||
job: {
|
||
id: job.id,
|
||
userId: job.userId,
|
||
jobType: job.jobType,
|
||
targetType: job.targetType,
|
||
targetId: job.targetId,
|
||
snapshotId: snap?.id || null,
|
||
promptVersion: def.prompt.promptVersion,
|
||
outputSchemaVersion: def.output.schemaVersion,
|
||
},
|
||
snapshot,
|
||
validatedOutput: parsedOutput,
|
||
},
|
||
);
|
||
} catch (projectorErr: any) {
|
||
// M-AI-04-GATE-FIX-02: Projector 失败独立观测
|
||
if (job.jobType === 'active_recall') {
|
||
this.observability.incrementProjectorFailed();
|
||
this.logger.error(
|
||
`[ActiveRecall] Projector failed: jobId=${aiJobId} ` +
|
||
`projectorKey=${def.projectorKey} error=${projectorErr.message}`,
|
||
);
|
||
}
|
||
// M-AI-05-06: Feynman Projector 失败观测
|
||
if (job.jobType === 'feynman_evaluation') {
|
||
this.feynmanObs.incrementProjectorFailed();
|
||
this.logger.error(
|
||
`[Feynman] Projector failed: jobId=${aiJobId} ` +
|
||
`projectorKey=${def.projectorKey} error=${projectorErr.message}`,
|
||
);
|
||
}
|
||
throw projectorErr; // 传播到外层 catch → classifyError + markFailed
|
||
}
|
||
|
||
await context.updateProgress(90);
|
||
|
||
// ── COMPLETE ──
|
||
// Usage logging
|
||
await this.writeUsageLog(job.userId, aiJobId, def, response, lockedJob.attemptCount || 0).catch((err) => {
|
||
this.logger.error(`Usage log failed for job=${aiJobId}: ${err.message}`);
|
||
});
|
||
|
||
this.logger.log(
|
||
`Job ${aiJobId} (${job.jobType}) completed: ${artifacts.length} artifact(s), hash=${outputHash}`,
|
||
);
|
||
|
||
await context.updateProgress(100);
|
||
this.logger.log(`Job ${aiJobId} (${job.jobType}) completed successfully`);
|
||
|
||
// M-AI-04-06: ActiveRecall 执行成功观测
|
||
if (job.jobType === 'active_recall') {
|
||
const durationMs = Date.now() - new Date(job.startedAt || job.queuedAt || Date.now()).getTime();
|
||
this.observability.incrementUnifiedExecuteSuccess(durationMs);
|
||
this.observability.logExecutionCompleted({
|
||
requestId: 'engine',
|
||
jobId: aiJobId,
|
||
activeRecallId: job.targetId || '',
|
||
userId: job.userId,
|
||
engineMode: 'unified',
|
||
jobType: job.jobType,
|
||
queueName: def.queue.queueName,
|
||
durationMs,
|
||
lifecycleStatus: 'succeeded',
|
||
attemptCount: lockedJob.attemptCount,
|
||
});
|
||
}
|
||
|
||
// M-AI-05-06: Feynman 执行成功观测
|
||
if (job.jobType === 'feynman_evaluation') {
|
||
const durationMs = Date.now() - new Date(job.startedAt || job.queuedAt || Date.now()).getTime();
|
||
const focusItemCount = artifacts.filter((a: any) => a.artifactType === 'FocusItem').length;
|
||
const reviewCardCount = artifacts.filter((a: any) => a.artifactType === 'ReviewCard').length;
|
||
this.feynmanObs.incrementUnifiedExecuteSuccess(durationMs);
|
||
this.feynmanObs.addFocusItemCreated(focusItemCount);
|
||
this.feynmanObs.addReviewCardCreated(reviewCardCount);
|
||
this.feynmanObs.logExecutionCompleted({
|
||
requestId: 'engine',
|
||
jobId: aiJobId,
|
||
knowledgeItemId: job.targetId || '',
|
||
userId: job.userId,
|
||
engineMode: 'unified',
|
||
jobType: job.jobType,
|
||
queueName: def.queue.queueName,
|
||
durationMs,
|
||
lifecycleStatus: 'succeeded',
|
||
attemptCount: lockedJob.attemptCount,
|
||
focusItemCount,
|
||
reviewCardCount,
|
||
});
|
||
}
|
||
} catch (execErr: any) {
|
||
// 取消检查
|
||
if (execErr?.message?.includes('cancelled')) {
|
||
await this.lifecycleRepo.markCancelled(aiJobId);
|
||
return;
|
||
}
|
||
|
||
// 错误分类与处理
|
||
const classified = classifyError(execErr);
|
||
this.logger.error(
|
||
`Job ${aiJobId} execution error: ${classified.errorCode} ` +
|
||
`retryable=${classified.retryable} msg=${execErr.message}`,
|
||
);
|
||
|
||
// M-AI-04-06: ActiveRecall 执行失败 + 重试观测
|
||
if (job.jobType === 'active_recall') {
|
||
if (classified.retryable) {
|
||
this.observability.incrementUnifiedRetry();
|
||
} else {
|
||
this.observability.incrementUnifiedExecuteFailed();
|
||
}
|
||
this.observability.logExecutionFailed(
|
||
{
|
||
requestId: 'engine',
|
||
jobId: aiJobId,
|
||
activeRecallId: job.targetId || '',
|
||
userId: job.userId,
|
||
engineMode: 'unified',
|
||
jobType: job.jobType,
|
||
queueName: def.queue.queueName,
|
||
errorCode: classified.errorCode,
|
||
},
|
||
execErr.message,
|
||
);
|
||
}
|
||
|
||
// M-AI-05-06: Feynman 执行失败 + 重试观测
|
||
if (job.jobType === 'feynman_evaluation') {
|
||
if (classified.retryable) {
|
||
this.feynmanObs.incrementUnifiedRetry();
|
||
} else {
|
||
this.feynmanObs.incrementUnifiedExecuteFailed();
|
||
}
|
||
this.feynmanObs.logExecutionFailed(
|
||
{
|
||
requestId: 'engine',
|
||
jobId: aiJobId,
|
||
knowledgeItemId: job.targetId || '',
|
||
userId: job.userId,
|
||
engineMode: 'unified',
|
||
jobType: job.jobType,
|
||
queueName: def.queue.queueName,
|
||
errorCode: classified.errorCode,
|
||
},
|
||
execErr.message,
|
||
);
|
||
}
|
||
|
||
if (classified.retryable) {
|
||
// 重试：先解锁回 queued（BullMQ retry → lockJob 可再次抢锁），然后抛给 BullMQ
|
||
await this.unlockForRetry(aiJobId);
|
||
throw execErr;
|
||
}
|
||
|
||
// 永久错误：无论 attemptCount 多少，直接 markFailed（不重试）
|
||
await this.lifecycleRepo.markFailed(aiJobId, {
|
||
errorCode: classified.errorCode,
|
||
publicErrorMessage: classified.publicMessage,
|
||
internalErrorMessage: execErr.message?.slice(0, 500),
|
||
});
|
||
// markFailed 后不抛错 → BullMQ 认为 job 完成（不再重试）
|
||
return;
|
||
}
|
||
} catch (outerErr: any) {
|
||
// 捕获 RESOLVE 阶段的错误
|
||
const classified = classifyError(outerErr);
|
||
this.logger.error(
|
||
`Job ${aiJobId} resolve error: ${classified.errorCode} — ${outerErr.message}`,
|
||
);
|
||
if (!classified.retryable) {
|
||
await this.lifecycleRepo.markFailed(aiJobId, {
|
||
errorCode: classified.errorCode,
|
||
publicErrorMessage: classified.publicMessage,
|
||
internalErrorMessage: outerErr.message?.slice(0, 500),
|
||
}).catch(() => {});
|
||
}
|
||
throw outerErr;
|
||
}
|
||
}
|
||
|
||
// ── helpers ──
|
||
|
||
/**
|
||
* 解锁 Job 回 queued 状态，供 BullMQ 重试。
|
||
* 保持 attemptCount（已在 lockJob 中原子递增）。
|
||
*/
|
||
private async unlockForRetry(jobId: string): Promise<void> {
|
||
// 仅持有锁的 Worker 调用 — 无并发竞争，update 即可。
|
||
await this.prisma.aiJob.update({
|
||
where: { id: jobId },
|
||
data: {
|
||
lifecycleStatus: 'queued',
|
||
status: 'pending',
|
||
lockedBy: null,
|
||
lockedAt: null,
|
||
lockUntil: null,
|
||
},
|
||
});
|
||
}
|
||
|
||
private computeHash(content: string): string {
|
||
return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16);
|
||
}
|
||
|
||
private async writeUsageLog(
|
||
userId: string,
|
||
jobId: string,
|
||
definition: any,
|
||
response: any,
|
||
attemptNo: number,
|
||
): Promise<void> {
|
||
await this.prisma.aiUsageLog.create({
|
||
data: {
|
||
userId,
|
||
jobId,
|
||
provider: response?.usage?.provider || 'deepseek',
|
||
model: response?.usage?.model || 'deepseek-chat',
|
||
tier: 'primary',
|
||
promptKey: definition?.prompt?.promptKey || 'unknown',
|
||
promptVersion: definition?.prompt?.promptVersion || 'unknown',
|
||
inputTokens: response?.usage?.inputTokens || 0,
|
||
outputTokens: response?.usage?.outputTokens || 0,
|
||
estimatedCost: response?.usage?.estimatedCost || 0,
|
||
latencyMs: response?.usage?.latencyMs || 0,
|
||
success: true,
|
||
attemptNo,
|
||
credentialMode: 'platform_key',
|
||
} as any,
|
||
});
|
||
}
|
||
}
|