import type { AgentGenerateRequest, AgentJobState, AgentJobStatusResponse } from './agent-api-contracts'; import { AGENT_ENDPOINTS, buildAgentJobResultPath } from './agent-api-paths.mjs'; import { completeAgentExecutionState, createArtifactPersistenceError, createCompletionPersistenceError, deleteAgentExecutionFiles, errorToAgentErrorBody, executeAgentGenerate, hydrateAgentReplayResponse, saveAgentExecutionArtifacts, type AgentGeneratePreparation } from './agent-image-service'; import type { AgentRequestRecord, AgentStateStore } from './agent-state-store'; import { AgentApiError, toTerminalAgentErrorBody, type AgentErrorBody } from './api-error-response'; import { appLogger } from './app-logger'; const DEFAULT_JOB_RETRY_AFTER_SECONDS = 5; const MAX_JOB_LEASE_REFRESH_INTERVAL_MS = 30_000; export function buildAgentJobStatusResponse( record: AgentRequestRecord, options: { now?: Date; retryAfterSeconds?: number } = {} ): AgentJobStatusResponse { const state = readAgentJobState(record, options.now); return { job: { id: record.requestId, request_id: record.requestId, idempotency_key: record.idempotencyKey, mode: record.mode, state, created_at: record.createdAt, updated_at: record.updatedAt, expires_at: record.expiresAt, ...(state !== 'expired' ? { result_url: buildAgentJobResultPath(record.requestId) } : {}), ...(state === 'running' || state === 'queued' ? { retry_after_seconds: options.retryAfterSeconds ?? DEFAULT_JOB_RETRY_AFTER_SECONDS } : {}), ...(state === 'failed' && record.errorJson ? { error: summarizeJobError(record.errorJson) } : {}) } }; } export function readAgentJobState(record: AgentRequestRecord, now = new Date()): AgentJobState { if (isExpiredJobRecord(record, now)) return 'expired'; if (record.status === 'pending') return 'queued'; if (record.status === 'running') return 'running'; if (record.status === 'succeeded') return 'succeeded'; if (record.status === 'failed') return 'failed'; return 'expired'; } export function assertReadableJobRecord(record: AgentRequestRecord | undefined, id: string): AgentRequestRecord { if (!record) { throw new AgentApiError({ code: 'job_not_found', message: 'Agent job 不存在。', status: 404, retryable: false, details: { id } }); } if (readAgentJobState(record) === 'expired') { throw new AgentApiError({ code: 'job_expired', message: 'Agent job 已过期。', status: 410, retryable: false, details: { id } }); } return record; } export async function readJobResult(store: AgentStateStore, record: AgentRequestRecord): Promise { if (record.status === 'succeeded' && record.responseJson) { return { type: 'response', response: await hydrateAgentReplayResponse(store, record, record.responseJson, record.responseJson.cached) }; } if (record.status === 'failed' && record.errorJson) { return { type: 'stored_error', error: toTerminalJobErrorBody(record.errorJson) }; } return { type: 'running' }; } export function startAgentGenerateJob(options: { store: AgentStateStore; request: AgentGenerateRequest; headers: Headers; requestId: string; idempotencyKey: string; leaseMs: number; preparation?: AgentGeneratePreparation; transportEndpoint?: string; }): void { void runAgentGenerateJob(options).catch((error) => { appLogger.error('Agent generate job 后台执行失败。', error); }); } type ResponseReadyResult = { type: 'response'; response: unknown } | { type: 'stored_error'; error: AgentErrorBody } | { type: 'running' }; async function runAgentGenerateJob(options: { store: AgentStateStore; request: AgentGenerateRequest; headers: Headers; requestId: string; idempotencyKey: string; leaseMs: number; preparation?: AgentGeneratePreparation; transportEndpoint?: string; }): Promise { let heartbeat: { stop: () => void } | undefined; try { heartbeat = startAgentJobLeaseHeartbeat(options.store, options.requestId, options.leaseMs); const execution = await executeAgentGenerate({ request: options.request, headers: options.headers, requestId: options.requestId, idempotencyKey: options.idempotencyKey, cached: false, preparation: options.preparation, transport: { transport: 'agent_job_polling', endpoint: options.transportEndpoint ?? AGENT_ENDPOINTS.create_generate_job, route_mode: 'job' } }); await persistAgentJobSuccess(options.store, execution, options.requestId); } catch (error) { await failAgentJob(options.store, options.requestId, error); } finally { heartbeat?.stop(); } } function startAgentJobLeaseHeartbeat(store: AgentStateStore, requestId: string, leaseMs: number): { stop: () => void } { if (!Number.isFinite(leaseMs) || leaseMs <= 0) { throw new Error('leaseMs must be a positive number.'); } const intervalMs = Math.max(100, Math.min(MAX_JOB_LEASE_REFRESH_INTERVAL_MS, Math.floor(leaseMs / 2))); const timer = setInterval(() => { void store.refreshRequestLease({ requestId, leaseMs }).catch((error) => { appLogger.error('刷新 Agent job lease 失败。', error); }); }, intervalMs); timer.unref?.(); return { stop: () => clearInterval(timer) }; } async function persistAgentJobSuccess( store: AgentStateStore, execution: Awaited>, requestId: string ): Promise { try { await saveAgentExecutionArtifacts(store, execution); } catch (error) { appLogger.error('保存 Agent job 产物元数据失败。', error); await cleanupFailedAgentJob(store, execution, requestId); return; } try { await completeAgentExecutionState(store, execution); } catch (error) { appLogger.error('保存 Agent job 完成状态失败。', error); await failAgentJob(store, requestId, createCompletionPersistenceError()); } } async function cleanupFailedAgentJob( store: AgentStateStore, execution: Awaited>, requestId: string ): Promise { try { await deleteAgentExecutionFiles(execution); } catch (error) { appLogger.error('Agent job 产物元数据保存失败后清理文件失败。', error); } await failAgentJob(store, requestId, createArtifactPersistenceError()); } async function failAgentJob(store: AgentStateStore, requestId: string, error: unknown): Promise { try { await store.failRequest({ requestId, error: errorToAgentErrorBody(error, requestId) }); } catch (storeError) { appLogger.error('保存 Agent job 失败状态失败。', storeError); } } function isExpiredJobRecord(record: AgentRequestRecord, now: Date): boolean { return new Date(record.expiresAt).getTime() <= now.getTime(); } function summarizeJobError(errorBody: AgentErrorBody) { const error = toTerminalJobError(errorBody); return { code: error.code, message: error.message, retryable: error.retryable, ...(error.details ? { details: error.details } : {}), ...(error.upstream_status !== undefined ? { upstream_status: error.upstream_status } : {}), ...(error.diagnostics ? { diagnostics: error.diagnostics } : {}) }; } function toTerminalJobErrorBody(errorBody: AgentErrorBody): AgentErrorBody { return toTerminalAgentErrorBody(errorBody); } function toTerminalJobError(errorBody: AgentErrorBody): AgentErrorBody['error'] { return toTerminalAgentErrorBody(errorBody).error; }