api-server/src/modules/ai-job/ai-job-execution-engine.spec.ts

import { Test, TestingModule } from '@nestjs/testing';
import { AiJobExecutionEngineImpl } from './ai-job-execution-engine';
import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository';
import { JobDefinitionRegistry } from './job-definition-registry';
import { AiJobStateMachine } from './ai-job-state-machine';
import { AiGatewayService } from '../ai/gateway/ai-gateway.service';
import { ProjectionExecutor } from './projection-executor.service';
import { ActiveRecallExecutor } from './active-recall-executor';
import { ActiveRecallObservabilityService } from './active-recall-observability.service';
import { FeynmanExecutor } from './feynman-executor';
import { FeynmanBusinessValidator, FeynmanReferenceValidator } from './feynman-validator';
import { FeynmanObservabilityService } from './feynman-observability.service';
import { ReviewCardGenerationExecutor } from './review-card-generation-executor';
import { ReviewCardGenerationValidator } from './review-card-generation-validator';
import { PrismaService } from '../../infrastructure/database/prisma.service';
import { JobLockConflictError, JobAlreadyTerminalError } from './ai-job.errors';

function validDef() {
  return {
    jobType: 'test_job',
    queue: { queueName: 'ai-interactive', defaultPriority: 0 },
    execution: { timeoutMs: 5000, maxRetries: 2, retryBackoff: { type: 'exponential', delay: 2000 }, cancellable: true, abortStrategy: 'fail' },
    input: { schemaVersion: '1.0' },
    output: { schemaVersion: '1.0' },
    prompt: { promptKey: 'test', promptVersion: '1.0' },
    model: { modelTier: 'primary', modelProvider: 'deepseek', modelName: 'deepseek-chat' },
    credential: { allowedModes: ['platform_key'], defaultMode: 'platform_key' },
    security: { contentSafetyCheck: false, outputRedaction: false },
    metadata: { label: 'Test', description: '', domain: 'analysis', version: '1.0' },
  };
}

