diff --git a/docs/architecture/m-ai-03-current-execution-audit.md b/docs/architecture/m-ai-03-current-execution-audit.md new file mode 100644 index 0000000..bd06671 --- /dev/null +++ b/docs/architecture/m-ai-03-current-execution-audit.md @@ -0,0 +1,474 @@ +# M-AI-03 现有执行边界审计 + +> 审计日期：2026-06-20 +> 审计范围：`api-server` 仓库所有 Job 创建、Worker 执行、Queue Producer、Provider 调用、EventBus 与 Outbox 路径 +> 审计方法：全文搜索 + 逐文件阅读 + 调用链追踪 + +--- + +## 1. 总体架构概览 + +``` +┌────────────────────────────────────────────────────────────┐ +│ api-server │ +│ │ +│ Job System A: Legacy AiJob (BullMQ) │ +│ ┌──────────────────────────────────────────────────┐ │ +│ │ AiAnalysisService → AiJob DB → BullMQ │ │ +│ │ → ai-analysis queue → AiAnalysisWorker │ │ +│ │ → AiGatewayService → DeepSeek → AiAnalysisResult│ │ +│ └──────────────────────────────────────────────────┘ │ +│ │ +│ Job System B: AiRuntimeJob (REST Poll) │ +│ ┌──────────────────────────────────────────────────┐ │ +│ │ UserAiService → AiRuntimeJob DB → Runtime polls │ │ +│ │ via REST → RuntimeInternalService │ │ +│ │ → zhixi-heavy-runtime → submitResult │ │ +│ └──────────────────────────────────────────────────┘ │ +│ │ +│ Job System C: DocumentImport (BullMQ) │ +│ ┌──────────────────────────────────────────────────┐ │ +│ │ DocumentImportService → DocumentImport DB │ │ +│ │ → BullMQ document-import queue │ │ +│ │ → DocumentImportWorker → KnowledgeImportWorkflow│ │ +│ └──────────────────────────────────────────────────┘ │ +│ │ +│ Supporting Queues (BullMQ) │ +│ ┌──────────────────────────────────────────────────┐ │ +│ │ notification, domain-events, audit-logs, │ │ +│ │ file-cleanup │ │ +│ └──────────────────────────────────────────────────┘ │ +│ │ +│ Outbox (exists, unused) │ +│ ┌──────────────────────────────────────────────────┐ │ +│ │ OutboxRepository — 表就绪，无生产者，无 Dispatcher │ │ +│ └──────────────────────────────────────────────────┘ │ +└────────────────────────────────────────────────────────────┘ +``` + +**关键发现**：存在 **两套独立的 Job 系统**（AiJob + AiRuntimeJob），使用完全不同的调度机制（BullMQ vs REST Poll）。DocumentImport 使用独立的 model 和队列。Outbox Repository 已实现但没有任何调用方。 + +--- + +## 2. Job Producer 矩阵（仅含活跃生产者） + +| # | Producer Class | Method | Trigger | DB Table | Queue | Payload | 状态写入 | 代码位置 | +|---|---------------|--------|---------|----------|-------|---------|----------|----------| +| P1 | `AiAnalysisService` | `analyze()` | User API | `AiJob` | `ai-analysis` | `{jobId, userId, type:'active-recall', questionText, knowledgeItemContent, userAnswer}` | `status:pending`, `lifecycleStatus:queued` | `ai-analysis.service.ts:19-31` | +| P2 | `AiAnalysisService` | `evaluateFeynman()` | User API | `AiJob` | `ai-analysis` | `{jobId, userId, type:'feynman-evaluation', knowledgeItemTitle, knowledgeItemContent, userExplanation}` | `status:pending`, `lifecycleStatus:queued` | `ai-analysis.service.ts:40-51` | +| P3 | `DocumentImportService` | `createImport()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, rawText, fileName}` | `status:QUEUED`(via Repository) | `document-import.service.ts:43-49` | +| P4 | `KnowledgeSourceService` | `addSource()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, sourceId, fileName}` | `status:QUEUED`(via Repository) | `knowledge-source.service.ts:43-50` | +| P5 | `KnowledgeSourceService` | `triggerParse()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, sourceId, fileName}` | `status:QUEUED`(via Repository) | `knowledge-source.service.ts:90-97` | +| P6 | `UserAiService` | `requestJob()` | User API | `AiRuntimeJob` | **无 BullMQ** | REST Poll，不适用 | `status:pending` | `user-ai.service.ts:282-299` | +| P7 | `AdminFilesController` | COS cleanup | Admin API | 无 | `file-cleanup` | `{objectKey, bucket, region}` | 无（仅入队） | `admin-files.controller.ts:41` | + +### 死代码（定义为入队方法但零调用方） + +| 方法 | 文件 | 入队目标 | 说明 | +|------|------|---------|------| +| `EventBusService.publishAsync()` | `event-bus.service.ts:26-35` | `domain-events` | **零调用方** — 全文搜索确认无任何代码调用此方法；`domain-events` 队列当前无活跃生产者 | + +### 孤儿队列（Consumer 已注册但生产者未找到） + +| 队列 | Consumer | 状态 | +|------|----------|------| +| `notification` | `NotificationWorker` (`workers/notification.worker.ts:8`) | ⚠️ 无 `queue.add('notification', ...)` 调用方；`NotificationsService.send()` 仅写 DB + sync eventBus.publish()，不入队 | +| `domain-events` | 无专用 Consumer（`EventBusService.publishAsync` 是唯一入队路径但为零调用方） | ⚠️ 完全闲置 | +| `audit-logs` | `AuditLogProcessor` (`modules/admin-audit-log/audit-log.processor.ts:7`) | ⚠️ 无 `queue.add('audit-logs', ...)` 调用方；审计日志走 `prisma.adminAuditLog.create()` 直接写 DB | + +> **注意**：`AdminEventsController` 仅**读取**队列状态（`getWaitingCount`/`getActiveCount`/`getFailed` 等）和**重试**已有失败 Job（`job.retry()`），**从不创建新 Job**。因此不列为 Producer。 + +### Producer 代码证据 + +**P1–P2** `src/modules/ai-analysis/ai-analysis.service.ts:19-31, 40-51` +```typescript +const job = await this.repository.createJob(userId, 'active-recall', input.sessionId, input.answerId); +await this.queue.add('ai-analysis', { jobId: job.id, userId, type: 'active-recall', ... }); +``` + +**P3** `src/modules/document-import/document-import.service.ts:43-49` +```typescript +const job = await this.repository.create(dto); +await this.queue.add('document-import', { importId: job.id, userId, knowledgeBaseId, rawText, fileName }); +``` + +**P4–P5** `src/modules/knowledge-source/knowledge-source.service.ts:44-50, 91-97` +```typescript +const importJob = await this.importRepo.create({ ... }); +await this.queue.add('document-import', { importId: importJob.id, userId, ... }); +``` + +**P6** `src/modules/ai-runtime/user-ai.service.ts:282-299` +```typescript +const job = await this.prisma.aiRuntimeJob.create({ data: { userId, jobType, status: 'pending', ... } }); +``` + +**P7** `src/modules/files/admin-files.controller.ts:41` +```typescript +await this.queue.add(QUEUE_FILE_CLEANUP, { objectKey: file.objectKey, bucket: file.bucket, region: 'ap-beijing' }); +``` + +### 死代码与孤儿队列证据 + +**`publishAsync()` — 零调用方** `src/common/event-bus/event-bus.service.ts:26-35` +```typescript +async publishAsync(event: BaseDomainEvent): Promise { + if (!this.queue) return ''; + const job = await this.queue.add('domain-events', { eventType, eventId, payload, occurredAt }); + return job.id || ''; +} +// 全文搜索确认：整个代码库中无任何代码调用 publishAsync() +``` + +**`NotificationsService.send()` — 仅写 DB，不入队** `src/modules/notifications/notifications.service.ts:44-49` +```typescript +async send(data: { userId: string; type: string; title: string; body: string }) { + const notification = await this.repository.create(data); // 仅写 DB + this.eventBus?.publish(new NotificationSentEvent(...)); // sync 事件，不入队 + return notification; +} +``` + +**`AdminEventsController` — 只读 + 重试，不创建新 Job** `src/modules/admin-events/admin-events.controller.ts:1-163` +- 所有方法：`getWaitingCount`, `getActiveCount`, `getFailedCount`, `getJob()`, `job.retry()` — 无 `queue.add()` + +--- + +## 3. Worker / Processor 矩阵 + +| # | Worker Class | Queue | 所在 Module | 注入的 Provider | 结果写入 | 发布事件 | +|---|-------------|-------|------------|-----------------|---------|---------| +| W1 | `AiAnalysisWorker` | `ai-analysis` | `WorkerModule` | `ActiveRecallAnalysisWorkflow`, `FeynmanEvaluationWorkflow`, `AiAnalysisRepository`, `EventBusService?`, `FocusItemsService?` | `AiAnalysisResult` | `ai.analysis.completed` | +| W2 | `DocumentImportWorker` | `document-import` | `WorkerModule` | `DocumentImportRepository`, `KnowledgeItemsRepository`, `KnowledgeImportWorkflow`, `RedisService` | `KnowledgeItem` (多个) | 无 | +| W3 | `NotificationWorker` | `notification` | `WorkerModule` | `NotificationsService` | `Notification` 记录 | 无 | +| W4 | `AuditLogProcessor` | `audit-logs` | `WorkerModule` | `PrismaService` | `AdminAuditLog` | 无 | +| W5 | `FileCleanupProcessor` | `file-cleanup` | `WorkerModule` | `CosStorageProvider` | COS 删除 | 无 | + +### Worker 代码证据 + +**W1** `src/workers/ai-analysis.worker.ts:18-106` +- 行 48: `await this.repository.updateJobStatus(jobId, 'processing')` +- 行 59–64: 调用 `feynmanWorkflow.execute()` / `recallWorkflow.execute()` +- 行 67–68: `createResult()` + `updateJobStatus(jobId, 'completed')` +- 行 72: `this.eventBus?.publish(new AIAnalysisCompleted({...}))` +- 行 86–95: 对每个 weakness 创建 `FocusItem` + +**W2** `src/workers/document-import.worker.ts:11-92` +- 行 42–45: rawText 为空时直接标记 completed +- 行 50–53: 否则 → Redis 写进度，调用 `workflow.execute()` +- 行 65–77: 逐个创建 `KnowledgeItem` +- 行 79–82: `updateStatus(importId, 'completed')` + +--- + +## 4. AiRuntimeJob 完整链路（System B — REST Poll） + +``` +User API → UserAiService.requestJob() + → SnapshotBuilder.buildSnapshot() + → prisma.aiRuntimeJob.create({ status: 'pending' }) + → prisma.questionGenerationPlan / flashcardGenerationPlan (if applicable) + → return { jobId, status: 'pending' } + +zhixi-heavy-runtime (external) → RuntimeInternalService.pollJobs() + → prisma.aiRuntimeJob.findMany({ status: 'pending', jobType: { in: [...] } }) + → filter by snapshotVersion / outputSchemaVersion capacity + → return { jobs: [...] } + +zhixi-heavy-runtime → RuntimeInternalService.lockJob() + → CAS updateMany(status:pending → status:locked, lockUntil:now+60s) + +zhixi-heavy-runtime → RuntimeInternalService.heartbeatJob() + → updateMany(status:locked → status:running, startedAt:now) + → extend lockUntil (status:running) + → check cancelRequestedAt + +zhixi-heavy-runtime → RuntimeInternalService.getSnapshot() +zhixi-heavy-runtime → RuntimeInternalService.resolveCredential() + +zhixi-heavy-runtime → RuntimeInternalService.submitResult() + → prisma.aiRuntimeResult.create() + → prisma.aiRuntimeJob.update(status:succeeded) + → persistResult() → AiLearningAnalysis / WeakPointCandidate / NextActionRecommendation / Quiz / Flashcard + → notifyJobComplete() → Notification + +zhixi-heavy-runtime → RuntimeInternalService.submitFailure() + → retry: status:pending, retryCount++ + → exhausted: status:failed, notifyJobComplete() +``` + +### 关键代码位置 + +| 步骤 | 文件 | 行号 | +|------|------|------| +| 创建 Job | `src/modules/ai-runtime/user-ai.service.ts` | 270–298 | +| Poll | `src/modules/ai-runtime/internal/runtime-internal.service.ts` | 22–81 | +| Lock (CAS) | 同上 | 85–114 | +| Heartbeat | 同上 | 118–149 | +| Get Snapshot | 同上 | 153–198 | +| Resolve Credential | 同上 | 201–217 | +| Submit Result | 同上 | 220–284 | +| Persist (by jobType) | 同上 | 286–400 | +| Submit Failure | 同上 | 572–625 | +| Notify Complete | 同上 | 629–646 | +| Invocation Logs | 同上 | 650–694 | +| Cancel (user API) | `src/modules/ai-runtime/user-ai.service.ts` | 343–364 | +| Reaper (stuck jobs) | `src/modules/ai-runtime/job-reaper.service.ts` | 24–115 | + +--- + +## 5. AiGatewayService — AI Provider 统一网关 + +### 调用关系 + +``` +AiGatewayService + ├── RAG Chat (rag-chat.service.ts:174,249) + ├── Vector Service (vector.service.ts:147) + ├── Active Recall Workflow (active-recall-analysis.workflow.ts:15) + ├── Feynman Workflow (feynman-evaluation.workflow.ts:15) + ├── Knowledge Import Workflow (knowledge-import.workflow.ts:14) + ├── Learning Trend Workflow (learning-trend.workflow.ts:28) + └── Review Card Workflow (review-card-generation.workflow.ts:15) +``` + +### 内部结构 + +`src/modules/ai/gateway/ai-gateway.service.ts:25-271` + +- **Retry**: `ModelRouter.resolve(tier)` → `preferred` provider → fallback on error → `fallback` provider +- **Safety**: `ContentSafetyService.check()` before returning output +- **Cost**: `AiCostCalculatorService.calculate()` per call +- **Usage Log**: `AiUsageLogService.log()` every attempt +- **Event**: `AIUsageRecorded` event on success, `ModelFallbackTriggered` on fallback +- **Parse**: 3-layer JSON extraction (direct → markdown fence → regex) +- **Stream**: `generateStream()` method for SSE use cases + +--- + +## 6. 队列配置矩阵 + +`src/infrastructure/queue/queue-definitions.ts:94-101` + `src/infrastructure/queue/queue.constants.ts:1-7` + +| Queue Name | concurrency | lockDuration | stalledInterval | maxStalledCount | attempts | backoff | 可环境变量覆盖 | +|------------|-------------|-------------|-----------------|-----------------|----------|---------|----------------| +| `ai-analysis` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_AI_ANALYSIS_*` | +| `document-import` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_DOCUMENT_IMPORT_*` | +| `notification` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_NOTIFICATION_*` | +| `domain-events` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_DOMAIN_EVENTS_*` | +| `audit-logs` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_AUDIT_LOGS_*` | +| `file-cleanup` | 1 | 30s | 30s | 1 | 3 | exponential, 1s | `BULL_FILE_CLEANUP_*` | + +**所有队列 concurrency 均为 1**，没有业务超时（只有 BullMQ 锁机制）。 + +--- + +## 7. EventBus 使用矩阵 + +`src/common/event-bus/event-bus.service.ts:10-37` + +| 调用方 | Event Type | 发布方式 | 触发时机 | +|--------|-----------|---------|---------| +| `AiAnalysisWorker` | `ai.analysis.completed` | sync | Job 完成后 | +| `AiGatewayService` | `ai.usage.recorded` | sync | AI 调用成功后 | +| `AiGatewayService` | `ai.fallback.triggered` | sync | Provider 降级时 | +| `GrowthService` | `StreakUpdatedEvent` | sync | 学习连续天数更新 | +| `WorkspaceService` | `ItemFavoritedEvent` | sync | 收藏操作 | +| `WorkspaceService` | `ItemUnfavoritedEvent` | sync | 取消收藏 | +| `WorkspaceService` | `TagCreatedEvent` | sync | 创建标签 | +| `WorkspaceService` | `TagDeletedEvent` | sync | 删除标签 | +| `WorkspaceService` | `SearchPerformedEvent` | sync | 执行搜索 | +| `NotificationsService` | `NotificationReadEvent` | sync | 通知已读 | +| `NotificationsService` | `NotificationSentEvent` | sync | 发送通知 | +| `NotificationsService` | `NotificationPreferenceChangedEvent` | sync | 偏好变更 | +| `QueueService` | `task.enqueued` | sync | 入队后 | + +### 关键发现 + +- **全部 sync 发布** — `publishAsync()` 方法已定义但**全文搜索确认为死代码**（零调用方）；`domain-events` 队列当前无活跃生产者 +- EventBus 当前仅作为 In-Process Event Emitter 使用；其异步域事件能力（BullMQ `domain-events` 队列）处于闲置状态 +- `QueueService` 中同步发布的 `task.enqueued` 事件无 Subscriber 消费 + +--- + +## 8. Outbox 现状评估 + +### Repository 分析 + +`src/infrastructure/outbox/outbox.repository.ts:27-139` + +| 能力 | 实现状态 | 备注 | +|------|---------|------| +| `createInTransaction(tx, input)` | ✅ 已实现 | 接受外部 `Prisma.TransactionClient` | +| `findDispatchable(limit)` | ✅ 已实现 | 简单 `findMany`，无锁 | +| `markProcessing(eventId, lockedBy)` | ✅ 已实现 | CAS via `updateMany(status:pending → processing)` | +| `markPublished(eventId)` | ✅ 已实现 | 简单 update | +| `markFailed(eventId, ...)` | ✅ 已实现 | increment attemptCount | +| `releaseExpiredLocks(thresholdMs)` | ✅ 已实现 | 重置超时 processing → pending | +| **是否有生产写入** | ❌ 无 | **所有搜索返回零调用方** | +| **是否有 Dispatcher** | ❌ 无 | 没有 Dispatcher Service/Worker | +| **是否支持并发领取** | ⚠️ 部分 | CAS 保证了唯一领取，但 `findDispatchable` 可能多 Worker 读到相同事件（CAS 后有竞态窗口） | +| **是否使用 SKIP LOCKED** | ❌ 否 | 使用应用层 CAS（updateMany + status check），非数据库层 SKIP LOCKED | +| **重复发布风险** | ⚠️ 存在 | `findDispatchable` 读取后 `markProcessing` 若 CAS 失败不重试；若 Dispatcher 崩溃后重启，`releaseExpiredLocks` 可恢复但非实时 | + +### 表结构 (Prisma) + +`prisma/schema.prisma:2323` — 拥有 `id`, `eventType`, `aggregateType`, `aggregateId`, `dedupeKey`, `payload`, `status`, `attemptCount`, `availableAt`, `lockedAt`, `lockedBy`, `publishedAt`, `lastErrorCode`, `lastErrorMessage` + +**dedupeKey 有 UNIQUE 约束** → 幂等保证存在 + +--- + +## 9. Job 状态机现状 + +### AiJob（Legacy） + +`src/modules/ai-analysis/ai-analysis.repository.ts:8-16` + +``` +pending → processing → completed / failed +``` + +- `lifecycleStatus` (M-AI-02-10 Shadow Write): `pending→queued`, `processing→running`, `completed→succeeded`, `failed→failed` +- 状态由 `AiAnalysisWorker` 直接写入 +- **无锁机制** — 依赖 BullMQ 内置的 stalled job 检测 +- **无取消支持** +- **无 retry/reaper** — 完全依赖 BullMQ 的 attempts/backoff + +### AiRuntimeJob（新） + +`src/modules/ai-runtime/job-reaper.service.ts` + `runtime-internal.service.ts` + +``` +pending → locked → running → succeeded / failed / cancelled + ↑ ↓ ↓ + └── retried ←── expired ←── (timeout) +``` + +- 通过 `lockedBy` + `lockUntil` 实现分布式锁 +- `RuntimeInternalService.lockJob()` CAS 抢锁 +- `RuntimeInternalService.heartbeatJob()` 续约 +- `JobReaperService.reap()` 每 30s 收割过期锁和超时 running +- `cancelRequestedAt` → `cancelledAt` → 取消路径 +- 重试：`retryCount < maxRetryCount` → 重置为 `pending`；否则 → `failed` + +--- + +## 10. 超时 / 重试 / 取消矩阵 + +| 系统 | 超时机制 | 重试次数 | 重试间隔 | 取消支持 | 文件位置 | +|------|---------|---------|---------|---------|---------| +| AiJob (BullMQ) | BullMQ lockDuration=30s, stalledInterval=30s, maxStalledCount=1 | 3 | exponential 1s | ❌ | `queue-definitions.ts:52-57` | +| AiRuntimeJob | timeoutSeconds=120s, Reaper 30s | maxRetryCount=3 | N/A (reaper) | ✅ cancelRequestedAt → cancelledAt | `job-reaper.service.ts`, `user-ai.service.ts:343-364` | +| DocumentImport (BullMQ) | BullMQ lockDuration=30s | 3 | exponential 1s | ❌ | `queue-definitions.ts:96` | +| AiGatewayService | DEFAULT_TIMEOUT_MS=30000, AbortController | tierConfig.maxRetries | sequential attempts | ❌ | `ai-gateway.service.ts:27,51-151` | + +--- + +## 11. 依赖边界分析 + +### 必须在 WorkerModule 的模块 + +| 模块 | 原因 | 证据 | +|------|------|------| +| `AiAnalysisWorker` | 仅 Worker 侧 BullMQ Processor，不应在 App 中注册 | `worker.module.ts:74` | +| `DocumentImportWorker` | 同上 | `worker.module.ts:75` | +| `NotificationWorker` | 同上 | `worker.module.ts:76` | +| `AuditLogProcessor` | 仅后台审计日志写入 | `worker.module.ts:77` | +| `FileCleanupProcessor` | 仅后台文件清理 | `worker.module.ts:78` | + +### 必须在 AppModule + WorkerModule 共用的模块 + +| 模块 | 原因 | 证据 | +|------|------|------| +| `AiAnalysisModule` | API 侧创建 Job → `AiAnalysisService`；Worker 侧消费 → `AiAnalysisWorker` | `app.module.ts:40`, `worker.module.ts:20` | +| `DocumentImportModule` | API 侧创建 Import → `DocumentImportService`；Worker 侧消费 → `DocumentImportWorker` | `app.module.ts:37`, `worker.module.ts:21` | +| `AiModule` | API 侧 RAG/Vector 调用 AiGateway；Worker 侧 Workflows 调用 AiGateway | `app.module.ts:10`, `worker.module.ts:8` | +| `EventBusModule` | 双向：API 侧发布事件，Worker 侧发布/消费事件 | `app.module.ts:9`, `worker.module.ts:24` | +| `NotificationsModule` | API 侧 CRUD；Worker 侧 NotificationWorker | `app.module.ts:44`, `worker.module.ts:23` | + +### 仅在 AppModule 的模块（不涉及 Worker） + +| 模块 | 原因 | +|------|------| +| `AiRuntimeModule` | REST Poll 模式，Runtime 外部消费，无 BullMQ Worker | + +### 循环依赖检查 + +| 路径 | 状态 | 说明 | +|------|------|------| +| `EventBusService` → `QueueService` | ⚠️ `forwardRef` 解决 | `event-bus.service.ts:15` — `@Inject(forwardRef(() => require(...QueueService)))` | +| `AiAnalysisModule` → `AiModule` (via Workflow) → AiGateway → `EventBusService` → `QueueService` | ✅ 单向 | 无需 forwardRef | +| API ↔ Worker 循环 | ✅ 无 | Worker 仅消费队列，不调用 API 端点 | + +--- + +## 12. 现状总结与 M-AI-03 风险点 + +### 必须保持的旧代码 + +1. **`AiAnalysisRepository`** — 当前 AI Job 创建的唯一入口（P1、P2 路径） +2. **`AiAnalysisWorker`** — 处理 active-recall 和 feynman-evaluation 的已有业务 +3. **`DocumentImportWorker`** — 已有文档导入链路 +4. **`RuntimeInternalService`** — AiRuntimeJob 的 poll/lock/submit 链路（zhixi-heavy-runtime 依赖） +5. **`JobReaperService`** — AiRuntimeJob 的过期锁收割 +6. **All 6 BullMQ queues** — 已有业务依赖 + +### 可以独立新增的模块 + +1. **新 Job Engine** — 在 M-AI-02 Schema Expand 基础上纯代码层实现 +2. **新 Outbox Dispatcher** — 独立 Service/Worker，不修改已有队列 +3. **新 Registry** — 独立 Module，不依赖已有 Processor +4. **新状态机** — 独立于现有 `AiAnalysisRepository.STATUS_TO_LIFECYCLE` +5. **新 Projector** — 独立于现有 `RuntimeInternalService.persistResult()` + +### M-AI-03 需要避免的冲突 + +- **不修改** `AiAnalysisWorker` 的业务逻辑 +- **不修改** `RuntimeInternalService` 的 REST 接口（heavy-runtime 依赖） +- **不修改** `AiGatewayService` 的 provider 调用链 +- **不修改** BullMQ 队列定义（已有队列保持） +- **新 Engine 不接管** 已有 `ai-analysis` 和 `document-import` 队列 +- **新代码仅在** `AiJob` 表（即 `AiAnalysisJob`），不碰 `AiRuntimeJob` 表 + +### 关键风险 + +| 风险 | 严重度 | 说明 | +|------|--------|------| +| 两套 Job 状态不同步 | 🔴 高 | `AiJob.status`(legacy enum) 与 `AiJob.lifecycleStatus`(M-AI-02 新字段) 存在映射但不完整 | +| Outbox 无 Dispatcher | 🟡 中 | 需新建但独立 | +| `forwardRef` EventBus↔Queue | 🟢 低 | 已有解决方案，不扩大 | +| DocumentImport 使用独立 model | 🟡 中 | 不在 M-AI-03 范围，但未来统一需注意 | +| RuntimeInternal 无事务保证 | 🔴 高 | result + job update 分两步，非原子 | +| 孤儿队列（notification/domain-events/audit-logs） | 🟡 中 | Consumer 已注册但无活跃 Producer；若从未有消息入队，`NotificationWorker` 和 `AuditLogProcessor` 实为闲置进程 + +--- + +## 附录 A：完整文件清单 + +| 文件 | 角色 | +|------|------| +| `src/modules/ai-analysis/ai-analysis.repository.ts` | AiJob 持久层 | +| `src/modules/ai-analysis/ai-analysis.service.ts` | AiJob Producer | +| `src/modules/ai-analysis/ai-analysis.module.ts` | 模块注册 | +| `src/workers/ai-analysis.worker.ts` | AiJob Consumer | +| `src/modules/document-import/document-import.service.ts` | Document Import Producer | +| `src/modules/document-import/document-import.repository.ts` | Document Import 持久层 | +| `src/workers/document-import.worker.ts` | Document Import Consumer | +| `src/modules/knowledge-source/knowledge-source.service.ts` | Knowledge Source Producer（间接） | +| `src/workers/notification.worker.ts` | Notification Consumer | +| `src/modules/admin-audit-log/audit-log.processor.ts` | Audit Log Consumer | +| `src/modules/files/file-cleanup.processor.ts` | File Cleanup Consumer | +| `src/infrastructure/queue/queue.service.ts` | QueueService — 统一入队入口 | +| `src/infrastructure/queue/queue-definitions.ts` | 队列定义注册表 | +| `src/infrastructure/queue/queue.constants.ts` | 队列名常量 | +| `src/infrastructure/queue/queue.module.ts` | 队列模块（BullMQ 注册） | +| `src/infrastructure/outbox/outbox.repository.ts` | OutboxRepository（无调用方） | +| `src/common/event-bus/event-bus.service.ts` | EventBus — 同步/异步发布 | +| `src/modules/ai/gateway/ai-gateway.service.ts` | AI 网关 — 统一 Provider 路由 | +| `src/modules/ai-runtime/internal/runtime-internal.service.ts` | Runtime Internal REST API | +| `src/modules/ai-runtime/user-ai.service.ts` | AiRuntimeJob Producer + Cancel | +| `src/modules/ai-runtime/job-reaper.service.ts` | AiRuntimeJob Reaper（过期锁收割） | +| `src/app.module.ts` | API 进程模块组装 | +| `src/worker.module.ts` | Worker 进程模块组装 | +| `prisma/schema.prisma` | Prisma Schema（AiJob, AiJobSnapshot, AiJobArtifact, AiRuntimeJob, OutboxEvent 等） |