api-server/docs/architecture/adr-003-unified-ai-job-engine.md

ADR-003：统一 AI Job Engine、Registry、Outbox 与状态机

状态

已提出（2026-06-20）

背景

M-AI-03-01 审计（docs/architecture/m-ai-03-current-execution-audit.md）确认：

• 存在两套 Job 系统：AiJob（BullMQ + AiAnalysisWorker，2 种 jobType）和 AiRuntimeJob（REST Poll + 外部 heavy-runtime，5 种 jobType，0 行生产数据）
• M-AI-02 已在 AiAnalysisJob 物理表上完成 Schema Expand，19 个新字段就绪，Outbox/Artifact/Snapshot 表已创建
• 3 个 BullMQ 队列（notification、domain-events、audit-logs）的 Consumer 已注册但无活跃 Producer
• OutboxRepository 已完整实现但零调用方，无 Dispatcher
• queueName 不一致：AiAnalysisRepository 写入 queueName: 'ai-interactive'，实际 BullMQ 路由到 ai-analysis
• STATUS_TO_LIFECYCLE 映射缺口：仅 4 个状态，缺少 cancel_requested / cancelled

M-AI-03 的目标是在 M-AI-02 Schema Expand 之上，纯代码层实现统一 Job Engine、Registry、Outbox Dispatcher 与 Projector——无 Prisma Migration、不迁移真实业务 Job、不修改已有 Worker/队列。

---

决策

1. 状态机

1.1 完整状态定义

                ┌──────────┐
                │  queued  │  ← 唯一初始状态（AiJobLifecycleRepository.createJob）
                └────┬─────┘
                     │ lockJob (CAS)
                ┌────▼─────┐
                │  running  │
                └────┬─────┘
           ┌────────┼──────────┐
           │        │          │
    ┌──────▼──┐ ┌──▼───┐ ┌───▼──────┐
    │succeeded│ │failed│ │cancelled │  ← 终态
    └─────────┘ └──────┘ └──────────┘

状态	说明	终态	进入条件
queued	Job 已创建并入队，等待 Worker 领取	否	AiJobLifecycleRepository.createJob()
running	Worker 已锁定并开始执行	否	AiJobLifecycleRepository.lockJob() CAS 成功
succeeded	执行成功，Projector 已提交结果	是	AiJobLifecycleRepository.markSucceeded()
failed	执行失败且重试耗尽	是	AiJobLifecycleRepository.markFailed()
cancelled	用户/系统取消	是	AiJobLifecycleRepository.markCancelled()

1.2 允许迁移

queued    → running (lockJob)
queued    → cancelled (用户取消，仅 queued 状态允许)
running   → succeeded (submitResult)
running   → failed (submitFailure，retryCount >= maxRetryCount)
running   → cancelled (cancelRequestedAt 已设置，heartbeat 检测)

1.3 非法迁移

以下迁移在任何情况下都不允许：

queued    → succeeded    (必须先 lockJob → running)
queued    → failed       (必须先 lockJob → running)
succeeded → *            (终态，不可变)
failed    → *            (终态，不可变；需重试应创建新 Job)
cancelled → *            (终态，不可变)
running   → queued       (不可回退；需重试应通过 retryCount 机制)

1.4 唯一写入入口

所有 lifecycleStatus 的写入必须通过 AiJobLifecycleRepository 的以下方法：

// AiJobLifecycleRepository — 唯一写入 lifecycleStatus 的入口
export class AiJobLifecycleRepository {
  /**
   * 在给定事务中创建 Job（lifecycleStatus = 'queued'）。
   * 调用方负责在同一 tx 中写入 OutboxEvent，以保证 Job + Outbox 原子性。
   */
  createJob(tx: Prisma.TransactionClient, input: CreateJobInput): Promise<AiJob>;

  /**
   * CAS 抢锁（queued → running）。
   * 必须是独立操作（非事务内）：CAS 需要立即对其他 Worker 可见。
   */
  lockJob(jobId: string, instanceId: string): Promise<AiJob>;

  /**
   * 标记成功（→ succeeded）。
   * @param tx 可选事务客户端。由 Projector 在事务内调用以确保 Projector + markSucceeded 原子性。
   *           若未传 tx，方法内部使用独立事务（非 Projector 路径使用）。
   */
  markSucceeded(jobId: string, tx?: Prisma.TransactionClient): Promise<AiJob>;

  /**
   * 标记失败（→ failed）。
   * @param tx 可选事务客户端。若非 Projector 路径（如超时/Provider 错误），传 undefined 使用独立事务。
   */
  markFailed(jobId: string, error: JobErrorInfo, tx?: Prisma.TransactionClient): Promise<AiJob>;

