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

**W4** `src/modules/admin-audit-log/audit-log.processor.ts:8-27`
```typescript
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 完成通知机制：

```typescript
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 个状态：

```typescript
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 等） |