api-server/docs/architecture/m-ai-03-current-execution-audit.md

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

⚠️ queueName 与实际队列不一致：AiAnalysisRepository.createJob() 写入 queueName: 'ai-interactive'（ai-analysis.repository.ts:28），但 AiAnalysisService 实际将 BullMQ Job 发送到 ai-analysis 队列（ai-analysis.service.ts:21,42）。数据库记录的 queueName 字段与真实 BullMQ 队列路由存在偏差。M-AI-03 构建统一 Engine 将依赖 queueName 字段进行路由决策，此为关键风险。 | 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 |

注：QueueService.add()（queue.service.ts:41-58）是上述所有 BullMQ Producer 的公共抽象层，不作为独立 Producer 列出。其副作用：写入 TaskLog 记录（status:enqueued）和同步发布 task.enqueued 事件。

死代码（定义为入队方法但零调用方）

方法	文件	入队目标	说明
EventBusService.publishAsync()	event-bus.service.ts:26-35	domain-events	零调用方 — 全文搜索确认无任何代码调用此方法；domain-events 队列当前无活跃生产者

孤儿队列（Consumer 已注册但生产者未找到）

队列	Consumer	WorkerModule 注册	生产者状态
notification	NotificationWorker (workers/notification.worker.ts:8)	worker.module.ts:76	❌ 无生产者 — 全文搜索确认零 queue.add('notification', ...)；NotificationsService.send() 仅写 DB + sync eventBus，不入队
domain-events	无专用 Consumer（只有 EventBusService.publishAsync 可入队）	N/A	❌ 无生产者 — publishAsync() 为零调用方死代码；队列完全闲置
audit-logs	AuditLogProcessor (modules/admin-audit-log/audit-log.processor.ts:7)	worker.module.ts:77	❌ 无生产者 — 全文搜索确认零 queue.add('audit-logs', ...)；所有审计日志通过 prisma.adminAuditLog.create() 直接写 DB

含义：NotificationWorker、AuditLogProcessor 两个 Worker 进程注册在 WorkerModule 中，每 30s 从各自队列 poll，但这两个队列从未有消息进入——实为永久空转的闲置进程。

注意：AdminEventsController 仅读取队列状态（getWaitingCount/getActiveCount/getFailed 等）和重试已有失败 Job（job.retry()），从不创建新 Job。因此不列为 Producer。

Producer 代码证据

P1–P2 src/modules/ai-analysis/ai-analysis.service.ts:19-31, 40-51

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

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

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

const job = await this.prisma.aiRuntimeJob.create({ data: { userId, jobType, status: 'pending', ... } });

P7 src/modules/files/admin-files.controller.ts:41

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

