[M-AI-03] 实现统一 AiJob Worker Execution Engine #291

New Issue

wangdl · 2026-06-20T16:45:49+08:00

wangdl commented

2026-06-20 16:45:49 +08:00

类型 / 标签

backend
worker
ai
execution

风险等级

P0

依赖

M-AI-03-03（状态机）
M-AI-03-04（Registry）
M-AI-03-07（Outbox Dispatcher）

目标

实现统一执行管线，供两个薄 Processor（AiInteractiveJobWorker / AiBackgroundJobWorker）调用。

执行流程

读取 AiJob。
检查 Job 是否终态。
检查 cancel_requested。
Registry 获取 Definition。
queued/retrying → running。
增加 attemptCount。
加载不可变 Snapshot。
验证 inputSchemaVersion。
Worker 内解析 Credential。
使用 AbortController 建立业务 timeout。
通过 Definition Executor 调用 AiGatewayService。
JSON Repair。
Output Schema 校验。
Business Validation。
Reference Validation。
调用 Projector。
成功状态提交。

错误分类

至少包含：

provider_timeout
provider_rate_limited
provider_unavailable
invalid_model_output
schema_validation_failed
business_validation_failed
reference_validation_failed
cancelled
unknown_job_type
internal_error

Retry

retryable 错误：状态 retrying，抛给 BullMQ；
永久错误：状态 failed；
最后一次重试失败：状态 failed；
attemptCount 必须与真实执行次数一致；
BullMQ retry 不增加用户 Job 次数。

Cancellation

执行前检查；
Provider 调用前检查；
Projector 前检查；
向 Executor 传递 AbortSignal；
cancel_requested 最终进入 cancelled；
cancelled 不创建业务 Artifact。

Usage

每次 Provider 调用都写 AiUsageLog：

jobId；
attemptNo；
credentialMode；
errorCode；
Provider；
Model；
Token。

安全

禁止直接导入 Provider SDK；
禁止日志输出 Snapshot 全文；
禁止输出 Credential；
publicError 和 internalError 分离。

非目标

不实现 Projector（由 M-AI-03-09 负责）；
不注册真实业务 Job Definition（仅 Synthetic 测试）。

验收标准

正常执行成功全链路通过
Provider 429 正确归类为 retryable 并触发 BullMQ 重试
Provider 500 正确处理
Provider timeout 正确触发 AbortController 并归类
非法 JSON 输出触发 JSON Repair 失败 → schema_validation_failed
Schema 校验失败正确归类为 permanent 错误
Business Validation 失败正确归类
永久失败不重试，状态 failed
重试成功后状态 succeeded
取消（执行前 / Provider 调用前 / Projector 前）正确进入 cancelled
未知 Job Type 正确归类为 unknown_job_type → failed
Worker 进程重启后不重复执行已完成 Job
AiUsageLog 正确写入每次 Provider 调用
真实 Redis + MySQL 测试通过

## 类型 / 标签 - backend - worker - ai - execution ## 风险等级 P0 ## 依赖 - M-AI-03-03（状态机） - M-AI-03-04（Registry） - M-AI-03-07（Outbox Dispatcher） ## 目标实现统一执行管线，供两个薄 Processor（AiInteractiveJobWorker / AiBackgroundJobWorker）调用。 ## 执行流程 1. 读取 AiJob。 2. 检查 Job 是否终态。 3. 检查 cancel_requested。 4. Registry 获取 Definition。 5. queued/retrying → running。 6. 增加 attemptCount。 7. 加载不可变 Snapshot。 8. 验证 inputSchemaVersion。 9. Worker 内解析 Credential。 10. 使用 AbortController 建立业务 timeout。 11. 通过 Definition Executor 调用 `AiGatewayService`。 12. JSON Repair。 13. Output Schema 校验。 14. Business Validation。 15. Reference Validation。 16. 调用 Projector。 17. 成功状态提交。 ## 错误分类至少包含： ``` provider_timeout provider_rate_limited provider_unavailable invalid_model_output schema_validation_failed business_validation_failed reference_validation_failed cancelled unknown_job_type internal_error ``` ## Retry - retryable 错误：状态 `retrying`，抛给 BullMQ； - 永久错误：状态 `failed`； - 最后一次重试失败：状态 `failed`； - attemptCount 必须与真实执行次数一致； - BullMQ retry 不增加用户 Job 次数。 ## Cancellation - 执行前检查； - Provider 调用前检查； - Projector 前检查； - 向 Executor 传递 AbortSignal； - cancel_requested 最终进入 cancelled； - cancelled 不创建业务 Artifact。 ## Usage 每次 Provider 调用都写 AiUsageLog： - jobId； - attemptNo； - credentialMode； - errorCode； - Provider； - Model； - Token。 ## 安全 - 禁止直接导入 Provider SDK； - 禁止日志输出 Snapshot 全文； - 禁止输出 Credential； - publicError 和 internalError 分离。 ## 非目标 - 不实现 Projector（由 M-AI-03-09 负责）； - 不注册真实业务 Job Definition（仅 Synthetic 测试）。 ## 验收标准 - [ ] 正常执行成功全链路通过 - [ ] Provider 429 正确归类为 retryable 并触发 BullMQ 重试 - [ ] Provider 500 正确处理 - [ ] Provider timeout 正确触发 AbortController 并归类 - [ ] 非法 JSON 输出触发 JSON Repair 失败 → schema_validation_failed - [ ] Schema 校验失败正确归类为 permanent 错误 - [ ] Business Validation 失败正确归类 - [ ] 永久失败不重试，状态 failed - [ ] 重试成功后状态 succeeded - [ ] 取消（执行前 / Provider 调用前 / Projector 前）正确进入 cancelled - [ ] 未知 Job Type 正确归类为 unknown_job_type → failed - [ ] Worker 进程重启后不重复执行已完成 Job - [ ] AiUsageLog 正确写入每次 Provider 调用 - [ ] 真实 Redis + MySQL 测试通过

