diff --git a/src/modules/ai-job/ai-job-execution-engine.ts b/src/modules/ai-job/ai-job-execution-engine.ts index 6b5d1c4..e2a13c0 100644 --- a/src/modules/ai-job/ai-job-execution-engine.ts +++ b/src/modules/ai-job/ai-job-execution-engine.ts @@ -1,4 +1,5 @@ import { Injectable, Logger } from '@nestjs/common'; +import * as crypto from 'crypto'; import { AiGatewayService } from '../ai/gateway/ai-gateway.service'; import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; import { JobDefinitionRegistry } from './job-definition-registry'; @@ -142,26 +143,23 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { snapshot = snap.content; } - // 7. Provider timeout via AbortController - const controller = new AbortController(); + // 7. 取消检查(Provider 调用前) + const currentJob = await this.prisma.aiJob.findUnique({ + where: { id: aiJobId }, + select: { cancelRequestedAt: true }, + }); + if (currentJob?.cancelRequestedAt) { + await this.lifecycleRepo.markCancelled(aiJobId); + return; + } + + await context.updateProgress(30); + + // ── EXECUTE ── + // 8. 调用 AiGatewayService(不直接导入 Provider SDK) + // timeout 通过 timeoutMs 参数委托给 AiGatewayService 内部的 AbortController const timeoutMs = def.execution.timeoutMs || 30000; - const timeoutId = setTimeout(() => controller.abort(), timeoutMs); - try { - // 8. 取消检查(Provider 调用前) - const currentJob = await this.prisma.aiJob.findUnique({ - where: { id: aiJobId }, - select: { cancelRequestedAt: true }, - }); - if (currentJob?.cancelRequestedAt) { - await this.lifecycleRepo.markCancelled(aiJobId); - return; - } - - await context.updateProgress(30); - - // ── EXECUTE ── - // 9. 调用 AiGatewayService(不直接导入 Provider SDK) const response = await this.aiGateway.generate( { userId: job.userId, @@ -169,15 +167,13 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { tier: def.model.modelTier as any, promptKey: def.prompt.promptKey, promptVersion: def.prompt.promptVersion, - messages: [], // Snapshot content 应通过 promptTemplate 渲染,此处由 AiGateway 处理 + messages: [], maxTokens: def.model.maxTokens, - outputSchema: undefined, // 由 AiGateway 的 parseJson 处理 + outputSchema: undefined, }, timeoutMs, ); - clearTimeout(timeoutId); - await context.updateProgress(70); // 10. 取消检查(Projector 前) @@ -203,19 +199,17 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { await context.updateProgress(90); // ── COMPLETE ── - // 11. Usage logging + // 9. Usage logging await this.writeUsageLog(job.userId, aiJobId, def, response, lockedJob.attemptCount || 0).catch((err) => { this.logger.error(`Usage log failed for job=${aiJobId}: ${err.message}`); }); - // 12. 标记成功 + // 10. 标记成功 await this.lifecycleRepo.markSucceeded(aiJobId); await context.updateProgress(100); this.logger.log(`Job ${aiJobId} (${job.jobType}) completed successfully`); } catch (execErr: any) { - clearTimeout(timeoutId); - // 取消检查 if (execErr?.message?.includes('cancelled')) { await this.lifecycleRepo.markCancelled(aiJobId); @@ -282,14 +276,13 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { } private computeHash(content: string): string { - const crypto = require('crypto'); return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16); } private async writeUsageLog( userId: string, jobId: string, - _def: any, + definition: any, response: any, attemptNo: number, ): Promise { @@ -300,8 +293,8 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { provider: response?.usage?.provider || 'deepseek', model: response?.usage?.model || 'deepseek-chat', tier: 'primary', - promptKey: _def?.prompt?.promptKey || 'unknown', - promptVersion: _def?.prompt?.promptVersion || 'unknown', + promptKey: definition?.prompt?.promptKey || 'unknown', + promptVersion: definition?.prompt?.promptVersion || 'unknown', inputTokens: response?.usage?.inputTokens || 0, outputTokens: response?.usage?.outputTokens || 0, estimatedCost: response?.usage?.estimatedCost || 0,