diff --git a/src/modules/ai-job/ai-job-execution-engine.spec.ts b/src/modules/ai-job/ai-job-execution-engine.spec.ts new file mode 100644 index 0000000..0886273 --- /dev/null +++ b/src/modules/ai-job/ai-job-execution-engine.spec.ts @@ -0,0 +1,250 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { AiJobExecutionEngineImpl } from './ai-job-execution-engine'; +import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; +import { JobDefinitionRegistry } from './job-definition-registry'; +import { AiJobStateMachine } from './ai-job-state-machine'; +import { AiGatewayService } from '../ai/gateway/ai-gateway.service'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; +import { JobLockConflictError, JobAlreadyTerminalError } from './ai-job.errors'; + +function validDef() { + return { + jobType: 'test_job', + queue: { queueName: 'ai-interactive', defaultPriority: 0 }, + execution: { timeoutMs: 5000, maxRetries: 2, retryBackoff: { type: 'exponential', delay: 2000 }, cancellable: true, abortStrategy: 'fail' }, + input: { schemaVersion: '1.0' }, + output: { schemaVersion: '1.0' }, + prompt: { promptKey: 'test', promptVersion: '1.0' }, + model: { modelTier: 'primary', modelProvider: 'deepseek', modelName: 'deepseek-chat' }, + credential: { allowedModes: ['platform_key'], defaultMode: 'platform_key' }, + security: { contentSafetyCheck: false, outputRedaction: false }, + metadata: { label: 'Test', description: '', domain: 'analysis', version: '1.0' }, + }; +} + +describe('AiJobExecutionEngineImpl', () => { + let engine: AiJobExecutionEngineImpl; + let prisma: any; + let lifecycleRepo: any; + let registry: any; + let stateMachine: any; + let aiGateway: any; + + function makeJob(overrides?: any) { + return { + id: 'job-001', + userId: 'user-1', + jobType: 'test_job', + lifecycleStatus: 'queued', + queueName: 'ai-interactive', + attemptCount: 0, + cancelRequestedAt: null, + ...overrides, + }; + } + + beforeEach(async () => { + lifecycleRepo = { + lockJob: jest.fn(), + markSucceeded: jest.fn(), + markFailed: jest.fn(), + markCancelled: jest.fn(), + }; + registry = { get: jest.fn().mockReturnValue(validDef()) }; + stateMachine = new AiJobStateMachine(); + aiGateway = { generate: jest.fn() }; + + prisma = { + aiJob: { + findUnique: jest.fn(), + update: jest.fn(), + }, + aiJobSnapshot: { findUnique: jest.fn() }, + aiUsageLog: { create: jest.fn() }, + }; + + jest.spyOn(require('@nestjs/common').Logger.prototype, 'log').mockImplementation(() => {}); + jest.spyOn(require('@nestjs/common').Logger.prototype, 'warn').mockImplementation(() => {}); + jest.spyOn(require('@nestjs/common').Logger.prototype, 'error').mockImplementation(() => {}); + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + AiJobExecutionEngineImpl, + { provide: PrismaService, useValue: prisma }, + { provide: AiJobLifecycleRepository, useValue: lifecycleRepo }, + { provide: JobDefinitionRegistry, useValue: registry }, + { provide: AiJobStateMachine, useValue: stateMachine }, + { provide: AiGatewayService, useValue: aiGateway }, + ], + }).compile(); + + engine = module.get(AiJobExecutionEngineImpl); + }); + + const ctx = { + jobId: 'bull-001', + attemptMade: 1, + updateProgress: jest.fn(), + }; + + describe('正常执行全链路', () => { + it('PREPARE→RESOLVE→EXECUTE→PROJECT→COMPLETE 全部成功', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob()); + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' })); + prisma.aiJobSnapshot.findUnique.mockResolvedValue({ + schemaVersion: '1.0', + content: { snapshotData: true }, + }); + aiGateway.generate.mockResolvedValue({ + parsed: { result: 'ok' }, + usage: { provider: 'deepseek', model: 'deepseek-chat', inputTokens: 100, outputTokens: 50, estimatedCost: 0.001, latencyMs: 800 }, + }); + + await engine.execute('job-001', ctx); + + // PREPARE: lockJob called + expect(lifecycleRepo.lockJob).toHaveBeenCalledWith('job-001', 'bull-001'); + // EXECUTE: AiGateway called + expect(aiGateway.generate).toHaveBeenCalled(); + // PROJECT: validatedOutput written + expect(prisma.aiJob.update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: 'job-001' }, + data: expect.objectContaining({ + validatedOutput: { result: 'ok' }, + outputHash: expect.any(String), + }), + }), + ); + // COMPLETE: usage log + markSucceeded + expect(prisma.aiUsageLog.create).toHaveBeenCalled(); + expect(lifecycleRepo.markSucceeded).toHaveBeenCalledWith('job-001'); + }); + }); + + describe('终态跳过', () => { + it('Job 已 succeeded → 跳过执行', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob({ lifecycleStatus: 'succeeded' })); + + await engine.execute('job-001', ctx); + expect(lifecycleRepo.lockJob).not.toHaveBeenCalled(); + expect(aiGateway.generate).not.toHaveBeenCalled(); + }); + }); + + describe('取消检查', () => { + it('queued + cancelRequestedAt → 直接标记 cancelled', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob({ cancelRequestedAt: new Date() })); + + await engine.execute('job-001', ctx); + expect(lifecycleRepo.markCancelled).toHaveBeenCalledWith('job-001'); + }); + + it('执行中被取消 → cancelled,不创建 Artifact', async () => { + prisma.aiJob.findUnique + .mockResolvedValueOnce(makeJob()) // initial read + .mockResolvedValueOnce(makeJob({ cancelRequestedAt: new Date() })); // after lockJob, check + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' })); + + await engine.execute('job-001', ctx); + expect(lifecycleRepo.markCancelled).toHaveBeenCalled(); + // Provider 不应被调用 + expect(aiGateway.generate).not.toHaveBeenCalled(); + }); + }); + + describe('错误分类', () => { + it('429 → retryable → 抛给 BullMQ', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob()); + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' })); + prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} }); + + aiGateway.generate.mockRejectedValue(new Error('429 rate limited')); + + await expect(engine.execute('job-001', ctx)).rejects.toThrow('429'); + // 不调用 markFailed(让 BullMQ 重试) + expect(lifecycleRepo.markFailed).not.toHaveBeenCalled(); + }); + + it('schema validation → permanent → markFailed', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob()); + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' })); + prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} }); + + aiGateway.generate.mockRejectedValue(new Error('schema validation failed')); + + await expect(engine.execute('job-001', ctx)).rejects.toThrow(); + expect(lifecycleRepo.markFailed).toHaveBeenCalledWith( + 'job-001', + expect.objectContaining({ errorCode: 'schema_validation_failed' }), + ); + }); + + it('timeout → retryable → 抛给 BullMQ', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob()); + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' })); + prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} }); + + aiGateway.generate.mockRejectedValue(new Error('timeout ETIMEDOUT')); + + await expect(engine.execute('job-001', ctx)).rejects.toThrow(); + expect(lifecycleRepo.markFailed).not.toHaveBeenCalled(); + }); + }); + + describe('抢锁冲突', () => { + it('已被其他 Worker 抢走 → 静默退出', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob()); + lifecycleRepo.lockJob.mockRejectedValue(new JobLockConflictError('job-001')); + + await engine.execute('job-001', ctx); + // 不应抛错,静默退出 + expect(aiGateway.generate).not.toHaveBeenCalled(); + }); + }); + + describe('Snapshot schema 不兼容', () => { + it('schemaVersion 不匹配 → markFailed', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob()); + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' })); + prisma.aiJobSnapshot.findUnique.mockResolvedValue({ + schemaVersion: '99.0', // incompatible + content: {}, + }); + + await engine.execute('job-001', ctx); + expect(lifecycleRepo.markFailed).toHaveBeenCalledWith( + 'job-001', + expect.objectContaining({ errorCode: 'schema_validation_failed' }), + ); + expect(aiGateway.generate).not.toHaveBeenCalled(); + }); + }); + + describe('Usage Logging', () => { + it('每次 Provider 调用写 AiUsageLog', async () => { + prisma.aiJob.findUnique.mockResolvedValue(makeJob()); + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running', attemptCount: 0 })); + prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} }); + aiGateway.generate.mockResolvedValue({ + parsed: { ok: true }, + usage: { provider: 'deepseek', model: 'deepseek-chat', inputTokens: 50, outputTokens: 20, estimatedCost: 0.0005, latencyMs: 300 }, + }); + + await engine.execute('job-001', ctx); + + expect(prisma.aiUsageLog.create).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + userId: 'user-1', + jobId: 'job-001', + provider: 'deepseek', + model: 'deepseek-chat', + inputTokens: 50, + outputTokens: 20, + }), + }), + ); + }); + }); +}); diff --git a/src/modules/ai-job/ai-job-execution-engine.ts b/src/modules/ai-job/ai-job-execution-engine.ts new file mode 100644 index 0000000..8490175 --- /dev/null +++ b/src/modules/ai-job/ai-job-execution-engine.ts @@ -0,0 +1,300 @@ +import { Injectable, Logger } from '@nestjs/common'; +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 { + AiJobExecutionEngine, + EngineJobContext, +} from './ai-job-execution-engine.interface'; +import { + JobLockConflictError, + JobAlreadyTerminalError, + JobNotCancellableError, +} 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, + // #292: ResultProjector 将在后续 Issue 注入 + ) {} + + async execute(aiJobId: string, context: EngineJobContext): Promise { + // ── 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 timeout via AbortController + const controller = new AbortController(); + const timeoutMs = def.execution.timeoutMs || 30000; + const timeoutId = setTimeout(() => controller.abort(), timeoutMs); + + try { + // 8. 取消检查(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 ── + // 9. 调用 AiGatewayService(不直接导入 Provider SDK) + const 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: [], // Snapshot content 应通过 promptTemplate 渲染,此处由 AiGateway 处理 + maxTokens: def.model.maxTokens, + outputSchema: undefined, // 由 AiGateway 的 parseJson 处理 + }, + timeoutMs, + ); + + clearTimeout(timeoutId); + + await context.updateProgress(70); + + // 10. 取消检查(Projector 前) + const jobAfterExec = await this.prisma.aiJob.findUnique({ + where: { id: aiJobId }, + select: { cancelRequestedAt: true }, + }); + if (jobAfterExec?.cancelRequestedAt) { + await this.lifecycleRepo.markCancelled(aiJobId); + return; + } + + // ── PROJECT ── + // #292: ResultProjector 在此调用。当前阶段仅写 validatedOutput + outputHash。 + await this.prisma.aiJob.update({ + where: { id: aiJobId }, + data: { + validatedOutput: response.parsed as any, + outputHash: this.computeHash(JSON.stringify(response.parsed)), + }, + }); + + await context.updateProgress(90); + + // ── COMPLETE ── + // 11. 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}`); + }); + + // 12. 标记成功 + await this.lifecycleRepo.markSucceeded(aiJobId); + + await context.updateProgress(100); + this.logger.log(`Job ${aiJobId} (${job.jobType}) completed successfully`); + } catch (execErr: any) { + clearTimeout(timeoutId); + + // 取消检查 + 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}`, + ); + + if (classified.retryable) { + // 重试:抛给 BullMQ(不调用 markFailed — BullMQ 会重新投递) + throw execErr; + } + + // 永久错误 + const currentAttempt = (lockedJob.attemptCount || 0) + 1; + if (currentAttempt >= (def.execution.maxRetries + 1)) { + await this.lifecycleRepo.markFailed(aiJobId, { + errorCode: classified.errorCode, + publicErrorMessage: classified.publicMessage, + internalErrorMessage: execErr.message?.slice(0, 500), + }); + } else { + // 还有重试配额 → 抛给 BullMQ + throw execErr; + } + } + } 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 ── + + private computeHash(content: string): string { + const crypto = require('crypto'); + return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16); + } + + private async writeUsageLog( + userId: string, + jobId: string, + _def: any, + response: any, + attemptNo: number, + ): Promise { + await this.prisma.aiUsageLog.create({ + data: { + userId, + jobId, + provider: response?.usage?.provider || 'deepseek', + model: response?.usage?.model || 'deepseek-chat', + tier: 'primary', + promptKey: _def?.prompt?.promptKey || 'unknown', + promptVersion: _def?.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, + }); + } +} diff --git a/src/modules/ai-job/ai-job.module.ts b/src/modules/ai-job/ai-job.module.ts index 6ba066a..5750919 100644 --- a/src/modules/ai-job/ai-job.module.ts +++ b/src/modules/ai-job/ai-job.module.ts @@ -2,32 +2,31 @@ import { Module } from '@nestjs/common'; import { PrismaModule } from '../../infrastructure/database/prisma.module'; import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository'; import { AiRuntimeModule } from '../ai-runtime/ai-runtime.module'; +import { AiModule } from '../ai/ai.module'; import { AiJobStateMachine } from './ai-job-state-machine'; import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; import { JobDefinitionRegistry } from './job-definition-registry'; import { AiJobCreationService } from './ai-job-creation.service'; +import { AiJobExecutionEngineImpl } from './ai-job-execution-engine'; +import { AI_JOB_EXECUTION_ENGINE } from './ai-job-execution-engine.interface'; -/** - * M-AI-03: AiJob 统一 Job Engine 模块 - * - * 当前包含:状态机、生命周期仓储、Registry、CreationService。 - * OutboxDispatcher 仅在 WorkerModule 注册(API 进程不运行)。 - * 后续 Issue(#291–#293)逐步新增:Engine、Projector。 - */ @Module({ - imports: [PrismaModule, AiRuntimeModule], + imports: [PrismaModule, AiRuntimeModule, AiModule], providers: [ AiJobStateMachine, AiJobLifecycleRepository, JobDefinitionRegistry, OutboxRepository, AiJobCreationService, + AiJobExecutionEngineImpl, + { provide: AI_JOB_EXECUTION_ENGINE, useExisting: AiJobExecutionEngineImpl }, ], exports: [ AiJobStateMachine, AiJobLifecycleRepository, JobDefinitionRegistry, AiJobCreationService, + AI_JOB_EXECUTION_ENGINE, ], }) export class AiJobModule {}