diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 7b4c72c..3672661 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -292,10 +292,43 @@ jobs: timeout-minutes: 5 # ═══════════════════════════════════════════════════════════ - # Job 4: Deploy (only if all gates pass) + # Job 4: M-AI-03 Synthetic E2E + # ═══════════════════════════════════════════════════════════ + m-ai-03-synthetic-e2e: + needs: build-and-unit + runs-on: prod + steps: + - name: Restore build artifacts + run: | + cp -a $ARTIFACT_DIR/dist $WORKDIR/dist + cp -a $ARTIFACT_DIR/node_modules $WORKDIR/node_modules + cp -a $ARTIFACT_DIR/prisma $WORKDIR/prisma + + - name: Ensure infrastructure + run: | + docker start mysql redis 2>/dev/null || true + sleep 2 + + - name: Apply database migrations + run: | + cd $WORKDIR + DATABASE_URL="${{ secrets.DATABASE_URL }}" npx prisma migrate deploy + timeout-minutes: 5 + + - name: M-AI-03 Synthetic E2E + run: | + cd $WORKDIR + NODE_ENV=test AI_JOB_SYNTHETIC_ENABLED=true \ + DATABASE_URL="${{ secrets.DATABASE_URL }}" \ + REDIS_URL="${{ secrets.REDIS_URL || 'redis://localhost:6379' }}" \ + npx jest --config test/jest-e2e.json --testPathPatterns="m-ai-03-synthetic" --forceExit --verbose + timeout-minutes: 10 + + # ═══════════════════════════════════════════════════════════ + # Job 5: Deploy (only if all gates pass) # ═══════════════════════════════════════════════════════════ deploy: - needs: [current-integration, backward-compat] + needs: [current-integration, backward-compat, m-ai-03-synthetic-e2e] runs-on: prod steps: - name: Restore build artifacts diff --git a/docs/architecture/m-ai-03-gate-audit.md b/docs/architecture/m-ai-03-gate-audit.md new file mode 100644 index 0000000..7b4224f --- /dev/null +++ b/docs/architecture/m-ai-03-gate-audit.md @@ -0,0 +1,255 @@ +# M-AI-03 GATE 独立验收审核报告 + +> 审核日期：2026-06-21 +> 审核人：独立审核代理（非 M-AI-03 开发执行者） +> 基线：`98e442e` (M-AI-02 GATE) +> HEAD：`5108a9a` + +--- + +## 1. 审核结论 + +**CONDITIONAL PASS** + +- 无 P0 问题 +- 1 个 P1 问题（Admin 重试绕过 Registry + CreationService + 非原子） +- 2 个 P2 问题（queue-definitions.spec.ts 测试断言过时、18/19 Synthetic E2E pending infra） +- Condition：P1 Admin 重试修正 + E2E infra 就绪后，升级为 PASS + +--- + +## 2. Commit 范围 + +| 项目 | 值 | +|------|-----| +| 基线 | `98e442e` fix(M-AI-02-GATE): restore physical column names | +| HEAD | `5108a9a` fix: queue-definitions.spec.ts 显式导入 | +| 文件 | 50 changed, +6739/-101 lines | +| Migration | **零 M-AI-03 Migration**（最新: `20260620000008`，M-AI-02） | + +--- + +## 3. 运行环境 + +| 组件 | 状态 | +|------|:---:| +| Node | ✅ v22 | +| Prisma validate | ✅ schema valid | +| Prisma generate | ✅ v5.22.0 | +| npm run build | ✅ nest build | +| MySQL (真实) | ⚠️ 未在本机运行（CI 环境需要） | +| Redis (真实) | ⚠️ 未在本机运行 | +| BullMQ (真实) | ⚠️ 未在本机运行 | +| API Process | ✅ app.module.ts 编译通过 | +| Worker Process | ✅ worker.module.ts 编译通过 | +| CI | ⚠️ 未触发最新 CI run | + +--- + +## 4. 验收矩阵 + +| 验收项 | 结论 | 证据 | 风险 | +|--------|:---:|------|------| +| 无 Prisma Migration | ✅ | `prisma/migrations/` 最新为 M-AI-02 | 无 | +| 无真实业务切流 | ✅ | AiAnalysisWorker 未修改 | 无 | +| 无通用公开创建接口 | ✅ | `ai-job.controller.ts:13,19` 仅 GET + POST cancel | 无 | +| 无巨型 switch | ✅ | `JobDefinitionRegistry` Map-based | 无 | +| API 无 Processor | ✅ | `app.module.ts` 零 @Processor | 无 | +| Worker 无 Controller | ✅ | `worker.module.ts` 零 @Controller | 无 | +| Job+Snapshot+Outbox 原子 | ✅ | `ai-job-creation.service.ts:101-150` $transaction | 无 | +| 幂等键生效 | ✅ | P2002 catch | 无 | +| Outbox payload 最小化 | ✅ | `{jobId: job.id}` | 无 | +| CAS 并发安全 | ✅ | `lockJob` + `markProcessing` 均用 updateMany CAS | 无 | +| 崩溃窗口恢复 | ✅ | BullMQ jobId 幂等 + releaseExpiredLocks | 无 | +| 状态机完整 | ✅ | 5 状态 + 双向 Shadow Write | 无 | +| attemptCount 原子递增 | ✅ | `attemptCount: { increment: 1 }` in lockJob | 无 | +| 错误分类 10 类 | ✅ | `classifyError()` 7 retryable/permanent | 无 | +| 取消 3 检查点 | ✅ | 执行前/Provider前/Projector前 | 无 | +| Projector 原子 | ✅ | `$transaction(tx => projector(tx) + markSucceeded(tx))` | 无 | +| public/Admin 响应隔离 | ✅ | `toPublicResponse` 排除 validatedOutput/internalErrorMessage | 无 | +| 用户权限隔离 | ✅ | `job.userId !== userId → 403` | 无 | +| Synthetic 生产不注册 | ✅ | `isSyntheticEnabled()` + `assertNotProduction()` | 无 | +| Admin 重试走 Registry+Outbox | ❌ | `ai-job.service.ts:144-148` 注释自认绕过 | P1 | +| E2E 19/19 通过 | ⚠️ | 18 pending infra | P2 | +| 全量测试 | ⚠️ | 468/488 passed（2 预存失败 + 18 E2E skip） | P2 | + +--- + +## 5. 架构边界审核 + +| 边界 | 预期 | 实际 | 判定 | +|------|------|------|:---:| +| API Processor | 0 | 0 | ✅ | +| Worker Controller | 0 | 0 | ✅ | +| 公开创建 Job 路由 | 无 POST / | 仅 GET :jobId + POST :jobId/cancel | ✅ | +| Provider SDK 直接调用 | 0 | 0（通过 AiGatewayService） | ✅ | +| EventBus/domain-events 改动 | 0 | 0 | ✅ | +| 旧 ai-analysis 队列 | 保留 | 6 旧队列未修改 | ✅ | +| 旧 Worker 删除 | 0 | AiAnalysisWorker 等 5 个保留 | ✅ | +| lifecycleStatus 直接写入 | 仅 LifecycleRepo + 旧路径 | 旧 AiAnalysisRepository 保留，但 Admin retry 绕过 | ⚠️ | + +--- + +## 6. 状态机审核 + +- 状态定义：`ai-job-state-machine.ts:16-22` — queued/running/succeeded/failed/cancelled ✅ +- cancel_requested 为信号而非状态：`ai-job-state-machine.ts:62-63` ✅ +- 合法迁移：`queued→running/cancelled`, `running→succeeded/failed/cancelled` ✅ +- 非法迁移：12 种，终态无出边 ✅ +- CAS 更新：`WHERE lifecycleStatus IN (...)` ✅ +- Shadow Write：双向 `LIFECYCLE_TO_STATUS` + `STATUS_TO_LIFECYCLE`，含 cancelled→failed ✅ +- 直接 Prisma 写入：`ai-job.service.ts:167` Admin retry 绕过 LifecycleRepo ⚠️ + +--- + +## 7. Registry 审核 + +- `JobDefinitionRegistry`：Map-based，`register()` 含 10 类校验 ✅ +- 唯一 jobType：`DuplicateJobTypeError` ✅ +- 未知 jobType：`UnknownJobTypeError` ✅ +- 非法 queueName：`InvalidQueueNameError` ✅ +- Schema 非空：`MissingSchemaError` ✅ +- timeout/attempts 范围：`InvalidTimeoutError`/`InvalidRetriesError` ✅ +- Object.freeze：禁止运行时覆盖 ✅ +- 无巨型 switch：全仓搜索确认 ✅ +- 当前注册清单：`synthetic_job` → `ai-interactive`（仅测试环境）✅ + +--- + +## 8. Job 创建事务审核 + +- 三项在同一 `prisma.$transaction`：`ai-job-creation.service.ts:101-150` ✅ +- 幂等：P2002 catch + `findUniqueOrThrow` ✅ +- Outbox payload 仅 `{ jobId }`：`ai-job-creation.service.ts:145` ✅ +- 元数据从 Definition 写入：14 字段透传 ✅ + +**无法确认（需真实 MySQL）**： +- 真实事务回滚（Job 创建失败 → Snapshot/Outbox 不存在） +- 真实 P2002 并发场景 + +--- + +## 9. Outbox Dispatcher 审核 + +- CAS 先于 queue.add：`outbox-dispatcher.service.ts` ✅ +- 仅运行于 Worker：`worker.module.ts:85`，`app.module.ts` 无 ✅ +- 崩溃窗口：releaseExpiredLocks(60s) + BullMQ jobId 幂等 ✅ +- 未知 eventType：标记永久失败 ✅ + +**无法确认（需真实 MySQL+Redis+BullMQ）**： +- 多实例并发领取不重复 +- 崩溃恢复真实验证 +- 真实入队后崩溃→重启→不重复 + +--- + +## 10. Execution Engine 审核 + +- 5 阶段管线：PREPARE→RESOLVE→EXECUTE→PROJECT→COMPLETE ✅ +- 错误分类：7 类（provider_rate_limited/timeout/unavailable/schema_validation/business_validation/reference_validation/cancelled/internal_error）✅ +- 永久错误直接 markFailed：`engine.ts:232-239` ✅ +- retryable → unlockForRetry + throw：`engine.ts:227-228` ✅ +- attemptCount 原子递增：`lifecycle-repo.ts:147` `{ increment: 1 }` ✅ +- 取消 3 检查点：执行前/Provider前/Projector前 ✅ +- 通过 AiGatewayService 调用（非直接 Provider SDK）✅ + +--- + +## 11. Projector 与 Artifact 审核 + +- 事务内 `projector.project(tx)` + `markSucceeded(tx)`：`projection-executor.service.ts:90-109` ✅ +- 幂等：`SyntheticResultProjector` findMany 前置检查 ✅ +- Artifact UNIQUE：`@@unique([jobId, artifactType, artifactId])` ✅ +- 无 Projector 路径调用 markSucceeded：`projection-executor.service.ts:57,66` ✅ + +--- + +## 12. 取消与重试审核 + +- 用户取消 → `lifecycleRepo.requestCancellation()` ✅ +- queued → `markCancelled`（直接）✅ +- running → `cancelRequestedAt` 信号 ✅ +- 终态幂等 → `isTerminal` check ✅ +- Admin 重试 → **绕过 Registry + CreationService + 非原子** ❌（P1） + +--- + +## 13. API、权限与安全审核 + +- JWT 隔离：`job.userId !== userId → 403` ✅ +- 公开响应不含 validatedOutput/internalErrorMessage/snapshot/credential：`toPublicResponse:237` ✅ +- Admin 响应包含 internalErrorMessage/validatedOutput：`toAdminResponse:266-267` ✅ +- 无明文密钥/Token/密码在代码中：全仓扫描确认 ✅ + +--- + +## 14. Synthetic E2E 审核 + +- 环境门控：`NODE_ENV=test && AI_JOB_SYNTHETIC_ENABLED=true` ✅ +- 生产拒绝：`assertNotProduction()` ✅ +- API 层测试：JWT 保护验证（401 without token）✅ +- 完整链路 E2E：**18/19 pending infra** ⚠️ + +--- + +## 15. M-AI-01/M-AI-02 回归 + +- 全量单元测试：468/488 passed +- 2 个预存失败：`queue-definitions.spec.ts`（断言假定统一配置，但 ADR-003 引入 per-queue 差异） +- M-AI-01 Worker Integration：代码路径未修改（AiAnalysisWorker 保留） +- M-AI-02 backward compatibility：Migration 文件未修改，schema 仅加注释 + +--- + +## 16. CI 与部署门禁 + +- deploy.yml：包含 build-and-unit + current-integration + backward-compat + deploy +- M-AI-03 Synthetic E2E：测试文件存在（`test/m-ai-03-synthetic.e2e-spec.ts`），但 Workflow 未配置专门的 M-AI-03 E2E Job +- CI 阻断 deploy：`deploy` job 依赖 `current-integration` + `backward-compat`，但未依赖 M-AI-03 Synthetic E2E + +**无法确认**：最新 CI run 结果（本地无 CI runner） + +--- + +## 17. 问题列表 + +### P1：不得进入 M-AI-04 + +| ID | 文件 | 行 | 描述 | +|----|------|----|------| +| P1-01 | `ai-job.service.ts` | 144-148 | Admin retry 绕过 Registry（不校验 jobType）、绕过 CreationService（非原子 Job+Snapshot+Outbox）、代码自注"简化实现" | + +### P2：可修复后复审 + +| ID | 文件 | 描述 | +|----|------|------| +| P2-01 | `queue-definitions.spec.ts:62,195` | 2 tests 断言 concurrency=1/lockDuration=30s（统一配置），与 ADR-003 per-queue 差异冲突 | +| P2-02 | `test/m-ai-03-synthetic.e2e-spec.ts` | 18/19 Synthetic E2E 场景 pending（需真实 MySQL/Redis/BullMQ/Mock AI Server） | + +### P3：非阻断债务 + +| ID | 描述 | +|----|------| +| P3-01 | `queue-definitions.spec.ts` 显式导入 `QUEUE_AI_INTERACTIVE`/`QUEUE_AI_BACKGROUND` 但未更新断言（commit `5108a9a` 修复了 import，断言待更新） | + +--- + +## 18. 无法确认项 + +| 项目 | 缺失环境 | 是否阻断 PASS | +|------|---------|:---:| +| 真实 MySQL 事务测试 | 本地 MySQL | 否（已有单元测试覆盖逻辑） | +| 真实 Redis/BullMQ E2E | 本地 Redis + BullMQ | 否（已有单元测试覆盖逻辑） | +| Outbox 多实例并发 | 多进程环境 | 否（CAS 逻辑已验证） | +| Dispatcher 崩溃恢复 | 进程管理 | 否（releaseExpiredLocks 逻辑已验证） | +| CI Run 最新结果 | CI Runner | 否（Workflow YAML 已审查） | +| Admin retry 是否已绕过缺陷引入生产数据 | 需真实调用 | **是**（P1-01 必须在进入 M-AI-04 前修正） | + +--- + +## 19. 下一步建议 + +1. **修正 P1-01**（Admin retry 接入 Registry + CreationService），创建 Fix Issue +2. **修正 P2-01**（queue-definitions.spec.ts 更新断言），可随 P1-01 一同提交 +3. **CI 环境就绪后**：激活 `test/m-ai-03-synthetic.e2e-spec.ts` 中的 18 个 pending 场景，执行全部 19 场景 +4. **Condition 解除后**：升级为 PASS，允许进入 M-AI-04（Active Recall 端到端迁移） diff --git a/src/infrastructure/queue/queue-definitions.spec.ts b/src/infrastructure/queue/queue-definitions.spec.ts index 7ad2f6b..8298281 100644 --- a/src/infrastructure/queue/queue-definitions.spec.ts +++ b/src/infrastructure/queue/queue-definitions.spec.ts @@ -57,8 +57,13 @@ describe('Queue Definition Registry', () => { } }); - it('applies first-stage default values', () => { - for (const def of Object.values(QUEUE_DEFINITIONS)) { + it('applies first-stage default values (legacy 6 queues)', () => { + const legacyQueues = [ + QUEUE_AI_ANALYSIS, QUEUE_DOCUMENT_IMPORT, QUEUE_NOTIFICATION, + QUEUE_DOMAIN_EVENTS, QUEUE_AUDIT_LOG, QUEUE_FILE_CLEANUP, + ]; + for (const name of legacyQueues) { + const def = QUEUE_DEFINITIONS[name]; expect(def.workerOptions.concurrency).toBe(1); expect(def.defaultJobOptions.attempts).toBe(3); expect(def.defaultJobOptions.backoff.delay).toBe(1000); @@ -69,6 +74,21 @@ describe('Queue Definition Registry', () => { expect(def.defaultJobOptions.removeOnFail.age).toBe(7 * 24 * 3600); } }); + + it('M-AI-03 queues have ADR-003 per-queue config', () => { + // ai-interactive: concurrency=2, attempts=2, backoff=2000ms, lockDuration=30s + const iDef = QUEUE_DEFINITIONS[QUEUE_AI_INTERACTIVE]; + expect(iDef.workerOptions.concurrency).toBe(2); + expect(iDef.defaultJobOptions.attempts).toBe(2); + expect(iDef.defaultJobOptions.backoff.delay).toBe(2000); + + // ai-background: concurrency=1, attempts=3, backoff=5000ms, lockDuration=60s + const bDef = QUEUE_DEFINITIONS[QUEUE_AI_BACKGROUND]; + expect(bDef.workerOptions.concurrency).toBe(1); + expect(bDef.defaultJobOptions.attempts).toBe(3); + expect(bDef.defaultJobOptions.backoff.delay).toBe(5000); + expect(bDef.workerOptions.lockDuration).toBe(60_000); + }); }); describe('queueDef', () => { @@ -184,18 +204,18 @@ describe('Queue Definition Registry', () => { }); describe('lockDuration vs stalled vs business timeout', () => { - it('lockDuration (30s) ≠ business timeout — documented constraint', () => { + it('lockDuration ≠ business timeout — documented constraint', () => { // lockDuration is BullMQ lock renewal interval. - // If the processor runs past lockDuration but the worker is alive, - // BullMQ auto-extends the lock. Only if the worker crashes does - // stalledInterval (30s) fire, and maxStalledCount (1) controls - // how many times the job can be retried by another worker. // Business timeouts MUST be enforced inside processor logic. + // Each queue may have different lockDuration (e.g. ai-background=60s). for (const def of Object.values(QUEUE_DEFINITIONS)) { - expect(def.workerOptions.lockDuration).toBe(30_000); - expect(def.workerOptions.stalledInterval).toBe(30_000); - expect(def.workerOptions.maxStalledCount).toBe(1); + expect(def.workerOptions.lockDuration).toBeGreaterThanOrEqual(30_000); + expect(def.workerOptions.stalledInterval).toBeGreaterThanOrEqual(30_000); + expect(def.workerOptions.maxStalledCount).toBeGreaterThanOrEqual(1); } + // ai-background explicitly configured for longer jobs + expect(QUEUE_DEFINITIONS[QUEUE_AI_BACKGROUND].workerOptions.lockDuration).toBe(60_000); + expect(QUEUE_DEFINITIONS[QUEUE_AI_BACKGROUND].workerOptions.stalledInterval).toBe(60_000); }); }); }); diff --git a/src/modules/ai-job/ai-job-creation.service.ts b/src/modules/ai-job/ai-job-creation.service.ts index 3478e5c..34df2c8 100644 --- a/src/modules/ai-job/ai-job-creation.service.ts +++ b/src/modules/ai-job/ai-job-creation.service.ts @@ -31,10 +31,12 @@ export interface CreateAiJobInput { triggerType: 'user_api' | 'admin_manual' | 'system_scheduled' | 'parent_job'; targetType: string; targetId: string; - /** 父 Job ID（Job 链场景） */ + /** 父 Job ID（Job 链/Admin Retry 场景） */ parentJobId?: string; /** 幂等 Key。提供时启用幂等；不提供时每次创建新 Job */ idempotencyKey?: string; + /** Admin Retry：预构建的 Snapshot content（复用原 Snapshot 时传入） */ + retrySnapshotContent?: Record; /** 覆盖 Definition 默认值 */ overrides?: { priority?: number; @@ -70,11 +72,14 @@ export class AiJobCreationService { } // 3. Build snapshot（事务外：依赖外部 DB 读取，不应占用事务时间） - const snapshot = await this.snapshotBuilder.buildSnapshot( - input.userId, - input.targetType, - input.targetId, - ); + // Admin Retry 可传入预构建的 snapshot content（复用原 Snapshot 内容） + const snapshot = input.retrySnapshotContent + ? input.retrySnapshotContent + : await this.snapshotBuilder.buildSnapshot( + input.userId, + input.targetType, + input.targetId, + ); const contentHash = this.computeHash(JSON.stringify(snapshot)); diff --git a/src/modules/ai-job/ai-job.service.ts b/src/modules/ai-job/ai-job.service.ts index a54194d..fb735d6 100644 --- a/src/modules/ai-job/ai-job.service.ts +++ b/src/modules/ai-job/ai-job.service.ts @@ -2,6 +2,8 @@ import { Injectable, NotFoundException, ForbiddenException, ConflictException } import { PrismaService } from '../../infrastructure/database/prisma.service'; import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; import { AiJobStateMachine, LifecycleStatus, TERMINAL_STATUSES } from './ai-job-state-machine'; +import { AiJobCreationService } from './ai-job-creation.service'; +import { JobDefinitionRegistry } from './job-definition-registry'; import { JobNotCancellableError } from './ai-job.errors'; @Injectable() @@ -10,6 +12,8 @@ export class AiJobService { private readonly prisma: PrismaService, private readonly lifecycleRepo: AiJobLifecycleRepository, private readonly stateMachine: AiJobStateMachine, + private readonly creationService: AiJobCreationService, + private readonly registry: JobDefinitionRegistry, ) {} // ── 用户查询 ── @@ -134,73 +138,59 @@ export class AiJobService { } async retryJob(adminUserId: string, jobId: string) { + // 1. Load original Job const original = await this.prisma.aiJob.findUnique({ where: { id: jobId }, include: { snapshot: true }, }); if (!original) throw new NotFoundException('Original job not found'); - // 创建新 Job，parentJobId 指向原 Job - // 注意：这里不经过 Registry/Outbox/CreationService 完整链路， - // 而是直接创建（Admin 手动重试的特殊路径） - // Issue 要求"不得绕过 Registry 和 Outbox"——用 AiJobCreationService - // 但此处为简化实现，先直接创建，后续可改为调用 CreationService - const newJob = await this.prisma.aiJob.create({ - data: { - userId: original.userId, - jobType: original.jobType, - triggerType: 'admin_manual', - queueName: original.queueName, - targetType: original.targetType, - targetId: original.targetId, - parentJobId: original.id, - priority: original.priority, - credentialMode: original.credentialMode, - credentialId: original.credentialId, - modelProvider: original.modelProvider, - modelName: original.modelName, - promptVersion: original.promptVersion, - outputSchemaVersion: original.outputSchemaVersion, - inputSchemaVersion: original.inputSchemaVersion, - maxAttempts: original.maxAttempts, - timeoutMs: original.timeoutMs, - lifecycleStatus: 'queued', - status: 'pending', - queuedAt: new Date(), - attemptCount: 0, - }, - }); + // 2. Validate jobType via Registry + this.registry.get(original.jobType); // throws UnknownJobTypeError if not registered - // 复制 Snapshot - if (original.snapshot) { - await this.prisma.aiJobSnapshot.create({ - data: { - jobId: newJob.id, - jobType: newJob.jobType, - schemaVersion: original.snapshot.schemaVersion, - redactionVersion: original.snapshot.redactionVersion, - content: original.snapshot.content as any, - contentHash: original.snapshot.contentHash, - }, - }); + // 3. Validate retryable state (only terminal jobs can be retried) + const status = this.stateMachine.parse(original.lifecycleStatus); + if (!this.stateMachine.isTerminal(status)) { + throw new ConflictException( + `Cannot retry job in "${status}" status. Only terminal jobs (succeeded/failed/cancelled) can be retried.`, + ); } - // 创建 OutboxEvent - await this.prisma.outboxEvent.create({ - data: { - eventType: 'ai.job.enqueue', - aggregateType: 'AiJob', - aggregateId: newJob.id, - dedupeKey: `ai.job.enqueue:${newJob.id}`, - payload: { jobId: newJob.id } as any, - status: 'pending', - availableAt: new Date(), + // 4. Minimal validation: original must have targetType + targetId + if (!original.targetType || !original.targetId) { + throw new ConflictException('Original job missing targetType/targetId — cannot retry'); + } + + // 5. Idempotency key: allows same admin operation to return same retry job, + // but different retry operations produce different jobs + const retryIdempotencyKey = `admin-retry:${jobId}:${Date.now()}`; + + // 6. Delegate to AiJobCreationService (Registry + Snapshot + Outbox + atomic transaction) + const newJob = await this.creationService.createJob({ + userId: original.userId, + jobType: original.jobType, + triggerType: 'admin_manual', + targetType: original.targetType!, + targetId: original.targetId!, + parentJobId: original.id, + idempotencyKey: retryIdempotencyKey, + retrySnapshotContent: original.snapshot?.content as Record | undefined, + overrides: { + credentialMode: original.credentialMode as 'platform_key' | 'user_deepseek_key', + credentialId: original.credentialId ?? undefined, }, }); + this.logger.log( + `Admin retry: original=${jobId} → new=${newJob.id} ` + + `type=${newJob.jobType} parentJobId=${original.id}`, + ); + return this.toAdminResponse(newJob); } + private logger = new (require('@nestjs/common').Logger)(AiJobService.name); + // ── 可观测性 ── async getOutboxStatus() { diff --git a/test/jest-m-ai-03.json b/test/jest-m-ai-03.json new file mode 100644 index 0000000..d3e0f58 --- /dev/null +++ b/test/jest-m-ai-03.json @@ -0,0 +1,13 @@ +{ + "moduleFileExtensions": ["js", "json", "ts"], + "rootDir": "..", + "testEnvironment": "node", + "testRegex": "m-ai-03-synthetic.e2e-spec.ts$", + "transform": { + "^.+\\.(t|j)sx?$": ["ts-jest", { "useESM": false, "tsconfig": "tsconfig.json" }] + }, + "transformIgnorePatterns": ["/node_modules/"], + "moduleNameMapper": { + "^jose$": "/test/mocks/jose.mock.ts" + } +} diff --git a/test/m-ai-03-synthetic.e2e-spec.ts b/test/m-ai-03-synthetic.e2e-spec.ts index ce61278..2dad858 100644 --- a/test/m-ai-03-synthetic.e2e-spec.ts +++ b/test/m-ai-03-synthetic.e2e-spec.ts @@ -3,18 +3,29 @@ import { INestApplication } from '@nestjs/common'; import { JwtService } from '@nestjs/jwt'; import request from 'supertest'; import { AppModule } from '../src/app.module'; +import { AiJobCreationService } from '../src/modules/ai-job/ai-job-creation.service'; +import { JobDefinitionRegistry } from '../src/modules/ai-job/job-definition-registry'; +import { AiJobLifecycleRepository } from '../src/modules/ai-job/ai-job-lifecycle.repository'; +import { AiJobStateMachine } from '../src/modules/ai-job/ai-job-state-machine'; /** - * M-AI-03 Synthetic E2E — API 层验证 + * M-AI-03 Synthetic E2E — 真实基础设施 * - * 验证 API 路由注册、JWT 鉴权、环境门控。 - * 数据层集成场景需真实 MySQL/Redis/BullMQ 基础设施（见下方清单）。 + * 需要: MySQL + Redis + BullMQ + Mock AI HTTP Server + * 环境变量: NODE_ENV=test, AI_JOB_SYNTHETIC_ENABLED=true */ -describe('M-AI-03 Synthetic E2E', () => { +const userId = 'synthetic-e2e-user'; +const OLD_ENV = { ...process.env }; + +describe('M-AI-03 Synthetic E2E (real infra)', () => { let app: INestApplication; + let creationService: AiJobCreationService; + let registry: JobDefinitionRegistry; + let lifecycleRepo: AiJobLifecycleRepository; + let sm: AiJobStateMachine; let jwtService: JwtService; - const OLD_ENV = { ...process.env }; + let userToken: string; beforeAll(async () => { process.env.NODE_ENV = 'test'; @@ -28,49 +39,159 @@ describe('M-AI-03 Synthetic E2E', () => { app = module.createNestApplication(); app.setGlobalPrefix('api', { exclude: ['admin-api/(.*)', 'internal/(.*)'] }); await app.init(); + + creationService = module.get(AiJobCreationService); + registry = module.get(JobDefinitionRegistry); + lifecycleRepo = module.get(AiJobLifecycleRepository); + sm = module.get(AiJobStateMachine); jwtService = module.get(JwtService); - }); + + userToken = jwtService.sign({ + sub: userId, email: 'e2e@test.com', role: 'USER', type: 'user', + }); + }, 30000); afterAll(async () => { process.env = OLD_ENV; await app.close(); }); - // ── 通过的场景（mock infra 兼容）── + // ═══════════════ 场景 1-2: 创建成功 ═══════════════ + it('1. Synthetic Definition 已注册', () => { + expect(registry.has('synthetic_job')).toBe(true); + }); - it('场景: GET /api/ai/jobs/:id → 401 无 token（JWT 保护）', async () => { + it('2. AiJobCreationService 创建 synthetic_job → status=queued', async () => { + const job = await creationService.createJob({ + userId, jobType: 'synthetic_job', triggerType: 'user_api', + targetType: 'synthetic', targetId: 'test-1', + }); + expect(job).toBeDefined(); + expect(job.jobType).toBe('synthetic_job'); + expect(job.lifecycleStatus).toBe('queued'); + // 验证元数据来自 Definition + expect(job.queueName).toBe('ai-interactive'); + }); + + // ═══════════════ 场景 3: 幂等创建 ═══════════════ + it('3. 相同 idempotencyKey → 返回同一 Job', async () => { + const idemKey = `e2e-idem-${Date.now()}`; + const j1 = await creationService.createJob({ + userId, jobType: 'synthetic_job', triggerType: 'user_api', + targetType: 'synthetic', targetId: 'test-3', + idempotencyKey: idemKey, + }); + const j2 = await creationService.createJob({ + userId, jobType: 'synthetic_job', triggerType: 'user_api', + targetType: 'synthetic', targetId: 'test-3', + idempotencyKey: idemKey, + }); + expect(j2.id).toBe(j1.id); + }); + + // ═══════════════ 场景 4: 原子创建 ═══════════════ + it('4. Job + Snapshot + Outbox 原子创建', async () => { + const { PrismaClient } = require('@prisma/client'); + const prisma = new PrismaClient(); + const job = await creationService.createJob({ + userId, jobType: 'synthetic_job', triggerType: 'user_api', + targetType: 'synthetic', targetId: 'test-4', + idempotencyKey: `e2e-atomic-${Date.now()}`, + }); + // Verify snapshot exists + const snap = await prisma.aiJobSnapshot.findUnique({ where: { jobId: job.id } }); + expect(snap).toBeDefined(); + expect(snap.contentHash).toBeTruthy(); + // Verify outbox exists + const outbox = await prisma.outboxEvent.findFirst({ + where: { aggregateId: job.id }, + }); + expect(outbox).toBeDefined(); + expect(outbox.eventType).toBe('ai.job.enqueue'); + await prisma.$disconnect(); + }); + + // ═══════════════ 场景 13-14: Cancel ═══════════════ + it('5. queued Job → cancel → cancelled', async () => { + const job = await creationService.createJob({ + userId, jobType: 'synthetic_job', triggerType: 'user_api', + targetType: 'synthetic', targetId: 'test-5', + }); + const res = await request(app.getHttpServer()) + .post(`/api/ai/jobs/${job.id}/cancel`) + .set('Authorization', `Bearer ${userToken}`) + .expect(200); + expect(res.body.status).toBe('cancelled'); + }); + + // ═══════════════ API 层验证 ═══════════════ + it('6. GET /api/ai/jobs/:id — JWT 保护 401', async () => { await request(app.getHttpServer()).get('/api/ai/jobs/any').expect(401); }); - it('场景: POST /api/ai/jobs/:id/cancel → 401 无 token（JWT 保护）', async () => { - await request(app.getHttpServer()).post('/api/ai/jobs/any/cancel').expect(401); + it('7. GET /api/ai/jobs/:id — 用户隔离', async () => { + const otherToken = jwtService.sign({ + sub: 'other-user', email: 'o@t.com', role: 'USER', type: 'user', + }); + const job = await creationService.createJob({ + userId, jobType: 'synthetic_job', triggerType: 'user_api', + targetType: 'synthetic', targetId: 'test-7', + }); + // Own job → 200 + await request(app.getHttpServer()) + .get(`/api/ai/jobs/${job.id}`) + .set('Authorization', `Bearer ${userToken}`) + .expect(200); + // Other user → 403 + await request(app.getHttpServer()) + .get(`/api/ai/jobs/${job.id}`) + .set('Authorization', `Bearer ${otherToken}`) + .expect(403); }); - it('场景: AiJobController 仅有 GET + POST cancel（代码审查确认），无公开创建路由', () => { - // 代码证据: ai-job.controller.ts — 仅有 @Get(':jobId') + @Post(':jobId/cancel') - // AiJobCreationService.createJob() 为 internal-only，不暴露 HTTP - expect(true).toBe(true); + it('8. 公开响应不含敏感字段', async () => { + const job = await creationService.createJob({ + userId, jobType: 'synthetic_job', triggerType: 'user_api', + targetType: 'synthetic', targetId: 'test-8', + }); + const res = await request(app.getHttpServer()) + .get(`/api/ai/jobs/${job.id}`) + .set('Authorization', `Bearer ${userToken}`) + .expect(200); + expect(res.body).not.toHaveProperty('validatedOutput'); + expect(res.body).not.toHaveProperty('internalErrorMessage'); + expect(res.body).not.toHaveProperty('snapshot'); }); - it('场景 #19: NODE_ENV=test + AI_JOB_SYNTHETIC_ENABLED=true → 启动成功', () => { + // ═══════════════ 场景 16: 未知 JobType ═══════════════ + it('9. 未知 jobType → UnknownJobTypeError', async () => { + await expect( + creationService.createJob({ + userId, jobType: 'nonexistent_type', triggerType: 'user_api', + targetType: 'x', targetId: 'x', + }), + ).rejects.toThrow(); + }); + + // ═══════════════ 场景 19: 生产环境保护 ═══════════════ + it('10. NODE_ENV=test + AI_JOB_SYNTHETIC_ENABLED=true → 模块正常', () => { expect(process.env.NODE_ENV).toBe('test'); expect(process.env.AI_JOB_SYNTHETIC_ENABLED).toBe('true'); }); - // ── 需要真实基础设施的场景（已由单元测试覆盖逻辑）── - // 1. interactive 成功 → need real DB + BullMQ + AI mock - // 2. background 成功 → need real DB + BullMQ + AI mock - // 3. 幂等创建 → unit test: P2002 catch in ai-job-creation.service.spec - // 4. 重复 Outbox 领取 → unit test: CAS skipped in outbox-dispatcher.service.spec - // 5. 入队后 Dispatcher 崩溃 → need real BullMQ + Redis - // 6. Worker SIGKILL → need OS-level test - // 7. Projector/ACK crash → need real DB transaction - // 8-10. Provider 429/500/timeout → unit test: classifyError in engine spec - // 11-12. 非法JSON/Schema失败 → unit test: permanent error paths - // 13-14. queued/running cancel → unit test: requestCancellation paths - // 15. Admin retry → need real DB - // 16. 未知 JobType → unit test: UnknownJobTypeError - // 17. 非法 Queue → unit test: InvalidQueueNameError - // 18. Redis Payload 敏感信息扫描 → need real Redis - // 19. Synthetic 生产环境不注册 → ✅ this test + // ═══════════════ 状态机验证 ═══════════════ + it('11. 合法状态迁移全部通过', () => { + expect(() => sm.validate('queued', 'running')).not.toThrow(); + expect(() => sm.validate('queued', 'cancelled')).not.toThrow(); + expect(() => sm.validate('running', 'succeeded')).not.toThrow(); + expect(() => sm.validate('running', 'failed')).not.toThrow(); + expect(() => sm.validate('running', 'cancelled')).not.toThrow(); + }); + + it('12. succeeded/failed/cancelled 是终态', () => { + expect(sm.isTerminal('succeeded')).toBe(true); + expect(sm.isTerminal('failed')).toBe(true); + expect(sm.isTerminal('cancelled')).toBe(true); + expect(sm.isTerminal('queued')).toBe(false); + }); });