  /**
   * 标记取消（→ cancelled）。
   * @param tx 可选事务客户端。取消路径通常为独立操作，不传 tx。
   */
  markCancelled(jobId: string, tx?: Prisma.TransactionClient): Promise<AiJob>;

  // 只读
  findByLifecycleStatus(status: LifecycleStatus[], opts?: { queueName?: string; limit?: number }): Promise<AiJob[]>;
  findExpiredLocks(thresholdMs: number): Promise<AiJob[]>;
}

禁止：业务代码直接调用 prisma.aiJob.update({ where: { id }, data: { lifecycleStatus: '...' } }) 或 this.repository.updateJobStatus() 写入 lifecycleStatus。

1.5 Legacy status Shadow Write

过渡期状态映射（M-AI-02-10 已有，本 ADR 扩展）：

const STATUS_TO_LIFECYCLE: Record<string, string> = {
  pending:       'queued',
  processing:    'running',
  completed:     'succeeded',
  failed:        'failed',
  cancelled:     'cancelled',        // 新增 — 修正 #284 发现的映射缺口
  cancel_requested: 'cancelled',     // 新增 — 统一映射
};

当 AiJobLifecycleRepository 变更 lifecycleStatus 时，若 legacy status 列存在，同步写入映射值（Shadow Write）。M-AI-05 完成后删除 status 列。

1.6 并发状态更新策略

lockJob 使用 MySQL 乐观锁（CAS updateMany）：

async lockJob(jobId: string, instanceId: string): Promise<AiJob> {
  const now = new Date();
  const lockUntil = new Date(now.getTime() + 60_000); // 60s TTL

  const result = await this.prisma.aiJob.updateMany({
    where: {
      id: jobId,
      lifecycleStatus: 'queued',
    },
    data: {
      lifecycleStatus: 'running',
      lockedBy: instanceId,
      lockedAt: now,
      lockUntil,
      startedAt: now,
    },
  });

  if (result.count === 0) {
    throw new JobLockConflictError(jobId);
  }

  return this.prisma.aiJob.findUniqueOrThrow({ where: { id: jobId } });
}

1.7 Heartbeat 与过期锁收割

机制	实现	频率
Heartbeat	AiJobLifecycleRepository.extendLock(jobId, instanceId) → UPDATE ... SET lockUntil = now() + 60s WHERE lifecycleStatus = 'running' AND lockedBy = instanceId	每 30s
过期锁收割	AiJobLifecycleRepository.findExpiredLocks(90_000) → 锁超时 90s（60s TTL + 30s 宽限期）未续约 → markFailed，触发 BullMQ retry	每 30s

---

2. Job Definition Registry

2.1 TypeScript 接口（冻结）

// ── Job Definition ──

interface JobDefinition {
  /** 全局唯一 jobType，与 AiJob.jobType 对应 */
  jobType: string;

  /** 元数据 */
  metadata: {
    /** 人类可读标签 */
    label: string;
    /** 描述 */
    description: string;
    /** 所属领域 */
    domain: 'analysis' | 'generation' | 'import' | 'maintenance';
    /** 版本 */
    version: string;
  };

  /** 路由策略 */
  queue: {
    /** 目标队列名 */
    queueName: 'ai-interactive' | 'ai-background';
    /** 优先级 0=最高, 100=最低 */
    defaultPriority: number;
  };

  /** 执行策略 */
  execution: {
    /** 超时（毫秒） */
    timeoutMs: number;
    /** 最大重试次数（不含首次） */
    maxRetries: number;
    /** BullMQ 重试退避 */
    retryBackoff: { type: 'exponential'; delay: number };
    /** 是否支持取消 */
    cancellable: boolean;
    /** AbortSignal 超时后行为 */
    abortStrategy: 'fail' | 'retry';
  };

  /** 输入 Schema 版本（对应 SnapshotBuilder 产出） */
  input: {
    schemaVersion: string;
  };

  /** 输出 Schema 版本（对应 AI Provider 产出 JSON Schema） */
  output: {
    schemaVersion: string;
    /** Zod schema 引用，用于输出校验 */
    validator?: string;
  };

  /** 凭据模式 */
  credential: {
    /** 支持的凭据模式 */
    allowedModes: ('platform_key' | 'user_deepseek_key')[];
    /** 默认模式 */
    defaultMode: 'platform_key' | 'user_deepseek_key';
  };

