feat: M-AI-03 Result Projector 与 Artifact 原子提交框架

- result-projector.interface.ts: ResultProjector 接口冻结（ADR-003 §6.1）
  project(tx, context) → ArtifactReference[]
- synthetic-result-projector.ts: 测试用 Synthetic Projector
  幂等：已有 Artifact 时直接返回
- projection-executor.service.ts: ProjectionExecutor
  事务内：projector.project(tx) + markSucceeded(tx) 原子提交
  幂等检查：Job 已终态 → 跳过
  NestJS multi-provider 注入 RESULT_PROJECTORS
- ai-job-execution-engine.ts: PROJECT 阶段改为调用 ProjectionExecutor
- ai-job.module.ts: 注册 ProjectionExecutor + SyntheticResultProjector

Co-Authored-By: Claude <noreply@anthropic.com>

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');
     });
   });
 

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<void> {
@@ -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`);

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: [

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<string, ResultProjector>();
+
+  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<ArtifactReference[]> {
+    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;
+  }
+}

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<string, any>;
+}
+
+export interface ResultProjector {
+  /** 全局唯一 key，对应 JobDefinition.projectorKey */
+  readonly key: string;
+
+  /**
+   * 在 Prisma Transaction 内执行投影。
+   * 如果业务实体已存在（幂等），返回已有 Artifact 引用。
+   */
+  project(
+    tx: Prisma.TransactionClient,
+    context: ProjectionContext,
+  ): Promise<ArtifactReference[]>;
+}
+
+/** NestJS 注入 Token */
+export const RESULT_PROJECTORS = 'RESULT_PROJECTORS';

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<ArtifactReference[]> {
+    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;
+  }
+}