describe('AiJobExecutionEngineImpl', () => {
  let engine: AiJobExecutionEngineImpl;
  let prisma: any;
  let lifecycleRepo: any;
  let registry: any;
  let stateMachine: any;
  let aiGateway: any;
  let projectionExecutor: any;

  function makeJob(overrides?: any) {
    return {
      id: 'job-001',
      userId: 'user-1',
      jobType: 'test_job',
      lifecycleStatus: 'queued',
      queueName: 'ai-interactive',
      attemptCount: 0,
      cancelRequestedAt: null,
      ...overrides,
    };
  }

  beforeEach(async () => {
    lifecycleRepo = {
      lockJob: jest.fn(),
      markSucceeded: jest.fn(),
      markFailed: jest.fn(),
      markCancelled: jest.fn(),
    };
    registry = { get: jest.fn().mockReturnValue(validDef()) };
    stateMachine = new AiJobStateMachine();
    aiGateway = { generate: jest.fn() };
    projectionExecutor = { execute: jest.fn().mockResolvedValue([]) };

    prisma = {
      aiJob: {
        findUnique: jest.fn(),
        update: jest.fn(),
      },
      aiJobSnapshot: { findUnique: jest.fn() },
      aiUsageLog: { create: jest.fn() },
    };

    jest.spyOn(require('@nestjs/common').Logger.prototype, 'log').mockImplementation(() => {});
    jest.spyOn(require('@nestjs/common').Logger.prototype, 'warn').mockImplementation(() => {});
    jest.spyOn(require('@nestjs/common').Logger.prototype, 'error').mockImplementation(() => {});

    const module: TestingModule = await Test.createTestingModule({
      providers: [
        AiJobExecutionEngineImpl,
        { provide: PrismaService, useValue: prisma },
        { provide: AiJobLifecycleRepository, useValue: lifecycleRepo },
        { provide: JobDefinitionRegistry, useValue: registry },
        { provide: AiJobStateMachine, useValue: stateMachine },
        { provide: AiGatewayService, useValue: aiGateway },
        { provide: ProjectionExecutor, useValue: projectionExecutor },
        { provide: ActiveRecallExecutor, useValue: { execute: jest.fn() } },
        { provide: FeynmanExecutor, useValue: { execute: jest.fn() } },
        { provide: FeynmanBusinessValidator, useValue: { validate: jest.fn() } },
        { provide: FeynmanReferenceValidator, useValue: { validate: jest.fn() } },
        { provide: ReviewCardGenerationExecutor, useValue: { execute: jest.fn() } },
        { provide: ReviewCardGenerationValidator, useValue: { validate: jest.fn() } },
        { provide: ActiveRecallObservabilityService, useValue: {
          incrementUnifiedExecuteSuccess: jest.fn(),
          incrementUnifiedExecuteFailed: jest.fn(),
          incrementUnifiedRetry: jest.fn(),
          incrementProjectorFailed: jest.fn(),
          logExecutionCompleted: jest.fn(),
          logExecutionFailed: jest.fn(),
          logRollback: jest.fn(),
        } },
        { provide: FeynmanObservabilityService, useValue: {
          incrementUnifiedExecuteSuccess: jest.fn(),
          incrementUnifiedExecuteFailed: jest.fn(),
          incrementUnifiedRetry: jest.fn(),
          incrementProjectorFailed: jest.fn(),
          addFocusItemCreated: jest.fn(),
          addReviewCardCreated: jest.fn(),
          logExecutionCompleted: jest.fn(),
          logExecutionFailed: jest.fn(),
          logRollback: jest.fn(),
        } },
      ],
    }).compile();

    engine = module.get(AiJobExecutionEngineImpl);
  });

  const ctx = {
    jobId: 'bull-001',
    attemptMade: 1,
    updateProgress: jest.fn(),
  };

  describe('正常执行全链路', () => {
    it('PREPARE→RESOLVE→EXECUTE→PROJECT→COMPLETE 全部成功', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob());
      lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' }));
      prisma.aiJobSnapshot.findUnique.mockResolvedValue({
        schemaVersion: '1.0',
        content: { snapshotData: true },
      });
      aiGateway.generate.mockResolvedValue({
        parsed: { result: 'ok' },
        usage: { provider: 'deepseek', model: 'deepseek-chat', inputTokens: 100, outputTokens: 50, estimatedCost: 0.001, latencyMs: 800 },
      });

      await engine.execute('job-001', ctx);

      // PREPARE: lockJob called
      expect(lifecycleRepo.lockJob).toHaveBeenCalledWith('job-001', 'bull-001');
      // EXECUTE: AiGateway called
      expect(aiGateway.generate).toHaveBeenCalled();
      // PROJECT: validatedOutput + outputHash written
      expect(prisma.aiJob.update).toHaveBeenCalledWith(
        expect.objectContaining({
          where: { id: 'job-001' },
          data: expect.objectContaining({
            validatedOutput: { result: 'ok' },
            outputHash: expect.any(String),
          }),
        }),
      );
      // PROJECT: ProjectionExecutor called
      expect(projectionExecutor.execute).toHaveBeenCalledWith(
        undefined, // no projectorKey in validDef
        expect.objectContaining({
          job: expect.objectContaining({ id: 'job-001' }),
          validatedOutput: { result: 'ok' },
        }),
      );
      // COMPLETE: usage log
      expect(prisma.aiUsageLog.create).toHaveBeenCalled();
    });
  });

  describe('终态跳过', () => {
    it('Job 已 succeeded → 跳过执行', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob({ lifecycleStatus: 'succeeded' }));

      await engine.execute('job-001', ctx);
      expect(lifecycleRepo.lockJob).not.toHaveBeenCalled();
      expect(aiGateway.generate).not.toHaveBeenCalled();
    });
  });

  describe('取消检查', () => {
    it('queued + cancelRequestedAt → 直接标记 cancelled', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob({ cancelRequestedAt: new Date() }));

      await engine.execute('job-001', ctx);
      expect(lifecycleRepo.markCancelled).toHaveBeenCalledWith('job-001');
    });

    it('执行中被取消 → cancelled，不创建 Artifact', async () => {
      prisma.aiJob.findUnique
        .mockResolvedValueOnce(makeJob()) // initial read
        .mockResolvedValueOnce(makeJob({ cancelRequestedAt: new Date() })); // after lockJob, check
      lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' }));

      await engine.execute('job-001', ctx);
      expect(lifecycleRepo.markCancelled).toHaveBeenCalled();
      // Provider 不应被调用
      expect(aiGateway.generate).not.toHaveBeenCalled();
    });
  });

  describe('错误分类', () => {
    it('429 → retryable → 抛给 BullMQ', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob());
      lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' }));
      prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} });

      aiGateway.generate.mockRejectedValue(new Error('429 rate limited'));

      await expect(engine.execute('job-001', ctx)).rejects.toThrow('429');
      // 不调用 markFailed（让 BullMQ 重试）
      expect(lifecycleRepo.markFailed).not.toHaveBeenCalled();
    });

    it('schema validation → permanent → markFailed（不抛错，BullMQ 不再重试）', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob());
      lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running', attemptCount: 1 }));
      prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} });

      aiGateway.generate.mockRejectedValue(new Error('schema validation failed'));

      // 永久错误：不应抛出（避免 BullMQ retry）
      await engine.execute('job-001', ctx);
      expect(lifecycleRepo.markFailed).toHaveBeenCalledWith(
        'job-001',
        expect.objectContaining({ errorCode: 'schema_validation_failed' }),
      );
    });

    it('timeout → retryable → 抛给 BullMQ', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob());
      lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' }));
      prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} });

      aiGateway.generate.mockRejectedValue(new Error('timeout ETIMEDOUT'));

      await expect(engine.execute('job-001', ctx)).rejects.toThrow();
      expect(lifecycleRepo.markFailed).not.toHaveBeenCalled();
    });
  });

  describe('抢锁冲突', () => {
    it('已被其他 Worker 抢走 → 静默退出', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob());
      lifecycleRepo.lockJob.mockRejectedValue(new JobLockConflictError('job-001'));

      await engine.execute('job-001', ctx);
      // 不应抛错，静默退出
      expect(aiGateway.generate).not.toHaveBeenCalled();
    });
  });

  describe('Snapshot schema 不兼容', () => {
    it('schemaVersion 不匹配 → markFailed', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob());
      lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' }));
      prisma.aiJobSnapshot.findUnique.mockResolvedValue({
        schemaVersion: '99.0', // incompatible
        content: {},
      });

      await engine.execute('job-001', ctx);
      expect(lifecycleRepo.markFailed).toHaveBeenCalledWith(
        'job-001',
        expect.objectContaining({ errorCode: 'schema_validation_failed' }),
      );
      expect(aiGateway.generate).not.toHaveBeenCalled();
    });
  });

  describe('Usage Logging', () => {
    it('每次 Provider 调用写 AiUsageLog', async () => {
      prisma.aiJob.findUnique.mockResolvedValue(makeJob());
      lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running', attemptCount: 0 }));
      prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} });
      aiGateway.generate.mockResolvedValue({
        parsed: { ok: true },
        usage: { provider: 'deepseek', model: 'deepseek-chat', inputTokens: 50, outputTokens: 20, estimatedCost: 0.0005, latencyMs: 300 },
      });

      await engine.execute('job-001', ctx);

      expect(prisma.aiUsageLog.create).toHaveBeenCalledWith(
        expect.objectContaining({
          data: expect.objectContaining({
            userId: 'user-1',
            jobId: 'job-001',
            provider: 'deepseek',
            model: 'deepseek-chat',
            inputTokens: 50,
            outputTokens: 20,
          }),
        }),
      );
    });
  });
});