docs: ADR-003 冻结统一 Job Engine、Registry、Outbox 与状态机
冻结 M-AI-03 完整实现契约： - 状态机：5 状态、6 允许迁移、6 非法迁移、CAS 并发、Heartbeat/Reaper - Job Definition Registry：TypeScript 接口冻结、注册校验、禁止 switch(jobType) - 队列：ai-interactive/ai-background/ai-analysis 配置、Payload、Worker 边界 - Outbox：ai.job.enqueue 事件、CAS 先于 queue.add 的领取算法、崩溃窗口分析 - Execution Engine：PREPARE/RESOLVE/EXECUTE/PROJECT/COMPLETE 五阶段管线 - Projector：事务边界、AiJobArtifact 幂等、失败回滚 - API：4 端点、JobStatusResponse 接口、6 公开错误码 - 安全：Credential/Redis/日志/Snapshot/Outbox payload 约束 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wangdl
parent
8e5108751c
commit
608129be4e
782
docs/architecture/adr-003-unified-ai-job-engine.md
Normal file
782
docs/architecture/adr-003-unified-ai-job-engine.md
Normal file
@ -0,0 +1,782 @@
|
|||||||
|
# ADR-003：统一 AI Job Engine、Registry、Outbox 与状态机
|
||||||
|
|
||||||
|
## 状态
|
||||||
|
|
||||||
|
已提出（2026-06-20）
|
||||||
|
|
||||||
|
## 背景
|
||||||
|
|
||||||
|
M-AI-03-01 审计（`docs/architecture/m-ai-03-current-execution-audit.md`）确认：
|
||||||
|
|
||||||
|
- 存在两套 Job 系统：**AiJob**（BullMQ + `AiAnalysisWorker`，2 种 jobType）和 **AiRuntimeJob**（REST Poll + 外部 heavy-runtime，5 种 jobType，0 行生产数据）
|
||||||
|
- M-AI-02 已在 `AiAnalysisJob` 物理表上完成 Schema Expand，19 个新字段就绪，Outbox/Artifact/Snapshot 表已创建
|
||||||
|
- **3 个 BullMQ 队列**（notification、domain-events、audit-logs）的 Consumer 已注册但无活跃 Producer
|
||||||
|
- **OutboxRepository** 已完整实现但零调用方，无 Dispatcher
|
||||||
|
- **queueName 不一致**：AiAnalysisRepository 写入 `queueName: 'ai-interactive'`，实际 BullMQ 路由到 `ai-analysis`
|
||||||
|
- **STATUS_TO_LIFECYCLE 映射缺口**：仅 4 个状态，缺少 `cancel_requested` / `cancelled`
|
||||||
|
|
||||||
|
M-AI-03 的目标是在 M-AI-02 Schema Expand 之上，纯代码层实现统一 Job Engine、Registry、Outbox Dispatcher 与 Projector——**无 Prisma Migration**、**不迁移真实业务 Job**、**不修改已有 Worker/队列**。
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 决策
|
||||||
|
|
||||||
|
### 1. 状态机
|
||||||
|
|
||||||
|
#### 1.1 完整状态定义
|
||||||
|
|
||||||
|
```
|
||||||
|
┌──────────┐
|
||||||
|
│ queued │ ← 唯一初始状态（AiJobLifecycleRepository.createJob）
|
||||||
|
└────┬─────┘
|
||||||
|
│ lockJob (CAS)
|
||||||
|
┌────▼─────┐
|
||||||
|
│ running │
|
||||||
|
└────┬─────┘
|
||||||
|
┌────────┼──────────┐
|
||||||
|
│ │ │
|
||||||
|
┌──────▼──┐ ┌──▼───┐ ┌───▼──────┐
|
||||||
|
│succeeded│ │failed│ │cancelled │ ← 终态
|
||||||
|
└─────────┘ └──────┘ └──────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
| 状态 | 说明 | 终态 | 进入条件 |
|
||||||
|
|------|------|------|---------|
|
||||||
|
| `queued` | Job 已创建并入队，等待 Worker 领取 | 否 | `AiJobLifecycleRepository.createJob()` |
|
||||||
|
| `running` | Worker 已锁定并开始执行 | 否 | `AiJobLifecycleRepository.lockJob()` CAS 成功 |
|
||||||
|
| `succeeded` | 执行成功，Projector 已提交结果 | 是 | `AiJobLifecycleRepository.markSucceeded()` |
|
||||||
|
| `failed` | 执行失败且重试耗尽 | 是 | `AiJobLifecycleRepository.markFailed()` |
|
||||||
|
| `cancelled` | 用户/系统取消 | 是 | `AiJobLifecycleRepository.markCancelled()` |
|
||||||
|
|
||||||
|
#### 1.2 允许迁移
|
||||||
|
|
||||||
|
```
|
||||||
|
queued → running (lockJob)
|
||||||
|
queued → cancelled (用户取消，仅 queued 状态允许)
|
||||||
|
running → succeeded (submitResult)
|
||||||
|
running → failed (submitFailure，retryCount >= maxRetryCount)
|
||||||
|
running → cancelled (cancelRequestedAt 已设置，heartbeat 检测)
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 1.3 非法迁移
|
||||||
|
|
||||||
|
以下迁移在任何情况下都不允许：
|
||||||
|
|
||||||
|
```
|
||||||
|
queued → succeeded (必须先 lockJob → running)
|
||||||
|
queued → failed (必须先 lockJob → running)
|
||||||
|
succeeded → * (终态，不可变)
|
||||||
|
failed → * (终态，不可变；需重试应创建新 Job)
|
||||||
|
cancelled → * (终态，不可变)
|
||||||
|
running → queued (不可回退；需重试应通过 retryCount 机制)
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 1.4 唯一写入入口
|
||||||
|
|
||||||
|
**所有** `lifecycleStatus` 的写入必须通过 `AiJobLifecycleRepository` 的以下方法：
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
// AiJobLifecycleRepository — 唯一写入 lifeCycleStatus 的入口
|
||||||
|
export class AiJobLifecycleRepository {
|
||||||
|
createJob(tx: Prisma.TransactionClient, input: CreateJobInput): Promise<AiJob>;
|
||||||
|
lockJob(jobId: string, instanceId: string): Promise<AiJob>; // CAS: queued → running
|
||||||
|
markSucceeded(jobId: string): Promise<AiJob>;
|
||||||
|
markFailed(jobId: string, error: JobErrorInfo): Promise<AiJob>;
|
||||||
|
markCancelled(jobId: string): Promise<AiJob>;
|
||||||
|
|
||||||
|
// 只读
|
||||||
|
findByLifecycleStatus(status: LifecycleStatus[], opts?: { queueName?: string; limit?: number }): Promise<AiJob[]>;
|
||||||
|
findExpiredLocks(thresholdMs: number): Promise<AiJob[]>;
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**禁止**：业务代码直接调用 `prisma.aiJob.update({ where: { id }, data: { lifecycleStatus: '...' } })` 或 `this.repository.updateJobStatus()` 写入 `lifecycleStatus`。
|
||||||
|
|
||||||
|
#### 1.5 Legacy `status` Shadow Write
|
||||||
|
|
||||||
|
过渡期状态映射（M-AI-02-10 已有，本 ADR 扩展）：
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
const STATUS_TO_LIFECYCLE: Record<string, string> = {
|
||||||
|
pending: 'queued',
|
||||||
|
processing: 'running',
|
||||||
|
completed: 'succeeded',
|
||||||
|
failed: 'failed',
|
||||||
|
cancelled: 'cancelled', // 新增 — 修正 #284 发现的映射缺口
|
||||||
|
cancel_requested: 'cancelled', // 新增 — 统一映射
|
||||||
|
};
|
||||||
|
```
|
||||||
|
|
||||||
|
当 `AiJobLifecycleRepository` 变更 `lifecycleStatus` 时，若 legacy `status` 列存在，同步写入映射值（Shadow Write）。M-AI-05 完成后删除 `status` 列。
|
||||||
|
|
||||||
|
#### 1.6 并发状态更新策略
|
||||||
|
|
||||||
|
`lockJob` 使用 **MySQL 乐观锁**（CAS updateMany）：
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
async lockJob(jobId: string, instanceId: string): Promise<AiJob> {
|
||||||
|
const now = new Date();
|
||||||
|
const lockUntil = new Date(now.getTime() + 60_000); // 60s TTL
|
||||||
|
|
||||||
|
const result = await this.prisma.aiJob.updateMany({
|
||||||
|
where: {
|
||||||
|
id: jobId,
|
||||||
|
lifecycleStatus: 'queued',
|
||||||
|
},
|
||||||
|
data: {
|
||||||
|
lifecycleStatus: 'running',
|
||||||
|
lockedBy: instanceId,
|
||||||
|
lockedAt: now,
|
||||||
|
lockUntil,
|
||||||
|
startedAt: now,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
if (result.count === 0) {
|
||||||
|
throw new JobLockConflictError(jobId);
|
||||||
|
}
|
||||||
|
|
||||||
|
return this.prisma.aiJob.findUniqueOrThrow({ where: { id: jobId } });
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 1.7 Heartbeat 与过期锁收割
|
||||||
|
|
||||||
|
| 机制 | 实现 | 频率 |
|
||||||
|
|------|------|------|
|
||||||
|
| Heartbeat | `AiJobLifecycleRepository.extendLock(jobId, instanceId)` → `UPDATE ... SET lockUntil = now() + 60s WHERE lifecycleStatus = 'running' AND lockedBy = instanceId` | 每 30s |
|
||||||
|
| 过期锁收割 | `AiJobLifecycleRepository.findExpiredLocks(90_000)` → 锁超时 90s（60s TTL + 30s 宽限期）未续约 → `markFailed`，触发 BullMQ retry | 每 30s |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 2. Job Definition Registry
|
||||||
|
|
||||||
|
#### 2.1 TypeScript 接口（冻结）
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
// ── Job Definition ──
|
||||||
|
|
||||||
|
interface JobDefinition {
|
||||||
|
/** 全局唯一 jobType，与 AiJob.jobType 对应 */
|
||||||
|
jobType: string;
|
||||||
|
|
||||||
|
/** 元数据 */
|
||||||
|
metadata: {
|
||||||
|
/** 人类可读标签 */
|
||||||
|
label: string;
|
||||||
|
/** 描述 */
|
||||||
|
description: string;
|
||||||
|
/** 所属领域 */
|
||||||
|
domain: 'analysis' | 'generation' | 'import' | 'maintenance';
|
||||||
|
/** 版本 */
|
||||||
|
version: string;
|
||||||
|
};
|
||||||
|
|
||||||
|
/** 路由策略 */
|
||||||
|
queue: {
|
||||||
|
/** 目标队列名 */
|
||||||
|
queueName: 'ai-interactive' | 'ai-background';
|
||||||
|
/** 优先级 0=最高, 100=最低 */
|
||||||
|
defaultPriority: number;
|
||||||
|
};
|
||||||
|
|
||||||
|
/** 执行策略 */
|
||||||
|
execution: {
|
||||||
|
/** 超时（毫秒） */
|
||||||
|
timeoutMs: number;
|
||||||
|
/** 最大重试次数（不含首次） */
|
||||||
|
maxRetries: number;
|
||||||
|
/** BullMQ 重试退避 */
|
||||||
|
retryBackoff: { type: 'exponential'; delay: number };
|
||||||
|
/** 是否支持取消 */
|
||||||
|
cancellable: boolean;
|
||||||
|
/** AbortSignal 超时后行为 */
|
||||||
|
abortStrategy: 'fail' | 'retry';
|
||||||
|
};
|
||||||
|
|
||||||
|
/** 输入 Schema 版本（对应 SnapshotBuilder 产出） */
|
||||||
|
input: {
|
||||||
|
schemaVersion: string;
|
||||||
|
};
|
||||||
|
|
||||||
|
/** 输出 Schema 版本（对应 AI Provider 产出 JSON Schema） */
|
||||||
|
output: {
|
||||||
|
schemaVersion: string;
|
||||||
|
/** Zod schema 引用，用于输出校验 */
|
||||||
|
validator?: string;
|
||||||
|
};
|
||||||
|
|
||||||
|
/** 凭据模式 */
|
||||||
|
credential: {
|
||||||
|
/** 支持的凭据模式 */
|
||||||
|
allowedModes: ('platform_key' | 'user_deepseek_key')[];
|
||||||
|
/** 默认模式 */
|
||||||
|
defaultMode: 'platform_key' | 'user_deepseek_key';
|
||||||
|
};
|
||||||
|
|
||||||
|
/** Projector 引用 */
|
||||||
|
projector?: string; // 对应 ResultProjector 注册 key
|
||||||
|
|
||||||
|
/** 安全约束 */
|
||||||
|
security: {
|
||||||
|
/** 是否需要内容安全检查 */
|
||||||
|
contentSafetyCheck: boolean;
|
||||||
|
/** 输出是否需要脱敏后存储 */
|
||||||
|
outputRedaction: boolean;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 2.2 Registry
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
// JobDefinitionRegistry — 替代未来可能出现的巨型 switch(jobType)
|
||||||
|
|
||||||
|
@Injectable()
|
||||||
|
export class JobDefinitionRegistry {
|
||||||
|
private definitions = new Map<string, JobDefinition>();
|
||||||
|
|
||||||
|
register(def: JobDefinition): void {
|
||||||
|
if (this.definitions.has(def.jobType)) {
|
||||||
|
throw new DuplicateJobDefinitionError(def.jobType);
|
||||||
|
}
|
||||||
|
// 启动时校验
|
||||||
|
this.validate(def);
|
||||||
|
this.definitions.set(def.jobType, def);
|
||||||
|
}
|
||||||
|
|
||||||
|
get(jobType: string): JobDefinition {
|
||||||
|
const def = this.definitions.get(jobType);
|
||||||
|
if (!def) throw new UnknownJobTypeError(jobType);
|
||||||
|
return def;
|
||||||
|
}
|
||||||
|
|
||||||
|
getAll(): JobDefinition[] {
|
||||||
|
return Array.from(this.definitions.values());
|
||||||
|
}
|
||||||
|
|
||||||
|
private validate(def: JobDefinition): void {
|
||||||
|
if (!def.jobType.match(/^[a-z][a-z0-9_]{1,63}$/)) {
|
||||||
|
throw new InvalidJobTypeError(def.jobType);
|
||||||
|
}
|
||||||
|
if (!['ai-interactive', 'ai-background'].includes(def.queue.queueName)) {
|
||||||
|
throw new InvalidQueueError(def.queue.queueName);
|
||||||
|
}
|
||||||
|
if (def.execution.timeoutMs < 1000 || def.execution.timeoutMs > 600_000) {
|
||||||
|
throw new InvalidTimeoutError(def.execution.timeoutMs);
|
||||||
|
}
|
||||||
|
// ... 其他校验
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 2.3 注册生命周期
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
// 在 AiJobModule 的 onModuleInit 中注册
|
||||||
|
|
||||||
|
@Module({ /* ... */ })
|
||||||
|
export class AiJobModule implements OnModuleInit {
|
||||||
|
constructor(private readonly registry: JobDefinitionRegistry) {}
|
||||||
|
|
||||||
|
onModuleInit() {
|
||||||
|
// M-AI-04 才注册 active-recall / feynman-evaluation
|
||||||
|
// M-AI-06 才注册 quiz_generation / flashcard_generation
|
||||||
|
// M-AI-07 才注册 learning_state_analysis / weak_point_analysis / next_action_planning
|
||||||
|
|
||||||
|
// M-AI-03 仅注册 Synthetic Definition（测试用）
|
||||||
|
this.registry.register(SYNTHETIC_JOB_DEFINITION);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 3. 队列架构
|
||||||
|
|
||||||
|
#### 3.1 队列配置
|
||||||
|
|
||||||
|
| 队列名 | 用途 | Worker 进程 | 默认 concurrency | lockDuration | attempts |
|
||||||
|
|--------|------|------------|-----------------|-------------|----------|
|
||||||
|
| `ai-interactive` | 新统一 Engine — 交互式 Job（用户等待结果） | `WorkerModule` 新增 | 2 | 30s | 3 |
|
||||||
|
| `ai-background` | 新统一 Engine — 后台 Job（用户不等待） | `WorkerModule` 新增 | 3 | 60s | 3 |
|
||||||
|
| `ai-analysis` | 旧系统 A — active-recall + feynman-evaluation | `WorkerModule`（已有） | 1 | 30s | 3 |
|
||||||
|
|
||||||
|
**约束**：
|
||||||
|
|
||||||
|
- `ai-interactive` / `ai-background` 仅在 `WorkerModule` 中注册 Processor
|
||||||
|
- 旧 `ai-analysis` 队列不修改、不删除
|
||||||
|
- `document-import`、`notification`、`domain-events`、`audit-logs`、`file-cleanup` 五个队列不在 M-AI-03 范围
|
||||||
|
|
||||||
|
#### 3.2 Queue Payload（冻结）
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
// ── BullMQ Job Payload ──
|
||||||
|
|
||||||
|
interface AiJobPayload {
|
||||||
|
/** AiJob 数据库主键 */
|
||||||
|
jobId: string;
|
||||||
|
/** jobType — Worker 据此 lookup JobDefinition */
|
||||||
|
jobType: string;
|
||||||
|
/** 幂等 Key（与 Outbox dedupeKey 相同） */
|
||||||
|
idempotencyKey: string;
|
||||||
|
/** 创建时间戳 */
|
||||||
|
createdAt: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
// BullMQ jobId 格式: "{queueName}:{jobType}:{idempotencyKey}"
|
||||||
|
// 例如: "ai-interactive:active-recall:user123_20260620_001"
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 3.3 Worker 进程注册边界
|
||||||
|
|
||||||
|
```
|
||||||
|
WorkerModule 新增:
|
||||||
|
├── AiInteractiveJobWorker @Processor('ai-interactive')
|
||||||
|
│ └── AiJobExecutionEngine.execute()
|
||||||
|
└── AiBackgroundJobWorker @Processor('ai-background')
|
||||||
|
└── AiJobExecutionEngine.execute()
|
||||||
|
|
||||||
|
WorkerModule 已有（不修改）:
|
||||||
|
├── AiAnalysisWorker @Processor('ai-analysis')
|
||||||
|
├── DocumentImportWorker @Processor('document-import')
|
||||||
|
├── NotificationWorker @Processor('notification')
|
||||||
|
├── AuditLogProcessor @Processor('audit-logs')
|
||||||
|
└── FileCleanupProcessor @Processor('file-cleanup')
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 4. Outbox
|
||||||
|
|
||||||
|
#### 4.1 事件类型
|
||||||
|
|
||||||
|
| eventType | 触发时机 | dedupeKey 组成 | aggregateType |
|
||||||
|
|-----------|---------|---------------|---------------|
|
||||||
|
| `ai.job.enqueue` | 事务内创建 Job 后 | `{jobType}:{userId}:{idempotencyKey}` | `AiJob` |
|
||||||
|
|
||||||
|
**未来扩展**（不在 M-AI-03 范围）：`ai.job.completed`、`ai.job.failed`
|
||||||
|
|
||||||
|
#### 4.2 Dedupe 策略
|
||||||
|
|
||||||
|
`dedupeKey` 在 Outbox 表上有 **UNIQUE 约束**（`prisma/schema.prisma:2323`）。重复插入 → 数据库层拒绝，调用方视为幂等成功。
|
||||||
|
|
||||||
|
#### 4.3 领取算法（修正 #284 发现的并发竞态）
|
||||||
|
|
||||||
|
```
|
||||||
|
Dispatcher.cycle():
|
||||||
|
1. findDispatchable(N) → 读取 status='pending' AND availableAt <= now()
|
||||||
|
2. FOR EACH event:
|
||||||
|
a. locked = markProcessing(eventId, instanceId) // CAS: pending → processing
|
||||||
|
b. IF locked IS NULL → SKIP（其他 Dispatcher 已抢走）
|
||||||
|
c. success = queue.add(event.payload) // 投递到 BullMQ
|
||||||
|
d. IF success → markPublished(eventId)
|
||||||
|
ELSE → markFailed(eventId, 'DISPATCH_FAILED')
|
||||||
|
```
|
||||||
|
|
||||||
|
**核心修复**：先 CAS 抢锁（步骤 2a），锁成功**后**才投递（步骤 2c）。UT (#284 审计发现的）CAS 失败时跳过而非重试其他事件。修复了 `queue.add()` 先于 `markProcessing()` 导致的重复投递。
|
||||||
|
|
||||||
|
#### 4.4 锁超时与重试
|
||||||
|
|
||||||
|
| 参数 | 值 | 说明 |
|
||||||
|
|------|-----|------|
|
||||||
|
| `markProcessing` CAS 条件 | `status = 'pending'` | 仅 pending 事件可抢锁 |
|
||||||
|
| `lockedAt` TTL | 60s | Dispatcher 崩溃后事件锁有效时间 |
|
||||||
|
| `releaseExpiredLocks` 间隔 | 30s | 收割过期 processing → pending |
|
||||||
|
| 最大 `attemptCount` | 3 | 超过后 `status = 'dead'`（新增终态，区别于 `failed`） |
|
||||||
|
| 未知 eventType | 记录 warning 日志，标记 `published`（避免阻塞队列），不投递 | — |
|
||||||
|
|
||||||
|
#### 4.5 崩溃窗口处理
|
||||||
|
|
||||||
|
Dispatcher 或 BullMQ Redis 在 `markProcessing()` 成功但 `queue.add()` 尚未执行时崩溃：
|
||||||
|
|
||||||
|
- `lockedAt` 超时（60s）→ `releaseExpiredLocks()` 重置为 `pending` → 下一个周期重新领取
|
||||||
|
- 潜在重复投递窗口：**60s**
|
||||||
|
- 减轻措施：BullMQ `jobId = "{queueName}:{jobType}:{idempotencyKey}"` 确保相同 dedupeKey 的入队幂等（BullMQ 同 jobId 重复 add 视为 no-op）
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 5. Execution Engine
|
||||||
|
|
||||||
|
#### 5.1 执行阶段（管线）
|
||||||
|
|
||||||
|
```
|
||||||
|
Job (BullMQ)
|
||||||
|
│
|
||||||
|
├─ 1. PREPARE
|
||||||
|
│ ├─ lookup JobDefinition via Registry
|
||||||
|
│ ├─ load AiJob from DB
|
||||||
|
│ ├─ verify lifecycleStatus = 'queued'
|
||||||
|
│ └─ lockJob (CAS queued → running)
|
||||||
|
│
|
||||||
|
├─ 2. RESOLVE
|
||||||
|
│ ├─ resolve snapshot (SnapshotBuilder)
|
||||||
|
│ ├─ resolve credential (credentialMode / credentialId)
|
||||||
|
│ ├─ resolve prompt (PromptTemplateService)
|
||||||
|
│ └─ resolve provider (ModelRouter → AiProvider)
|
||||||
|
│
|
||||||
|
├─ 3. EXECUTE
|
||||||
|
│ ├─ AbortSignal (timeout from JobDefinition.execution.timeoutMs)
|
||||||
|
│ ├─ content safety check (if security.contentSafetyCheck)
|
||||||
|
│ ├─ AiGatewayService.generate(request)
|
||||||
|
│ ├─ parse & validate output (outputSchema)
|
||||||
|
│ └─ compute outputHash
|
||||||
|
│
|
||||||
|
├─ 4. PROJECT
|
||||||
|
│ ├─ begin Prisma transaction
|
||||||
|
│ ├─ ResultProjector.project(output) → Artifact(s)
|
||||||
|
│ ├─ AiJobLifecycleRepository.markSucceeded()
|
||||||
|
│ ├─ OutboxRepository.createInTransaction(tx, ai.job.completed) // 未来
|
||||||
|
│ └─ commit transaction
|
||||||
|
│
|
||||||
|
└─ 5. COMPLETE
|
||||||
|
├─ usage logging (ModelInvocationLog)
|
||||||
|
├─ publish domain event (sync EventBus)
|
||||||
|
└─ return
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 5.2 错误分类
|
||||||
|
|
||||||
|
| 错误类别 | 错误码前缀 | 可重试 | 行为 |
|
||||||
|
|---------|-----------|--------|------|
|
||||||
|
| `PROVIDER_*` | 外部 AI Provider 错误（超时、5xx、限流） | ✅ | retry with backoff |
|
||||||
|
| `VALIDATION_*` | 输出 Schema 校验失败 | ✅ | retry（重试时调整 prompt） |
|
||||||
|
| `CONTENT_SAFETY_*` | 内容安全拦截 | ❌ | 标记 failed，不重试 |
|
||||||
|
| `TIMEOUT` | Job 超时 | ✅（如果 abortStrategy = 'retry'） | 根据 Definition 决定 |
|
||||||
|
| `CREDENTIAL_*` | 凭据无效/过期 | ❌ | 标记 failed，用户通知 |
|
||||||
|
| `CANCELLED` | 用户/系统取消 | ❌ | 标记 cancelled |
|
||||||
|
| `INTERNAL_*` | 内部错误（DB、Redis、代码异常） | ✅（最多 1 次） | 重试 1 次后 failed |
|
||||||
|
|
||||||
|
#### 5.3 Timeout 与 AbortSignal
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
// Execution Engine 内部
|
||||||
|
const def = this.registry.get(jobType);
|
||||||
|
const controller = new AbortController();
|
||||||
|
const timeoutId = setTimeout(() => {
|
||||||
|
controller.abort();
|
||||||
|
this.logger.warn(`Job ${jobId} timed out after ${def.execution.timeoutMs}ms`);
|
||||||
|
}, def.execution.timeoutMs);
|
||||||
|
|
||||||
|
try {
|
||||||
|
const output = await this.aiGateway.generate(request, controller.signal);
|
||||||
|
clearTimeout(timeoutId);
|
||||||
|
// ...
|
||||||
|
} catch (err) {
|
||||||
|
clearTimeout(timeoutId);
|
||||||
|
if (controller.signal.aborted) {
|
||||||
|
throw new JobTimeoutError(jobId, def.execution.timeoutMs);
|
||||||
|
}
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 5.4 取消
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
// 用户取消流程
|
||||||
|
// 1. UserAiController.cancelJob() → AiJobLifecycleRepository.markCancelled(jobId)
|
||||||
|
// - queued → cancelled（直接）
|
||||||
|
// - running → 设置 cancelRequestedAt，Heartbeat 检测后 markCancelled
|
||||||
|
// - succeeded/failed/cancelled → 409 ConflictException（终态不可取消）
|
||||||
|
|
||||||
|
// 2. Worker Heartbeat 检测
|
||||||
|
// Execution Engine 在每次 Provider 调用前后检查：
|
||||||
|
// if (job.cancelRequestedAt != null) throw new JobCancelledError(jobId);
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 5.5 Retry
|
||||||
|
|
||||||
|
| 层级 | 机制 | 配置来源 |
|
||||||
|
|------|------|---------|
|
||||||
|
| BullMQ | attempts + backoff | `JobDefinition.execution.maxRetries` + `retryBackoff` |
|
||||||
|
| Execution Engine | 内部重试循环 | `JobDefinition.execution.maxRetries` |
|
||||||
|
| Outbox | attemptCount < 3 → pending；>= 3 → dead | OutboxRepository |
|
||||||
|
|
||||||
|
**约束**：BullMQ retry 与 Engine 内部 retry 不叠加。Engine 内部错误不抛出（不触发 BullMQ retry），而是在 Engine 内完成所有重试后，一次性 `markFailed` 或 `markSucceeded`。
|
||||||
|
|
||||||
|
#### 5.6 ValidatedOutput 与 OutputHash
|
||||||
|
|
||||||
|
```
|
||||||
|
outputHash = SHA256(JSON.stringify(validatedOutput, Object.keys(validatedOutput).sort()))
|
||||||
|
```
|
||||||
|
|
||||||
|
- 存储在 `AiJob.outputHash`
|
||||||
|
- 用于去重（相同 input + outputHash → 幂等返回已有结果）
|
||||||
|
- 用于 Artifact 去重（相同 outputHash → 跳过 Projector）
|
||||||
|
|
||||||
|
#### 5.7 Usage Logging
|
||||||
|
|
||||||
|
每次 AI Provider 调用后写入 `AiUsageLog`（M-AI-02-08 已扩展 `jobId`、`attemptNo`、`credentialMode`、`errorCode` 字段），关联 `AiJob.id`。
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 6. Result Projector
|
||||||
|
|
||||||
|
#### 6.1 接口（冻结）
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
interface ResultProjector {
|
||||||
|
/** 全局唯一 key，对应 JobDefinition.projector */
|
||||||
|
readonly key: string;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 在 Prisma Transaction 内执行投影。
|
||||||
|
* 如果 Artifact 已存在（幂等），返回已有引用。
|
||||||
|
*/
|
||||||
|
project(
|
||||||
|
tx: Prisma.TransactionClient,
|
||||||
|
job: AiJob,
|
||||||
|
validatedOutput: Record<string, any>,
|
||||||
|
): Promise<ProjectionResult>;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface ProjectionResult {
|
||||||
|
/** 创建的 Artifact 引用列表 */
|
||||||
|
artifacts: Array<{
|
||||||
|
artifactType: string;
|
||||||
|
artifactId: string;
|
||||||
|
ordinal: number;
|
||||||
|
}>;
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 6.2 事务边界
|
||||||
|
|
||||||
|
```
|
||||||
|
BEGIN TRANSACTION
|
||||||
|
1. ResultProjector.project(tx, job, validatedOutput)
|
||||||
|
→ 写入业务表（QuizQuestion / Flashcard / AiLearningAnalysis / WeakPointCandidate / NextActionRecommendation 等）
|
||||||
|
→ 写入 AiJobArtifact (jobId, artifactType, artifactId)
|
||||||
|
2. AiJobLifecycleRepository.markSucceeded(tx, jobId)
|
||||||
|
3. (未来) OutboxRepository.createInTransaction(tx, { eventType: 'ai.job.completed', ... })
|
||||||
|
COMMIT
|
||||||
|
```
|
||||||
|
|
||||||
|
- **全部成功** → Commit
|
||||||
|
- **任何步骤失败** → Rollback，Job 重新进入 retry 流程
|
||||||
|
- **Projector 执行中触发唯一约束冲突** → 视为幂等成功，跳过
|
||||||
|
|
||||||
|
#### 6.3 幂等保证
|
||||||
|
|
||||||
|
- `AiJobArtifact` 有 `@@unique([jobId, artifactType, artifactId])`
|
||||||
|
- 重复 Projection → DB UNIQUE 冲突 → catch → 返回已有 Artifact 引用
|
||||||
|
- `{{##duplicate}}` 标记在 Projector 日志中
|
||||||
|
|
||||||
|
#### 6.4 失败回滚
|
||||||
|
|
||||||
|
Projector 抛错 → Transaction Rollback → `JobFailedError` → Engine retry（如果可重试）或 markFailed
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 7. API 设计
|
||||||
|
|
||||||
|
#### 7.1 端点
|
||||||
|
|
||||||
|
| 方法 | 路径 | 权限 | 说明 |
|
||||||
|
|------|------|------|------|
|
||||||
|
| `GET` | `/api/ai/jobs/:jobId` | 用户（自己的 Job）/ Admin | Job 状态查询 |
|
||||||
|
| `POST` | `/api/ai/jobs/:jobId/cancel` | 用户（自己的 Job） | 取消 Job |
|
||||||
|
| `GET` | `/admin-api/ai/jobs` | Admin（SUPER_ADMIN） | Job 列表查询（支持按 jobType/status/queueName/时间范围筛选） |
|
||||||
|
| `POST` | `/admin-api/ai/jobs/:jobId/retry` | Admin（SUPER_ADMIN） | 手动重试失败 Job（创建新 Job，复制 input） |
|
||||||
|
|
||||||
|
#### 7.2 Job 状态响应（冻结）
|
||||||
|
|
||||||
|
```typescript
|
||||||
|
interface JobStatusResponse {
|
||||||
|
jobId: string;
|
||||||
|
jobType: string;
|
||||||
|
lifecycleStatus: 'queued' | 'running' | 'succeeded' | 'failed' | 'cancelled';
|
||||||
|
queueName: string;
|
||||||
|
priority: number;
|
||||||
|
progress: number; // 0–100
|
||||||
|
createdAt: string;
|
||||||
|
startedAt?: string;
|
||||||
|
finishedAt?: string;
|
||||||
|
errorCode?: string;
|
||||||
|
publicErrorMessage?: string; // 面向用户的错误，不含内部堆栈
|
||||||
|
result?: {
|
||||||
|
artifactCount: number;
|
||||||
|
artifacts: Array<{ artifactType: string; artifactId: string }>;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 7.3 公开错误码
|
||||||
|
|
||||||
|
| 错误码 | HTTP 状态 | 说明 |
|
||||||
|
|--------|----------|------|
|
||||||
|
| `JOB_NOT_FOUND` | 404 | Job 不存在或不属于当前用户 |
|
||||||
|
| `JOB_NOT_CANCELLABLE` | 409 | Job 处于终态，不可取消 |
|
||||||
|
| `JOB_ALREADY_CANCELLED` | 409 | Job 已被取消 |
|
||||||
|
| `TOO_MANY_JOBS` | 429 | 用户并发 Job 数超限 |
|
||||||
|
| `QUOTA_EXCEEDED` | 429 | 用户日/月额度用尽 |
|
||||||
|
| `PLATFORM_UNAVAILABLE` | 503 | 平台 AI 服务不可用 |
|
||||||
|
|
||||||
|
#### 7.4 Admin 公开错误
|
||||||
|
|
||||||
|
Admin 端点额外返回 `internalErrorMessage`（仅 SUPER_ADMIN 角色可见），包含内部堆栈信息。
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 8. 安全约束
|
||||||
|
|
||||||
|
#### 8.1 Credential
|
||||||
|
|
||||||
|
- 用户 Key（`user_deepseek_key` mode）：通过 `CredentialEncryptionService` 加解密，AiJob 表仅存储 `credentialId`（FK → `UserModelCredential`）
|
||||||
|
- 平台 Key（`platform_key` mode）：从环境变量注入，不在 DB 中存储
|
||||||
|
- Runtime Resolve 端点：仅验证 Runtime Service Token 后返回解密 Key
|
||||||
|
|
||||||
|
#### 8.2 Redis
|
||||||
|
|
||||||
|
- BullMQ 连接通过 Redis ACL 用户隔离
|
||||||
|
- Heartbeat Key：`worker:heartbeat:{instanceId}`，TTL 90s
|
||||||
|
- 取消信号：`job:cancel:{jobId}`，TTL 5min
|
||||||
|
- 无敏感数据（AiJob payload）存储在 Redis
|
||||||
|
|
||||||
|
#### 8.3 日志
|
||||||
|
|
||||||
|
- `AiJob.payload`（input）不在日志中输出
|
||||||
|
- `validatedOutput` 不在日志中输出（仅 outputHash）
|
||||||
|
- `publicErrorMessage` 不含 IP、路径、堆栈
|
||||||
|
- `internalErrorMessage` 仅 SUPER_ADMIN 可见
|
||||||
|
|
||||||
|
#### 8.4 Snapshot
|
||||||
|
|
||||||
|
- `AiJobSnapshot.content` 存储在 DB（JSON 列），遵循 `redactionVersion` 脱敏规则
|
||||||
|
- `expiresAt` 过期后由 `SnapshotCleanupService` 清理
|
||||||
|
|
||||||
|
#### 8.5 Outbox Payload
|
||||||
|
|
||||||
|
- `OutboxEvent.payload` 仅包含 `{ jobId, jobType, queueName, priority }`
|
||||||
|
- 不含 `userId`、`credentialId`、snapshot 内容
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 9. 模块图
|
||||||
|
|
||||||
|
```
|
||||||
|
AiJobModule（新增）
|
||||||
|
├── AiJobService // 统一 Job 创建入口（事务内：AiJob + Outbox）
|
||||||
|
├── AiJobLifecycleRepository // 唯一 lifecycleStatus 写入入口
|
||||||
|
├── JobDefinitionRegistry // JobType → Definition 注册表
|
||||||
|
├── AiJobExecutionEngine // 统一执行管线（PREPARE/RESOLVE/EXECUTE/PROJECT/COMPLETE）
|
||||||
|
├── OutboxDispatcher // 轮询 OutboxEvent → CAS 领取 → BullMQ 投递
|
||||||
|
├── ResultProjector // 投影器接口 + 内置实现（业务 Issue 注册）
|
||||||
|
│
|
||||||
|
├── AiInteractiveJobWorker // @Processor('ai-interactive')
|
||||||
|
└── AiBackgroundJobWorker // @Processor('ai-background')
|
||||||
|
|
||||||
|
依赖模块（已有，不修改）：
|
||||||
|
├── AiGatewayService // AI Provider 调用
|
||||||
|
├── ModelRouter // Provider 路由
|
||||||
|
├── PromptTemplateService // Prompt 模板
|
||||||
|
├── AiCostCalculatorService // 成本计算
|
||||||
|
├── AiUsageLogService // 用量记录
|
||||||
|
├── ContentSafetyService // 内容安全
|
||||||
|
├── CredentialEncryptionService // 凭据加解密
|
||||||
|
├── SnapshotBuilderService // Snapshot 构建
|
||||||
|
├── QueueService // BullMQ 入队（仅 OutboxDispatcher 使用）
|
||||||
|
└── EventBusService // 领域事件（sync publish）
|
||||||
|
```
|
||||||
|
|
||||||
|
### 10. 时序图
|
||||||
|
|
||||||
|
```
|
||||||
|
User AiJobService AiJobLifecycleRepo OutboxRepo BullMQ Worker Engine AiGateway
|
||||||
|
|
||||||
|
│ POST /ai/jobs │ │ │ │ │ │
|
||||||
|
├─────────────────────>│ │ │ │ │ │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │─ BEGIN TX ────────> │ │ │ │
|
||||||
|
│ │─ createJob(tx) ──>│ │ │ │ │
|
||||||
|
│ │ │─INSERT AiJob─────│ │ │ │
|
||||||
|
│ │ │ (queued) │ │ │ │
|
||||||
|
│ │─ createInTx(tx) ──────────────────>│ │ │ │
|
||||||
|
│ │ │ INSERT Outbox │ │ │ │
|
||||||
|
│ │─ COMMIT ───────────────────────────────────────────────────────────────────────────────│
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│<── { jobId, queued } │ │ │ │ │ │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │ │ ╔═════════════════════════════╗ │
|
||||||
|
│ │ │ ║ OutboxDispatcher 异步 ║ │
|
||||||
|
│ │ │ ║ ║ │
|
||||||
|
│ │ │ ║ findDispatchable() ║ │
|
||||||
|
│ │ │ ║ markProcessing(CAS) ║ │
|
||||||
|
│ │ │ ║ queue.add('ai-interactive')─>│ │
|
||||||
|
│ │ │ ║ markPublished() ║ │
|
||||||
|
│ │ │ ╚═════════════════════════════╝ │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │ │ │ ai-interactive│ │ │
|
||||||
|
│ │ │ │<───────────────│ │ │
|
||||||
|
│ │ │ │ │─ process() ─>│ │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │ │ │ │ │─ 1. PREPARE │
|
||||||
|
│ │ │ │ │ │ lockJob(CAS) │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │ │ │ │ │─ 2. RESOLVE │
|
||||||
|
│ │ │ │ │ │ snapshot │
|
||||||
|
│ │ │ │ │ │ credential │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │ │ │ │ │─ 3. EXECUTE ──────────────────>│
|
||||||
|
│ │ │ │ │ │ │generate()
|
||||||
|
│ │ │ │ │ │<──────── ─ ─ ─ ─ ─│
|
||||||
|
│ │ │ │ │ │ validatedOutput │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │ │ │ │ │─ 4. PROJECT │
|
||||||
|
│ │ │ │ │ │ markSucceeded │
|
||||||
|
│ │ │ │ │ │ create Artifact │
|
||||||
|
│ │ │ │ │ │ │
|
||||||
|
│ │ │ │ │ │─ 5. COMPLETE │
|
||||||
|
│ │ │ │ │ │ usage log │
|
||||||
|
│ │ │ │ │ │ event │
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 11. 测试矩阵
|
||||||
|
|
||||||
|
| 测试层级 | 范围 | Issue |
|
||||||
|
|---------|------|-------|
|
||||||
|
| 单元测试 | `AiJobLifecycleRepository` — 状态迁移、非法迁移、CAS 并发 | #286 |
|
||||||
|
| 单元测试 | `JobDefinitionRegistry` — 注册、查重、校验 | #287 |
|
||||||
|
| 单元测试 | `OutboxDispatcher` — 领取算法、CAS 竞争、崩溃恢复 | #290 |
|
||||||
|
| 单元测试 | `AiJobExecutionEngine` — 正常执行、超时、取消、重试 | #291 |
|
||||||
|
| 单元测试 | `ResultProjector` — 事务提交、幂等、失败回滚 | #292 |
|
||||||
|
| 集成测试 | Synthetic Job Definition 端到端 | #294 |
|
||||||
|
| 集成测试 | `AiJobService` + `OutboxRepository` 事务原子性 | #289 |
|
||||||
|
| 集成测试 | Admin API 查询/取消/重试 | #293 |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 12. 非目标（M-AI-03 明确不包含）
|
||||||
|
|
||||||
|
1. **无 Prisma Migration** — 全部在 M-AI-02 Schema 基础上实现
|
||||||
|
2. **不迁移真实业务 Job** — `ai-analysis` 队列的 active-recall 和 feynman-evaluation 继续走旧链路
|
||||||
|
3. **不创建通用公开创建接口** — `AiJobService.createJob()` 为 internal-only，外部通过具体业务 Service 调用
|
||||||
|
4. **不修改 AiRuntimeJob 表** — 不碰、不读、不写
|
||||||
|
5. **不修改已有 Worker** — `AiAnalysisWorker`、`DocumentImportWorker` 等不受影响
|
||||||
|
6. **不引入新的第三方依赖**
|
||||||
|
7. **不替换 BullMQ** — 队列基础设施不变
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 影响
|
||||||
|
|
||||||
|
- 所有 M-AI-03 后续 Issue（#286–#295）必须引用本 ADR
|
||||||
|
- `ai-interactive` / `ai-background` 队列在 `QueueModule` 中注册
|
||||||
|
- `OutboxDispatcher` 在 `WorkerModule` 中注册为独立 Service（非 BullMQ Processor）
|
||||||
|
|
||||||
|
## 回滚
|
||||||
|
|
||||||
|
每个 Issue 独立回滚：revert 对应 commit。本 ADR 无代码变更，无需回滚。
|
||||||
|
|
||||||
|
## 证据
|
||||||
|
|
||||||
|
- M-AI-03-01 审计：`docs/architecture/m-ai-03-current-execution-audit.md`
|
||||||
|
- M-AI-02 Schema 冻结：`docs/architecture/adr-002-ai-job-database-expand.md`
|
||||||
|
- M-AI-01 架构决策：`docs/architecture/adr-001-unified-ai-architecture.md`
|
||||||
|
- 当前 Prisma Schema：`prisma/schema.prisma:568-677`（AiJob、AiJobSnapshot、AiJobArtifact）
|
||||||
|
- Outbox Repository：`src/infrastructure/outbox/outbox.repository.ts:27-139`
|
||||||
|
- Queue Definitions：`src/infrastructure/queue/queue-definitions.ts:94-101`
|
||||||
|
- Worker Module：`src/worker.module.ts:1-82`
|
||||||
Loading…
x
Reference in New Issue
Block a user