diff --git a/docs/architecture/adr-003-unified-ai-job-engine.md b/docs/architecture/adr-003-unified-ai-job-engine.md new file mode 100644 index 0000000..dcb9755 --- /dev/null +++ b/docs/architecture/adr-003-unified-ai-job-engine.md @@ -0,0 +1,782 @@ +# 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` 的以下方法： + +```typescript +// AiJobLifecycleRepository — 唯一写入 lifeCycleStatus 的入口 +export class AiJobLifecycleRepository { + createJob(tx: Prisma.TransactionClient, input: CreateJobInput): Promise; + lockJob(jobId: string, instanceId: string): Promise; // CAS: queued → running + markSucceeded(jobId: string): Promise; + markFailed(jobId: string, error: JobErrorInfo): Promise; + markCancelled(jobId: string): Promise; + + // 只读 + findByLifecycleStatus(status: LifecycleStatus[], opts?: { queueName?: string; limit?: number }): Promise; + findExpiredLocks(thresholdMs: number): Promise; +} +``` + +**禁止**：业务代码直接调用 `prisma.aiJob.update({ where: { id }, data: { lifecycleStatus: '...' } })` 或 `this.repository.updateJobStatus()` 写入 `lifecycleStatus`。 + +#### 1.5 Legacy `status` Shadow Write + +过渡期状态映射（M-AI-02-10 已有，本 ADR 扩展）： + +```typescript +const STATUS_TO_LIFECYCLE: Record = { + 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）： + +```typescript +async lockJob(jobId: string, instanceId: string): Promise { + 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 接口（冻结） + +```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 + +```typescript +// JobDefinitionRegistry — 替代未来可能出现的巨型 switch(jobType) + +@Injectable() +export class JobDefinitionRegistry { + private definitions = new Map(); + + 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 注册生命周期 + +```typescript +// 在 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); + } +} +``` + +--- + +### 3. 队列架构 + +#### 3.1 队列配置 + +| 队列名 | 用途 | Worker 进程 | 默认 concurrency | lockDuration | attempts | +|--------|------|------------|-----------------|-------------|----------| +| `ai-interactive` | 新统一 Engine — 交互式 Job（用户等待结果） | `WorkerModule` 新增 | 2 | 30s | 3 | +| `ai-background` | 新统一 Engine — 后台 Job（用户不等待） | `WorkerModule` 新增 | 3 | 60s | 3 | +| `ai-analysis` | 旧系统 A — active-recall + feynman-evaluation | `WorkerModule`（已有） | 1 | 30s | 3 | + +**约束**： + +- `ai-interactive` / `ai-background` 仅在 `WorkerModule` 中注册 Processor +- 旧 `ai-analysis` 队列不修改、不删除 +- `document-import`、`notification`、`domain-events`、`audit-logs`、`file-cleanup` 五个队列不在 M-AI-03 范围 + +#### 3.2 Queue Payload（冻结） + +```typescript +// ── 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 + +```typescript +// 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 取消 + +```typescript +// 用户取消流程 +// 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 接口（冻结） + +```typescript +interface ResultProjector { + /** 全局唯一 key，对应 JobDefinition.projector */ + readonly key: string; + + /** + * 在 Prisma Transaction 内执行投影。 + * 如果 Artifact 已存在（幂等），返回已有引用。 + */ + project( + tx: Prisma.TransactionClient, + job: AiJob, + validatedOutput: Record, + ): Promise; +} + +interface ProjectionResult { + /** 创建的 Artifact 引用列表 */ + artifacts: Array<{ + artifactType: string; + artifactId: string; + ordinal: number; + }>; +} +``` + +#### 6.2 事务边界 + +``` +BEGIN TRANSACTION + 1. ResultProjector.project(tx, job, validatedOutput) + → 写入业务表（QuizQuestion / Flashcard / AiLearningAnalysis / WeakPointCandidate / NextActionRecommendation 等） + → 写入 AiJobArtifact (jobId, artifactType, artifactId) + 2. AiJobLifecycleRepository.markSucceeded(tx, jobId) + 3. (未来) OutboxRepository.createInTransaction(tx, { eventType: 'ai.job.completed', ... }) +COMMIT +``` + +- **全部成功** → 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 状态响应（冻结） + +```typescript +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`