diff --git a/src/modules/ai-job/ai-job-execution-engine.spec.ts b/src/modules/ai-job/ai-job-execution-engine.spec.ts index cf0aab4..1b61454 100644 --- a/src/modules/ai-job/ai-job-execution-engine.spec.ts +++ b/src/modules/ai-job/ai-job-execution-engine.spec.ts @@ -4,6 +4,7 @@ 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 { ProjectionExecutor } from './projection-executor.service'; import { PrismaService } from '../../infrastructure/database/prisma.service'; import { JobLockConflictError, JobAlreadyTerminalError } from './ai-job.errors'; @@ -29,6 +30,7 @@ describe('AiJobExecutionEngineImpl', () => { let registry: any; let stateMachine: any; let aiGateway: any; + let projectionExecutor: any; function makeJob(overrides?: any) { return { @@ -53,6 +55,7 @@ describe('AiJobExecutionEngineImpl', () => { registry = { get: jest.fn().mockReturnValue(validDef()) }; stateMachine = new AiJobStateMachine(); aiGateway = { generate: jest.fn() }; + projectionExecutor = { execute: jest.fn().mockResolvedValue([]) }; prisma = { aiJob: { @@ -75,6 +78,7 @@ describe('AiJobExecutionEngineImpl', () => { { provide: JobDefinitionRegistry, useValue: registry }, { provide: AiJobStateMachine, useValue: stateMachine }, { provide: AiGatewayService, useValue: aiGateway }, + { provide: ProjectionExecutor, useValue: projectionExecutor }, ], }).compile(); @@ -106,7 +110,7 @@ describe('AiJobExecutionEngineImpl', () => { expect(lifecycleRepo.lockJob).toHaveBeenCalledWith('job-001', 'bull-001'); // EXECUTE: AiGateway called expect(aiGateway.generate).toHaveBeenCalled(); - // PROJECT: validatedOutput written + // PROJECT: validatedOutput + outputHash written expect(prisma.aiJob.update).toHaveBeenCalledWith( expect.objectContaining({ where: { id: 'job-001' }, @@ -116,9 +120,16 @@ describe('AiJobExecutionEngineImpl', () => { }), }), ); - // COMPLETE: usage log + markSucceeded + // PROJECT: ProjectionExecutor called + expect(projectionExecutor.execute).toHaveBeenCalledWith( + undefined, // no projectorKey in validDef + expect.objectContaining({ + job: expect.objectContaining({ id: 'job-001' }), + validatedOutput: { result: 'ok' }, + }), + ); + // COMPLETE: usage log expect(prisma.aiUsageLog.create).toHaveBeenCalled(); - expect(lifecycleRepo.markSucceeded).toHaveBeenCalledWith('job-001'); }); }); diff --git a/src/modules/ai-job/ai-job-execution-engine.ts b/src/modules/ai-job/ai-job-execution-engine.ts index e2a13c0..5c3aafd 100644 --- a/src/modules/ai-job/ai-job-execution-engine.ts +++ b/src/modules/ai-job/ai-job-execution-engine.ts @@ -5,6 +5,7 @@ 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 { AiJobExecutionEngine, EngineJobContext, @@ -12,7 +13,6 @@ import { import { JobLockConflictError, JobAlreadyTerminalError, - JobNotCancellableError, } from './ai-job.errors'; // ═══════════════════════════════════════════════════════════ @@ -75,7 +75,7 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { private readonly registry: JobDefinitionRegistry, private readonly stateMachine: AiJobStateMachine, private readonly aiGateway: AiGatewayService, - // #292: ResultProjector 将在后续 Issue 注入 + private readonly projectionExecutor: ProjectionExecutor, ) {} async execute(aiJobId: string, context: EngineJobContext): Promise { @@ -187,25 +187,42 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { } // ── PROJECT ── - // #292: ResultProjector 在此调用。当前阶段仅写 validatedOutput + outputHash。 + // 调用 ProjectionExecutor（事务内：Projector + Artifact + markSucceeded） + const outputHash = this.computeHash(JSON.stringify(response.parsed)); await this.prisma.aiJob.update({ where: { id: aiJobId }, - data: { - validatedOutput: response.parsed as any, - outputHash: this.computeHash(JSON.stringify(response.parsed)), - }, + data: { validatedOutput: response.parsed as any, outputHash }, }); + const 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: response.parsed, + }, + ); + await context.updateProgress(90); // ── COMPLETE ── - // 9. Usage logging + // 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}`); }); - // 10. 标记成功 - await this.lifecycleRepo.markSucceeded(aiJobId); + 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`); diff --git a/src/modules/ai-job/ai-job.module.ts b/src/modules/ai-job/ai-job.module.ts index 5750919..f808121 100644 --- a/src/modules/ai-job/ai-job.module.ts +++ b/src/modules/ai-job/ai-job.module.ts @@ -9,6 +9,9 @@ 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'; +import { ProjectionExecutor } from './projection-executor.service'; +import { SyntheticResultProjector } from './synthetic-result-projector'; +import { RESULT_PROJECTORS } from './result-projector.interface'; @Module({ imports: [PrismaModule, AiRuntimeModule, AiModule], @@ -19,6 +22,8 @@ import { AI_JOB_EXECUTION_ENGINE } from './ai-job-execution-engine.interface'; OutboxRepository, AiJobCreationService, AiJobExecutionEngineImpl, + ProjectionExecutor, + { provide: RESULT_PROJECTORS, useFactory: (p: SyntheticResultProjector) => p, inject: [SyntheticResultProjector], multi: true } as any, { provide: AI_JOB_EXECUTION_ENGINE, useExisting: AiJobExecutionEngineImpl }, ], exports: [ diff --git a/src/modules/ai-job/projection-executor.service.ts b/src/modules/ai-job/projection-executor.service.ts new file mode 100644 index 0000000..83c53aa --- /dev/null +++ b/src/modules/ai-job/projection-executor.service.ts @@ -0,0 +1,116 @@ +import { Injectable, Logger, Inject, Optional } from '@nestjs/common'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; +import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; +import { AiJobStateMachine } from './ai-job-state-machine'; +import { ResultProjector, ProjectionContext, ArtifactReference, RESULT_PROJECTORS } from './result-projector.interface'; +import { JobAlreadyTerminalError } from './ai-job.errors'; + +/** + * M-AI-03-09: Projection Executor + * + * 在 Prisma Transaction 内原子执行： + * 1. ResultProjector.project(tx, context) → 业务实体 + * 2. AiJobArtifact 写入（project 内部已做，此处验证） + * 3. AiJobLifecycleRepository.markSucceeded(tx) → 原子标记成功 + * + * 幂等：Job 已 succeeded → 直接返回（不重复执行 Projector）。 + */ +@Injectable() +export class ProjectionExecutor { + private readonly logger = new Logger(ProjectionExecutor.name); + private readonly projectorMap = new Map(); + + constructor( + private readonly prisma: PrismaService, + private readonly lifecycleRepo: AiJobLifecycleRepository, + private readonly stateMachine: AiJobStateMachine, + @Optional() + @Inject(RESULT_PROJECTORS) + projectors?: ResultProjector[], + ) { + if (projectors) { + for (const p of projectors) { + if (this.projectorMap.has(p.key)) { + this.logger.warn(`Duplicate projector key "${p.key}" — overwriting`); + } + this.projectorMap.set(p.key, p); + } + } + this.logger.log( + `ProjectionExecutor initialized with ${this.projectorMap.size} projector(s): ` + + `[${Array.from(this.projectorMap.keys()).join(', ') || 'none'}]`, + ); + } + + /** + * 执行 Projector。 + * + * @param projectorKey JobDefinition.projectorKey + * @param context Projection 上下文 + * @returns Artifact 引用列表 + */ + async execute( + projectorKey: string | undefined, + context: ProjectionContext, + ): Promise { + if (!projectorKey) { + this.logger.log(`No projectorKey for jobType=${context.job.jobType} — skipping projection`); + return []; + } + + const projector = this.projectorMap.get(projectorKey); + if (!projector) { + // Projector 未注册，跳过（不应影响 job 成功） + this.logger.warn(`Projector "${projectorKey}" not registered — skipping projection for job=${context.job.id}`); + return []; + } + + // 幂等检查：Job 已终态则跳过 + const currentJob = await this.prisma.aiJob.findUnique({ + where: { id: context.job.id }, + select: { lifecycleStatus: true }, + }); + if (currentJob && this.stateMachine.isTerminal(currentJob.lifecycleStatus as any)) { + this.logger.warn( + `Job ${context.job.id} already terminal (${currentJob.lifecycleStatus}) — skipping projector`, + ); + // 返回已有 Artifact 引用 + const existing = await this.prisma.aiJobArtifact.findMany({ + where: { jobId: context.job.id }, + orderBy: { ordinal: 'asc' }, + }); + return existing.map((a) => ({ + artifactType: a.artifactType, + artifactId: a.artifactId, + ordinal: a.ordinal, + })); + } + + // 事务内：Projector + markSucceeded + const artifacts = await this.prisma.$transaction(async (tx) => { + const result = await projector.project(tx, context); + + // 验证 Artifact 引用的完整性 + if (result.length > 0) { + const dbArtifacts = await tx.aiJobArtifact.findMany({ + where: { jobId: context.job.id }, + }); + if (dbArtifacts.length < result.length) { + throw new Error( + `Artifact mismatch: projector returned ${result.length} but only ${dbArtifacts.length} in DB`, + ); + } + } + + // 标记成功（事务内，与 Projector 原子提交） + await this.lifecycleRepo.markSucceeded(context.job.id, tx); + + return result; + }); + + this.logger.log( + `Projection complete: job=${context.job.id} projector=${projectorKey} artifacts=${artifacts.length}`, + ); + return artifacts; + } +} diff --git a/src/modules/ai-job/result-projector.interface.ts b/src/modules/ai-job/result-projector.interface.ts new file mode 100644 index 0000000..b9eff00 --- /dev/null +++ b/src/modules/ai-job/result-projector.interface.ts @@ -0,0 +1,46 @@ +import type { Prisma } from '@prisma/client'; + +/** + * M-AI-03-09: Result Projector 接口（ADR-003 §6.1 冻结） + * + * 每个 Projector 负责将 validatedOutput 投影到具体业务实体。 + * 在 Prisma Transaction 内执行，与 AiJobArtifact + markSucceeded 共享事务。 + */ + +export interface ArtifactReference { + artifactType: string; + artifactId: string; + ordinal: number; +} + +export interface ProjectionContext { + job: { + id: string; + userId: string; + jobType: string; + targetType: string | null; + targetId: string | null; + snapshotId: string | null; + promptVersion: string | null; + outputSchemaVersion: string | null; + }; + snapshot: any; + validatedOutput: Record; +} + +export interface ResultProjector { + /** 全局唯一 key，对应 JobDefinition.projectorKey */ + readonly key: string; + + /** + * 在 Prisma Transaction 内执行投影。 + * 如果业务实体已存在（幂等），返回已有 Artifact 引用。 + */ + project( + tx: Prisma.TransactionClient, + context: ProjectionContext, + ): Promise; +} + +/** NestJS 注入 Token */ +export const RESULT_PROJECTORS = 'RESULT_PROJECTORS'; diff --git a/src/modules/ai-job/synthetic-result-projector.ts b/src/modules/ai-job/synthetic-result-projector.ts new file mode 100644 index 0000000..c374f88 --- /dev/null +++ b/src/modules/ai-job/synthetic-result-projector.ts @@ -0,0 +1,83 @@ +import { Injectable, Logger } from '@nestjs/common'; +import type { Prisma } from '@prisma/client'; +import { + ResultProjector, + ProjectionContext, + ArtifactReference, +} from './result-projector.interface'; + +/** + * M-AI-03-09: Synthetic Result Projector（仅测试环境注册） + * + * 将 synthetic_job 的 validatedOutput 写入测试用的 AiLearningAnalysis 表， + * 作为 Projector 框架的端到端验证。 + * + * 生产环境不注册此 Projector（不注册 synthetic_job Definition）。 + */ +@Injectable() +export class SyntheticResultProjector implements ResultProjector { + readonly key = 'synthetic_projector'; + private readonly logger = new Logger(SyntheticResultProjector.name); + + async project( + tx: Prisma.TransactionClient, + context: ProjectionContext, + ): Promise { + const { job, validatedOutput } = context; + + // 幂等：如果此 job 已有 artifact，直接返回 + const existing = await tx.aiJobArtifact.findMany({ + where: { jobId: job.id }, + orderBy: { ordinal: 'asc' }, + }); + if (existing.length > 0) { + this.logger.log(`Synthetic projector: returning ${existing.length} existing artifact(s) for job=${job.id}`); + return existing.map((a) => ({ + artifactType: a.artifactType, + artifactId: a.artifactId, + ordinal: a.ordinal, + })); + } + + const artifacts: ArtifactReference[] = []; + + // 写入测试分析结果 + if (validatedOutput) { + const analysis = await tx.aiLearningAnalysis.create({ + data: { + userId: job.userId, + jobId: job.id, + snapshotId: job.snapshotId, + targetType: job.targetType ?? 'synthetic', + targetId: job.targetId ?? 'test', + learningState: validatedOutput.learningState ?? null, + summary: validatedOutput.summary ?? 'Synthetic test result', + riskLevel: validatedOutput.riskLevel ?? 'low', + confidence: validatedOutput.confidence ?? 0.9, + evidence: validatedOutput.evidence ?? undefined, + promptVersion: job.promptVersion, + schemaVersion: job.outputSchemaVersion, + }, + }); + + // 写入 Artifact 引用 + await tx.aiJobArtifact.create({ + data: { + jobId: job.id, + artifactType: 'AiLearningAnalysis', + artifactId: analysis.id, + ordinal: 0, + metadata: { generatedBy: 'synthetic_projector' } as any, + }, + }); + artifacts.push({ + artifactType: 'AiLearningAnalysis', + artifactId: analysis.id, + ordinal: 0, + }); + } + + this.logger.log(`Synthetic projector: created ${artifacts.length} artifact(s) for job=${job.id}`); + return artifacts; + } +}