async publishAsync(event: BaseDomainEvent): Promise<string> {
  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

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

W4 src/modules/admin-audit-log/audit-log.processor.ts:8-27

async process(job: Job) {
  await this.prisma.adminAuditLog.create({
    data: {
      adminUserId: job.data.adminUserId,
      action: job.data.action,
      resourceType: job.data.resourceType,
      resourceId: job.data.resourceId,
      beforeJson: job.data.beforeJson,
      afterJson: job.data.afterJson,
      ip: job.data.ip,
      userAgent: job.data.userAgent,
      riskLevel: job.data.riskLevel,
      reason: job.data.reason,
    },
  });
}

• 行为：直接从 BullMQ Job payload 写入 AdminAuditLog 表
• 当前因队列无生产者，此 Processor 从未被触发

未使用的依赖注入

文件	注入	状态	建议
content-safety.service.ts:17	QueueService	❌ 未使用 — 构造函数注入但全文无 this.queue.add() 或 this.queue.getJob() 调用	若模块未来不计划使用队列，可清理以减少依赖
content-safety.module.ts:6,12	QueueService 导入 + 注册为 provider	仅为满足 ContentSafetyService 构造函数	同上

---

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 消费

补充：不在 EventBus 中的直接通知路径

RuntimeInternalService.notifyJobComplete()（runtime-internal.service.ts:629-646）绕过 EventBus 直接写入 prisma.notification.create()——在提交结果或 Job 彻底失败时触发。此路径不在上述事件矩阵中，是独立的 Job 完成通知机制：

private async notifyJobComplete(userId, jobId, jobType, status) {
  await this.prisma.notification.create({
    data: {
      userId,
      type: status === 'succeeded' ? 'ai_job_succeeded' : 'ai_job_failed',
      title: ..., content: ..., data: { jobId, jobType, status },
    },
  });
}

调用时机：submitResult() 成功后（runtime-internal.service.ts:278）和 submitFailure() 重试耗尽后（runtime-internal.service.ts:618）。

---

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
是否支持并发领取	❌ 否	findDispatchable 仅做无锁 findMany；markProcessing 使用应用层 CAS（updateMany WHERE status='pending'）后补救。若 CAS 失败，调用方静默丢弃该事件，不重试也不回退
是否使用 SKIP LOCKED	❌ 否	使用应用层 CAS（updateMany + status check），非数据库层 SKIP LOCKED
并发重复发布风险	⚠️ 存在	具体场景：两个 Dispatcher 实例同时调用 findDispatchable(50) → 读到同一批 [E1, E2, ...] → 各自投递到 BullMQ → 同一事件可能被发布两次。markProcessing 的 CAS 仅防止数据库状态被双重更新，但不阻止网络层重复投递（BullMQ queue.add() 一旦调用就无法回滚）

重复发布场景分析（Issue #290 相关）：

Dispatcher-A                            Dispatcher-B
  │                                        │
  ├─ findDispatchable() → [E1,E2]          │
  │                                        ├─ findDispatchable() → [E1,E2]
  ├─ queue.add(E1)  ← 第一次投递             │
  │                                        ├─ queue.add(E1)  ← 重复投递！
  ├─ markProcessing(E1) ← CAS 成功          │
  │                                        ├─ markProcessing(E1) ← CAS 失败，静默丢弃
  ├─ queue.add(E2)                          │
  │                                        ├─ queue.add(E2)  ← 重复投递！
  ├─ markProcessing(E2) ← CAS 成功          │
  │                                        ├─ markProcessing(E2) ← CAS 失败，静默丢弃

根因：queue.add() 与 markProcessing() 不是原子操作。BullMQ 投递成功后若 CAS 失败，没有补偿路径。修复方向：先抢锁（markProcessing CAS），锁定成功后再投递；或使用 DB SKIP LOCKED 在 findDispatchable 阶段就排他抢占。

表结构 (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

⚠️ STATUS_TO_LIFECYCLE 映射缺口

ai-analysis.repository.ts:10-15 仅映射 4 个状态：

private static readonly STATUS_TO_LIFECYCLE: Record<string, string> = {
  pending:    'queued',
  processing: 'running',
  completed:  'succeeded',
  failed:     'failed',
};

缺失：cancel_requested 和 cancelled（Issue #286 验收标准要求这两个状态也在映射范围内）。当前 AiJob 系统完全不支持取消操作，但如果 #286 引入统一状态机后 cancel_requested / cancelled 成为通用状态，此映射表将产生缺口。

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 表

关键风险

风险	严重度	说明
queueName 与实际路由不一致	🔴 高	DB 写入 queueName: 'ai-interactive'（ai-analysis.repository.ts:28），实际入队 ai-analysis（ai-analysis.service.ts:21）；M-AI-03 依赖 queueName 字段进行路由
STATUS_TO_LIFECYCLE 映射缺口	🟡 中	仅映射 4 个状态，缺 cancel_requested / cancelled；#286 引入统一状态机后将暴露此缺口
两套 Job 状态不同步	🔴 高	AiJob.status(legacy enum) 与 AiJob.lifecycleStatus(M-AI-02 新字段) 存在映射但不完整
Outbox 无 Dispatcher 且并发发布不安全	🔴 高	queue.add() 与 markProcessing() 非原子，两 Dispatcher 并发时可重复投递（详见第 8 节）
forwardRef EventBus↔Queue	🟢 低	已有解决方案，不扩大
DocumentImport 使用独立 model	🟡 中	不在 M-AI-03 范围，但未来统一需注意
RuntimeInternal 无事务保证	🔴 高	result + job update 分两步，非原子
孤儿队列（notification/domain-events/audit-logs）	🟡 中	Consumer 已注册但无活跃 Producer；NotificationWorker 和 AuditLogProcessor 实为永久空转
ContentSafetyService 注入未使用的 QueueService	🟢 低	构造函数注入但全文无 this.queue.* 调用；content-safety.module.ts 仅为满足此注入而注册 provider，可清理以减少依赖

---

附录 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 等）