wangdl/api-server

Fork 0

wangdl 64659825ae

Deploy API Server / build-and-unit (push) Successful in 31s

Details

Deploy API Server / current-integration (push) Successful in 29s

Details

Deploy API Server / backward-compat (push) Successful in 1s

Details

Deploy API Server / deploy (push) Successful in 1m0s

Details

docs: 修正审计文档 — 6 项 Non-blocking 问题

问题 5：新增 queueName 与实际队列不一致（ai-interactive vs ai-analysis）及风险标注
问题 6：移除 QueueService.add() 独立 Producer 身份，降级为基础设施注释
问题 7：细化孤儿队列描述 — 每个队列的生产者状态、WorkerModule 注册、闲置含义
问题 8：新增 STATUS_TO_LIFECYCLE 映射缺口分析（缺少 cancel_requested / cancelled）
问题 9：补充 RuntimeInternalService.notifyJobComplete() 的直接通知路径
问题 10：新增 Outbox 并发重复投递场景的时序分析

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

2026-06-20 16:59:34 +08:00

30 KiB

Raw Blame History

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

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 风险点

必须保持的旧代码

AiAnalysisRepository — 当前 AI Job 创建的唯一入口（P1、P2 路径）
AiAnalysisWorker — 处理 active-recall 和 feynman-evaluation 的已有业务
DocumentImportWorker — 已有文档导入链路
RuntimeInternalService — AiRuntimeJob 的 poll/lock/submit 链路（zhixi-heavy-runtime 依赖）
JobReaperService — AiRuntimeJob 的过期锁收割
All 6 BullMQ queues — 已有业务依赖

可以独立新增的模块

新 Job Engine — 在 M-AI-02 Schema Expand 基础上纯代码层实现
新 Outbox Dispatcher — 独立 Service/Worker，不修改已有队列
新 Registry — 独立 Module，不依赖已有 Processor
新状态机 — 独立于现有 AiAnalysisRepository.STATUS_TO_LIFECYCLE
新 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` 实为永久空转

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

30 KiB Raw Blame History Unescape Escape

M-AI-03 现有执行边界审计

1. 总体架构概览

2. Job Producer 矩阵（仅含活跃生产者）

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

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

Producer 代码证据

死代码与孤儿队列证据

3. Worker / Processor 矩阵

Worker 代码证据

4. AiRuntimeJob 完整链路（System B — REST Poll）

关键代码位置

5. AiGatewayService — AI Provider 统一网关

调用关系

内部结构

6. 队列配置矩阵

7. EventBus 使用矩阵

关键发现

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

8. Outbox 现状评估

Repository 分析

表结构 (Prisma)

9. Job 状态机现状

AiJob（Legacy）

⚠️ STATUS_TO_LIFECYCLE 映射缺口

AiRuntimeJob（新）

10. 超时 / 重试 / 取消矩阵

11. 依赖边界分析

必须在 WorkerModule 的模块

必须在 AppModule + WorkerModule 共用的模块

仅在 AppModule 的模块（不涉及 Worker）

循环依赖检查

12. 现状总结与 M-AI-03 风险点

必须保持的旧代码

可以独立新增的模块

M-AI-03 需要避免的冲突

关键风险

附录 A：完整文件清单

30 KiB

Raw Blame History