  /** Projector 引用 */
  projector?: string; // 对应 ResultProjector 注册 key

  /** 安全约束 */
  security: {
    /** 是否需要内容安全检查 */
    contentSafetyCheck: boolean;
    /** 输出是否需要脱敏后存储 */
    outputRedaction: boolean;
  };
}

2.2 Registry

// JobDefinitionRegistry — 替代未来可能出现的巨型 switch(jobType)

@Injectable()
export class JobDefinitionRegistry {
  private definitions = new Map<string, JobDefinition>();

  register(def: JobDefinition): void {
    if (this.definitions.has(def.jobType)) {
      throw new DuplicateJobDefinitionError(def.jobType);
    }
    // 启动时校验
    this.validate(def);
    this.definitions.set(def.jobType, def);
  }

  get(jobType: string): JobDefinition {
    const def = this.definitions.get(jobType);
    if (!def) throw new UnknownJobTypeError(jobType);
    return def;
  }

  getAll(): JobDefinition[] {
    return Array.from(this.definitions.values());
  }

  private validate(def: JobDefinition): void {
    if (!def.jobType.match(/^[a-z][a-z0-9_]{1,63}$/)) {
      throw new InvalidJobTypeError(def.jobType);
    }
    if (!['ai-interactive', 'ai-background'].includes(def.queue.queueName)) {
      throw new InvalidQueueError(def.queue.queueName);
    }
    if (def.execution.timeoutMs < 1000 || def.execution.timeoutMs > 600_000) {
      throw new InvalidTimeoutError(def.execution.timeoutMs);
    }
    // ... 其他校验
  }
}

2.3 注册生命周期

// 在 AiJobModule 的 onModuleInit 中注册

@Module({ /* ... */ })
export class AiJobModule implements OnModuleInit {
  constructor(private readonly registry: JobDefinitionRegistry) {}

  onModuleInit() {
    // M-AI-04 才注册 active-recall / feynman-evaluation
    // M-AI-06 才注册 quiz_generation / flashcard_generation
    // M-AI-07 才注册 learning_state_analysis / weak_point_analysis / next_action_planning

    // M-AI-03 仅注册 Synthetic Definition（测试用）
    this.registry.register(SYNTHETIC_JOB_DEFINITION);
  }
}

2.4 AiJobService — 统一 Job 创建入口与幂等性

// AiJobService — 唯一 Job 创建入口（internal-only）
@Injectable()
export class AiJobService {
  constructor(
    private readonly prisma: PrismaService,
    private readonly registry: JobDefinitionRegistry,
    private readonly lifecycleRepo: AiJobLifecycleRepository,
    private readonly outboxRepo: OutboxRepository,
    private readonly snapshotBuilder: SnapshotBuilderService,
  ) {}

  /**
   * 创建 Job + Snapshot + Outbox（一个 Prisma Transaction 内）。
   *
   * 幂等性：
   * - AiJob 表有 @@unique([userId, jobType, idempotencyKey])
   * - 如果 (userId, jobType, idempotencyKey) 三元组匹配到已有 Job，直接返回已有 Job
   * - 未提供 idempotencyKey 时每次创建新 Job
   *
   * 事务边界：
   * - BEGIN TRANSACTION
   * - createJob(tx) → INSERT AiJob (lifecycleStatus = 'queued')
   * - createInTransaction(tx) → INSERT OutboxEvent (eventType = 'ai.job.enqueue')
   * - COMMIT
   * - OutboxDispatcher 异步投递到 BullMQ
   */
  async createJob(input: CreateJobInput): Promise<AiJob>;
}

interface CreateJobInput {
  userId: string;
  jobType: string;
  targetType: string;
  targetId: string;
  /** 幂等 Key。提供时启用幂等；不提供时每次创建新 Job */
  idempotencyKey?: string;
  /** 覆盖 Definition 默认值 */
  overrides?: {
    priority?: number;
    credentialMode?: 'platform_key' | 'user_deepseek_key';
    credentialId?: string;
    promptVersion?: string;
    outputSchemaVersion?: string;
  };
}

幂等流程：