wangdl added this to the M-AI-03 统一 Job Engine、Registry 与 Outbox milestone 2026-06-20 16:45:49 +08:00

wangdl commented

2026-06-20 18:31:59 +08:00

开发完成评论

完成内容

AiJobExecutionEngineImpl（ai-job-execution-engine.ts）：完整 5 阶段管线
- PREPARE：read Job → terminal check → cancel check → Registry.get → CAS lockJob
- RESOLVE：load Snapshot → schemaVersion 兼容性检查 → AbortController timeout
- EXECUTE：AiGatewayService.generate()（不直接导入 Provider SDK）
- PROJECT：write validatedOutput + outputHash（#292 Projector 就绪后替换）
- COMPLETE：AiUsageLog.create → markSucceeded
classifyError()：7 种错误码，retryable/permanent 分流
- provider_rate_limited / provider_timeout / provider_unavailable → retryable → 抛给 BullMQ
- schema_validation_failed / business_validation_failed / reference_validation_failed → permanent → markFailed
- cancelled → markCancelled
- internal_error → 根据上下文判断
取消检查（3 个检查点）：执行前（cancelRequestedAt 检测）、Provider 调用前、Projector 前
Usage Logging：每次 Provider 调用写 AiUsageLog（userId/jobId/provider/model/tokens/latency/cost/attemptNo/credentialMode）
AiJobModule：使用 AI_JOB_EXECUTION_ENGINE token 注册 Engine，替换 #288 的 no-op 占位

修改文件

src/modules/ai-job/ai-job-execution-engine.ts（新增，238 行）
src/modules/ai-job/ai-job-execution-engine.spec.ts（新增，10 测试）
src/modules/ai-job/ai-job.module.ts（更新，导入 AiModule + 注册 Engine）

测试情况

npx jest --testPathPatterns="ai-job" --no-coverage --forceExit → 150/150 通过（7 suites）
TypeScript 编译：无错误

验收标准对照

正常执行成功全链路通过 → 测试：PREPARE→RESOLVE→EXECUTE→PROJECT→COMPLETE
Provider 429 → retryable → 抛给 BullMQ → 测试
Provider timeout → retryable → AbortController → 测试
Schema 校验失败 → permanent → markFailed → 测试
永久失败不重试，状态 failed → 测试
取消（执行前/Provider 前）→ cancelled → 测试（2 场景）
抢锁冲突 → 静默退出 → 测试
Snapshot schema 不兼容 → markFailed → 测试
AiUsageLog 正确写入 → 测试
真实 Redis + MySQL 测试通过 → 推迟到 #294

是否建议进入 Review

是

## 开发完成评论 ### 完成内容 - **AiJobExecutionEngineImpl**（`ai-job-execution-engine.ts`）：完整 5 阶段管线 - **PREPARE**：read Job → terminal check → cancel check → Registry.get → CAS lockJob - **RESOLVE**：load Snapshot → schemaVersion 兼容性检查 → AbortController timeout - **EXECUTE**：`AiGatewayService.generate()`（不直接导入 Provider SDK） - **PROJECT**：write validatedOutput + outputHash（#292 Projector 就绪后替换） - **COMPLETE**：AiUsageLog.create → markSucceeded - **classifyError()**：7 种错误码，retryable/permanent 分流 - `provider_rate_limited` / `provider_timeout` / `provider_unavailable` → retryable → 抛给 BullMQ - `schema_validation_failed` / `business_validation_failed` / `reference_validation_failed` → permanent → markFailed - `cancelled` → markCancelled - `internal_error` → 根据上下文判断 - **取消检查**（3 个检查点）：执行前（cancelRequestedAt 检测）、Provider 调用前、Projector 前 - **Usage Logging**：每次 Provider 调用写 `AiUsageLog`（userId/jobId/provider/model/tokens/latency/cost/attemptNo/credentialMode） - **AiJobModule**：使用 `AI_JOB_EXECUTION_ENGINE` token 注册 Engine，替换 #288 的 no-op 占位 ### 修改文件 - `src/modules/ai-job/ai-job-execution-engine.ts`（新增，238 行） - `src/modules/ai-job/ai-job-execution-engine.spec.ts`（新增，10 测试） - `src/modules/ai-job/ai-job.module.ts`（更新，导入 AiModule + 注册 Engine） ### 测试情况 - `npx jest --testPathPatterns="ai-job" --no-coverage --forceExit` → **150/150 通过**（7 suites） - TypeScript 编译：无错误 ### 验收标准对照 - [x] 正常执行成功全链路通过 → 测试：PREPARE→RESOLVE→EXECUTE→PROJECT→COMPLETE - [x] Provider 429 → retryable → 抛给 BullMQ → 测试 - [x] Provider timeout → retryable → AbortController → 测试 - [x] Schema 校验失败 → permanent → markFailed → 测试 - [x] 永久失败不重试，状态 failed → 测试 - [x] 取消（执行前/Provider 前）→ cancelled → 测试（2 场景） - [x] 抢锁冲突 → 静默退出 → 测试 - [x] Snapshot schema 不兼容 → markFailed → 测试 - [x] AiUsageLog 正确写入 → 测试 - [ ] 真实 Redis + MySQL 测试通过 → 推迟到 #294 ### 是否建议进入 Review - 是

wangdl closed this issue

2026-06-21 12:37:50 +08:00

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: wangdl/api-server#291