AiJobService.createJob(input)
  │
  ├─ 1. lookup JobDefinition via Registry.get(input.jobType)
  │
  ├─ 2. if input.idempotencyKey is provided:
  │     existing = prisma.aiJob.findUnique({
  │       where: { userId_jobType_idempotencyKey: {
  │         userId: input.userId,
  │         jobType: input.jobType,
  │         idempotencyKey: input.idempotencyKey
  │       }}
  │     })
  │     if existing → return existing  // 幂等返回，不创建新 Job
  │
  ├─ 3. build snapshot via SnapshotBuilder
  │
  ├─ 4. prisma.$transaction(async (tx) => {
  │       job = lifecycleRepo.createJob(tx, input)
  │       outboxRepo.createInTransaction(tx, {
  │         eventType: 'ai.job.enqueue',
  │         aggregateType: 'AiJob',
  │         aggregateId: job.id,
  │         dedupeKey: `${input.jobType}:${input.userId}:${input.idempotencyKey ?? job.id}`,
  │         payload: { jobId: job.id, jobType: input.jobType, queueName: def.queue.queueName, priority: job.priority },
  │         availableAt: new Date(),
  │       })
  │       return job
  │     })
  │
  └─ 5. return job

约束：

• 不提供通用公开创建接口 —— 外部通过具体业务 Service（如 ActiveRecallService.createJob()）调用
• CreateJobInput 的类型由 JobDefinition 的 input.schemaVersion 隐式约束
• 并发相同 (userId, jobType, idempotencyKey) → findUnique 无记录 → 两个事务同时 INSERT → 一个成功，一个 P2002 唯一冲突 → catch 后 findUnique 返回已有（赢者）Job

---

3. 队列架构

3.1 队列配置

以下值来自 Issue #288 原始规格。有意偏离需在对应 Issue 中注明原因。

队列名	用途	Worker 进程	concurrency	lockDuration	attempts	backoff
ai-interactive	新统一 Engine — 交互式 Job（用户等待结果）	WorkerModule 新增	2	30s	2	exponential 2000ms
ai-background	新统一 Engine — 后台 Job（用户不等待）	WorkerModule 新增	1	60s	3	exponential 5000ms
ai-analysis	旧系统 A — active-recall + feynman-evaluation	WorkerModule（已有）	1	30s	3	exponential 1000ms

设计理由：

• ai-interactive attempts=2（非 3）：用户等待结果，BullMQ retry 会延长感知延迟。Engine 内部已有一层重试，BullMQ 仅兜底 Worker 崩溃。
• ai-background concurrency=1（非 3）：后台 Job 以吞吐而非延迟为目标；单 Worker 串行消费避免 DeepSeek API 并发限流。若后续需要更大吞吐，通过水平扩展 Worker 实例而非提高 concurrency。
• ai-background backoff=5000ms（非默认 1000ms）：后台场景无实时性要求，更长退避减少无效重试。

约束：

• ai-interactive / ai-background 仅在 WorkerModule 中注册 Processor
• 旧 ai-analysis 队列不修改、不删除
• document-import、notification、domain-events、audit-logs、file-cleanup 五个队列不在 M-AI-03 范围

3.2 Queue Payload（冻结）

// ── BullMQ Job Payload ──

interface AiJobPayload {
  /** AiJob 数据库主键 */
  jobId: string;
  /** jobType — Worker 据此 lookup JobDefinition */
  jobType: string;
  /** 幂等 Key（与 Outbox dedupeKey 相同） */
  idempotencyKey: string;
  /** 创建时间戳 */
  createdAt: string;
}

// BullMQ jobId 格式: "{queueName}:{jobType}:{idempotencyKey}"
// 例如: "ai-interactive:active-recall:user123_20260620_001"

3.3 Worker 进程注册边界

WorkerModule 新增:
├── AiInteractiveJobWorker   @Processor('ai-interactive')
│   └── AiJobExecutionEngine.execute()
└── AiBackgroundJobWorker    @Processor('ai-background')
    └── AiJobExecutionEngine.execute()

WorkerModule 已有（不修改）:
├── AiAnalysisWorker          @Processor('ai-analysis')
├── DocumentImportWorker      @Processor('document-import')
├── NotificationWorker        @Processor('notification')
├── AuditLogProcessor         @Processor('audit-logs')
└── FileCleanupProcessor      @Processor('file-cleanup')

---

4. Outbox

4.1 事件类型

eventType	触发时机	dedupeKey 组成	aggregateType
ai.job.enqueue	事务内创建 Job 后	{jobType}:{userId}:{idempotencyKey}	AiJob

未来扩展（不在 M-AI-03 范围）：ai.job.completed、ai.job.failed

4.2 Dedupe 策略

dedupeKey 在 Outbox 表上有 UNIQUE 约束（prisma/schema.prisma:2323）。重复插入 → 数据库层拒绝，调用方视为幂等成功。

4.3 领取算法（修正 #284 发现的并发竞态）

Dispatcher.cycle():
  1. findDispatchable(N) → 读取 status='pending' AND availableAt <= now()
  2. FOR EACH event:
     a. locked = markProcessing(eventId, instanceId)  // CAS: pending → processing
     b. IF locked IS NULL → SKIP（其他 Dispatcher 已抢走）
     c. success = queue.add(event.payload)  // 投递到 BullMQ
     d. IF success → markPublished(eventId)
        ELSE → markFailed(eventId, 'DISPATCH_FAILED')

核心修复：先 CAS 抢锁（步骤 2a），锁成功后才投递（步骤 2c）。UT (#284 审计发现的）CAS 失败时跳过而非重试其他事件。修复了 queue.add() 先于 markProcessing() 导致的重复投递。

4.4 锁超时与重试

参数	值	说明
markProcessing CAS 条件	status = 'pending'	仅 pending 事件可抢锁
lockedAt TTL	60s	Dispatcher 崩溃后事件锁有效时间
releaseExpiredLocks 间隔	30s	收割过期 processing → pending
最大 attemptCount	3	超过后 status = 'dead'（新增终态，区别于 failed）
未知 eventType	记录 warning 日志，标记 published（避免阻塞队列），不投递	—

4.5 崩溃窗口处理

Dispatcher 或 BullMQ Redis 在 markProcessing() 成功但 queue.add() 尚未执行时崩溃：

• lockedAt 超时（60s）→ releaseExpiredLocks() 重置为 pending → 下一个周期重新领取
• 潜在重复投递窗口：60s
• 减轻措施：BullMQ jobId = "{queueName}:{jobType}:{idempotencyKey}" 确保相同 dedupeKey 的入队幂等（BullMQ 同 jobId 重复 add 视为 no-op）

---

5. Execution Engine

5.1 执行阶段（管线）

Job (BullMQ)
  │
  ├─ 1. PREPARE
  │   ├─ lookup JobDefinition via Registry
  │   ├─ load AiJob from DB
  │   ├─ verify lifecycleStatus = 'queued'
  │   └─ lockJob (CAS queued → running)
  │
  ├─ 2. RESOLVE
  │   ├─ resolve snapshot (SnapshotBuilder)
  │   ├─ resolve credential (credentialMode / credentialId)
  │   ├─ resolve prompt (PromptTemplateService)
  │   └─ resolve provider (ModelRouter → AiProvider)
  │
  ├─ 3. EXECUTE
  │   ├─ AbortSignal (timeout from JobDefinition.execution.timeoutMs)
  │   ├─ content safety check (if security.contentSafetyCheck)
  │   ├─ AiGatewayService.generate(request)
  │   ├─ parse & validate output (outputSchema)
  │   └─ compute outputHash
  │
  ├─ 4. PROJECT
  │   ├─ begin Prisma transaction
  │   ├─ ResultProjector.project(output) → Artifact(s)
  │   ├─ AiJobLifecycleRepository.markSucceeded()
  │   ├─ OutboxRepository.createInTransaction(tx, ai.job.completed)  // 未来
  │   └─ commit transaction
  │
  └─ 5. COMPLETE
      ├─ usage logging (ModelInvocationLog)
      ├─ publish domain event (sync EventBus)
      └─ return

5.2 错误分类

错误类别	错误码前缀	可重试	行为
PROVIDER_*	外部 AI Provider 错误（超时、5xx、限流）	✅	retry with backoff
VALIDATION_*	输出 Schema 校验失败	✅	retry（重试时调整 prompt）
CONTENT_SAFETY_*	内容安全拦截	❌	标记 failed，不重试
TIMEOUT	Job 超时	✅（如果 abortStrategy = 'retry'）	根据 Definition 决定
CREDENTIAL_*	凭据无效/过期	❌	标记 failed，用户通知
CANCELLED	用户/系统取消	❌	标记 cancelled
INTERNAL_*	内部错误（DB、Redis、代码异常）	✅（最多 1 次）	重试 1 次后 failed

5.3 Timeout 与 AbortSignal

// Execution Engine 内部
const def = this.registry.get(jobType);
const controller = new AbortController();
const timeoutId = setTimeout(() => {
  controller.abort();
  this.logger.warn(`Job ${jobId} timed out after ${def.execution.timeoutMs}ms`);
}, def.execution.timeoutMs);

try {
  const output = await this.aiGateway.generate(request, controller.signal);
  clearTimeout(timeoutId);
  // ...
} catch (err) {
  clearTimeout(timeoutId);
  if (controller.signal.aborted) {
    throw new JobTimeoutError(jobId, def.execution.timeoutMs);
  }
  throw err;
}

5.4 取消

// 用户取消流程
// 1. UserAiController.cancelJob() → AiJobLifecycleRepository.markCancelled(jobId)
//    - queued → cancelled（直接）
//    - running → 设置 cancelRequestedAt，Heartbeat 检测后 markCancelled
//    - succeeded/failed/cancelled → 409 ConflictException（终态不可取消）

// 2. Worker Heartbeat 检测
//    Execution Engine 在每次 Provider 调用前后检查：
//    if (job.cancelRequestedAt != null) throw new JobCancelledError(jobId);

5.5 Retry

层级	机制	配置来源
BullMQ	attempts + backoff	JobDefinition.execution.maxRetries + retryBackoff
Execution Engine	内部重试循环	JobDefinition.execution.maxRetries
Outbox	attemptCount < 3 → pending；>= 3 → dead	OutboxRepository

约束：BullMQ retry 与 Engine 内部 retry 不叠加。Engine 内部错误不抛出（不触发 BullMQ retry），而是在 Engine 内完成所有重试后，一次性 markFailed 或 markSucceeded。

5.6 ValidatedOutput 与 OutputHash

outputHash = SHA256(JSON.stringify(validatedOutput, Object.keys(validatedOutput).sort()))

• 存储在 AiJob.outputHash
• 用于去重（相同 input + outputHash → 幂等返回已有结果）
• 用于 Artifact 去重（相同 outputHash → 跳过 Projector）

5.7 Usage Logging

每次 AI Provider 调用后写入 AiUsageLog（M-AI-02-08 已扩展 jobId、attemptNo、credentialMode、errorCode 字段），关联 AiJob.id。

---

6. Result Projector

6.1 接口（冻结）

interface ResultProjector {
  /** 全局唯一 key，对应 JobDefinition.projector */
  readonly key: string;

  /**
   * 在 Prisma Transaction 内执行投影。
   * 如果 Artifact 已存在（幂等），返回已有引用。
   */
  project(
    tx: Prisma.TransactionClient,
    job: AiJob,
    validatedOutput: Record<string, any>,
  ): Promise<ProjectionResult>;
}

interface ProjectionResult {
  /** 创建的 Artifact 引用列表 */
  artifacts: Array<{
    artifactType: string;
    artifactId: string;
    ordinal: number;
  }>;
}

6.2 事务边界

// tx = await prisma.$transaction(async (tx) => { ... })
BEGIN TRANSACTION
  1. ResultProjector.project(tx, job, validatedOutput)
     → 写入业务表（QuizQuestion / Flashcard / AiLearningAnalysis / WeakPointCandidate / NextActionRecommendation 等）
     → 写入 AiJobArtifact (jobId, artifactType, artifactId)
  2. AiJobLifecycleRepository.markSucceeded(jobId, tx)  // 传入 tx，与 Projector 共享事务
  3. (未来) OutboxRepository.createInTransaction(tx, { eventType: 'ai.job.completed', ... })
COMMIT

• markSucceeded 接受可选 tx：传入 tx 时复用事务，不传时内部开启独立事务

• Projector 抛错 → Transaction Rollback → Job 回到 Engine retry 流程

• 全部成功 → Commit

• 任何步骤失败 → Rollback，Job 重新进入 retry 流程

• Projector 执行中触发唯一约束冲突 → 视为幂等成功，跳过

6.3 幂等保证

• AiJobArtifact 有 @@unique([jobId, artifactType, artifactId])
• 重复 Projection → DB UNIQUE 冲突 → catch → 返回已有 Artifact 引用
• {{##duplicate}} 标记在 Projector 日志中

6.4 失败回滚

Projector 抛错 → Transaction Rollback → JobFailedError → Engine retry（如果可重试）或 markFailed

---

7. API 设计

7.1 端点

方法	路径	权限	说明
GET	/api/ai/jobs/:jobId	用户（自己的 Job）/ Admin	Job 状态查询
POST	/api/ai/jobs/:jobId/cancel	用户（自己的 Job）	取消 Job
GET	/admin-api/ai/jobs	Admin（SUPER_ADMIN）	Job 列表查询（支持按 jobType/status/queueName/时间范围筛选）
POST	/admin-api/ai/jobs/:jobId/retry	Admin（SUPER_ADMIN）	手动重试失败 Job（创建新 Job，复制 input）

7.2 Job 状态响应（冻结）

interface JobStatusResponse {
  jobId: string;
  jobType: string;
  lifecycleStatus: 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled';
  queueName: string;
  priority: number;
  progress: number;         // 0–100
  createdAt: string;
  startedAt?: string;
  finishedAt?: string;
  errorCode?: string;
  publicErrorMessage?: string;   // 面向用户的错误，不含内部堆栈
  result?: {
    artifactCount: number;
    artifacts: Array<{ artifactType: string; artifactId: string }>;
  };
}

7.3 公开错误码

错误码	HTTP 状态	说明
JOB_NOT_FOUND	404	Job 不存在或不属于当前用户
JOB_NOT_CANCELLABLE	409	Job 处于终态，不可取消
JOB_ALREADY_CANCELLED	409	Job 已被取消
TOO_MANY_JOBS	429	用户并发 Job 数超限
QUOTA_EXCEEDED	429	用户日/月额度用尽
PLATFORM_UNAVAILABLE	503	平台 AI 服务不可用

7.4 Admin 公开错误

Admin 端点额外返回 internalErrorMessage（仅 SUPER_ADMIN 角色可见），包含内部堆栈信息。

---

8. 安全约束

8.1 Credential

• 用户 Key（user_deepseek_key mode）：通过 CredentialEncryptionService 加解密，AiJob 表仅存储 credentialId（FK → UserModelCredential）
• 平台 Key（platform_key mode）：从环境变量注入，不在 DB 中存储
• Runtime Resolve 端点：仅验证 Runtime Service Token 后返回解密 Key

8.2 Redis

• BullMQ 连接通过 Redis ACL 用户隔离
• Heartbeat Key：worker:heartbeat:{instanceId}，TTL 90s
• 取消信号：job:cancel:{jobId}，TTL 5min
• 无敏感数据（AiJob payload）存储在 Redis

8.3 日志

• AiJob.payload（input）不在日志中输出
• validatedOutput 不在日志中输出（仅 outputHash）
• publicErrorMessage 不含 IP、路径、堆栈
• internalErrorMessage 仅 SUPER_ADMIN 可见

8.4 Snapshot

• AiJobSnapshot.content 存储在 DB（JSON 列），遵循 redactionVersion 脱敏规则
• expiresAt 过期后由 SnapshotCleanupService 清理

8.5 Outbox Payload

• OutboxEvent.payload 仅包含 { jobId, jobType, queueName, priority }
• 不含 userId、credentialId、snapshot 内容

---

9. 模块图

AiJobModule（新增）
├── AiJobService                  // 统一 Job 创建入口（事务内：AiJob + Outbox）
├── AiJobLifecycleRepository      // 唯一 lifecycleStatus 写入入口
├── JobDefinitionRegistry         // JobType → Definition 注册表
├── AiJobExecutionEngine          // 统一执行管线（PREPARE/RESOLVE/EXECUTE/PROJECT/COMPLETE）
├── OutboxDispatcher              // 轮询 OutboxEvent → CAS 领取 → BullMQ 投递
├── ResultProjector               // 投影器接口 + 内置实现（业务 Issue 注册）
│
├── AiInteractiveJobWorker        // @Processor('ai-interactive')
└── AiBackgroundJobWorker         // @Processor('ai-background')

依赖模块（已有，不修改）：
├── AiGatewayService              // AI Provider 调用
├── ModelRouter                   // Provider 路由
├── PromptTemplateService         // Prompt 模板
├── AiCostCalculatorService       // 成本计算
├── AiUsageLogService             // 用量记录
├── ContentSafetyService          // 内容安全
├── CredentialEncryptionService   // 凭据加解密
├── SnapshotBuilderService        // Snapshot 构建
├── QueueService                  // BullMQ 入队（仅 OutboxDispatcher 使用）
└── EventBusService               // 领域事件（sync publish）

10. 时序图

User                AiJobService      AiJobLifecycleRepo   OutboxRepo         BullMQ         Worker         Engine              AiGateway

  │ POST /ai/jobs       │                   │                  │                 │              │                   │
  ├─────────────────────>│                   │                  │                 │              │                   │
  │                      │                   │                  │                 │              │                   │
  │                      │─ BEGIN TX ────────>                  │                 │              │                   │
  │                      │─ createJob(tx) ──>│                  │                 │              │                   │
  │                      │                   │─INSERT AiJob─────│                 │              │                   │
  │                      │                   │  (queued)        │                 │              │                   │
  │                      │─ createInTx(tx) ──────────────────>│                 │              │                   │
  │                      │                   │  INSERT Outbox   │                 │              │                   │
  │                      │─ COMMIT ───────────────────────────────────────────────────────────────────────────────│
  │                      │                   │                  │                 │              │                   │
  │<── { jobId, queued } │                   │                  │                 │              │                   │
  │                      │                   │                  │                 │              │                   │
  │                      │                   │     ╔═════════════════════════════╗                             │
  │                      │                   │     ║ OutboxDispatcher 异步       ║                             │
  │                      │                   │     ║                             ║                             │
  │                      │                   │     ║ findDispatchable()          ║                             │
  │                      │                   │     ║ markProcessing(CAS)         ║                             │
  │                      │                   │     ║ queue.add('ai-interactive')─>│                            │
  │                      │                   │     ║ markPublished()             ║                             │
  │                      │                   │     ╚═════════════════════════════╝                             │
  │                      │                   │                  │                 │              │                   │
  │                      │                   │                  │  ai-interactive│              │                   │
  │                      │                   │                  │<───────────────│              │                   │
  │                      │                   │                  │                 │─ process() ─>│                   │
  │                      │                   │                  │                 │              │                   │
  │                      │                   │                  │                 │              │─ 1. PREPARE       │
  │                      │                   │                  │                 │              │  lockJob(CAS)     │
  │                      │                   │                  │                 │              │                   │
  │                      │                   │                  │                 │              │─ 2. RESOLVE       │
  │                      │                   │                  │                 │              │  snapshot         │
  │                      │                   │                  │                 │              │  credential       │
  │                      │                   │                  │                 │              │                   │
  │                      │                   │                  │                 │              │─ 3. EXECUTE ──────────────────>│
  │                      │                   │                  │                 │              │                   │generate()
  │                      │                   │                  │                 │              │<──────── ─ ─ ─ ─ ─│
  │                      │                   │                  │                 │              │  validatedOutput  │
  │                      │                   │                  │                 │              │                   │
  │                      │                   │                  │                 │              │─ 4. PROJECT       │
  │                      │                   │                  │                 │              │  markSucceeded    │
  │                      │                   │                  │                 │              │  create Artifact  │
  │                      │                   │                  │                 │              │                   │
  │                      │                   │                  │                 │              │─ 5. COMPLETE      │
  │                      │                   │                  │                 │              │  usage log        │
  │                      │                   │                  │                 │              │  event            │

---

11. 测试矩阵

测试层级	范围	Issue
单元测试	AiJobLifecycleRepository — 状态迁移、非法迁移、CAS 并发	#286
单元测试	JobDefinitionRegistry — 注册、查重、校验	#287
单元测试	OutboxDispatcher — 领取算法、CAS 竞争、崩溃恢复	#290
单元测试	AiJobExecutionEngine — 正常执行、超时、取消、重试	#291
单元测试	ResultProjector — 事务提交、幂等、失败回滚	#292
集成测试	Synthetic Job Definition 端到端	#294
集成测试	AiJobService + OutboxRepository 事务原子性	#289
集成测试	Admin API 查询/取消/重试	#293

---

12. 非目标（M-AI-03 明确不包含）

1. 无 Prisma Migration — 全部在 M-AI-02 Schema 基础上实现
2. 不迁移真实业务 Job — ai-analysis 队列的 active-recall 和 feynman-evaluation 继续走旧链路
3. 不创建通用公开创建接口 — AiJobService.createJob() 为 internal-only，外部通过具体业务 Service 调用
4. 不修改 AiRuntimeJob 表 — 不碰、不读、不写
5. 不修改已有 Worker — AiAnalysisWorker、DocumentImportWorker 等不受影响
6. 不引入新的第三方依赖
7. 不替换 BullMQ — 队列基础设施不变

---

影响

• 所有 M-AI-03 后续 Issue（#286–#295）必须引用本 ADR
• ai-interactive / ai-background 队列在 QueueModule 中注册
• OutboxDispatcher 在 WorkerModule 中注册为独立 Service（非 BullMQ Processor）

回滚

每个 Issue 独立回滚：revert 对应 commit。本 ADR 无代码变更，无需回滚。

证据

• M-AI-03-01 审计：docs/architecture/m-ai-03-current-execution-audit.md
• M-AI-02 Schema 冻结：docs/architecture/adr-002-ai-job-database-expand.md
• M-AI-01 架构决策：docs/architecture/adr-001-unified-ai-architecture.md
• 当前 Prisma Schema：prisma/schema.prisma:568-677（AiJob、AiJobSnapshot、AiJobArtifact）
• Outbox Repository：src/infrastructure/outbox/outbox.repository.ts:27-139
• Queue Definitions：src/infrastructure/queue/queue-definitions.ts:94-101
• Worker Module：src/worker.module.ts:1-82