/** * JobManager — 异步 Skill 创建/重训任务管理器 * * 管理长时运行的 Skill 任务(Hunt→Digest→Evolve)的生命周期。 * 支持磁盘持久化,服务器重启后自动恢复历史 Job 状态。 */ const { randomUUID } = require('crypto'); const fs = require('fs'); const path = require('path'); // Job TTL: 完成后保留 7 天(持久化后可保留更久) const JOB_TTL = 7 * 24 * 60 * 60 * 1000; const CLEANUP_INTERVAL = 5 * 60 * 1000; // 每 5 分钟清理一次 class JobManager { /** * @param {object} [options] * @param {string} [options.storePath] - 持久化文件路径。如果提供则启用磁盘持久化。 */ constructor(options = {}) { /** @type {Map} */ this.jobs = new Map(); this.storePath = options.storePath || null; // 从磁盘恢复 if (this.storePath) { this._loadFromDisk(); } // 定期清理过期任务 this._cleanupTimer = setInterval(() => this._cleanup(), CLEANUP_INTERVAL); this._cleanupTimer.unref(); // 不阻止进程退出 } /** * 创建新任务 * @param {object} request * @returns {Job} */ createJob(request) { const jobId = `job-${Date.now()}-${randomUUID().substring(0, 8)}`; const now = new Date().toISOString(); const job = { jobId, status: 'queued', domain: request.domain || '', scope: request.scope || '', description: request.description || '', keywords: request.keywords || [], skillName: request.skillName || null, progress: '任务已加入队列', result: null, error: null, webhookUrl: request.webhookUrl || null, apiKey: request.apiKey || null, createdAt: now, updatedAt: now, // 额外字段透传 ...(request.type ? { type: request.type } : {}), ...(request.conversationId ? { conversationId: request.conversationId } : {}), }; this.jobs.set(jobId, job); this._persist(); console.log(`[JobManager] Job created: ${jobId}`); return job; } /** * 更新任务状态 * @param {string} jobId * @param {object} updates */ updateJob(jobId, updates) { const job = this.jobs.get(jobId); if (!job) return null; Object.assign(job, updates, { updatedAt: new Date().toISOString() }); this._persist(); // 如果任务完成或失败,触发 webhook if ((updates.status === 'completed' || updates.status === 'failed') && job.webhookUrl) { this._sendWebhook(job).catch(err => { console.warn(`[JobManager] Webhook failed for ${jobId}:`, err.message); }); } return job; } /** * 获取任务信息 * @param {string} jobId * @returns {Job|null} */ getJob(jobId) { return this.jobs.get(jobId) || null; } /** * 获取公开安全的 Job 信息(隐藏 apiKey, webhookUrl) */ getPublicJob(jobId) { const job = this.getJob(jobId); if (!job) return null; return JobManager.toPublicJob(job); } /** * 列出任务(可按 apiKey 过滤) * @param {object} [options] * @param {string} [options.apiKey] * @param {number} [options.limit=20] * @returns {Job[]} */ listJobs({ apiKey, limit = 20 } = {}) { let jobs = Array.from(this.jobs.values()); if (apiKey) { jobs = jobs.filter(j => j.apiKey === apiKey); } // 按创建时间倒序 jobs.sort((a, b) => new Date(b.createdAt) - new Date(a.createdAt)); return jobs.slice(0, limit).map(j => JobManager.toPublicJob(j)); } /** * 将 Job 转为公开安全字段 */ static toPublicJob(job) { const { apiKey, webhookUrl, ...rest } = job; return rest; } // ══════════════════════════════════════ // 磁盘持久化 // ══════════════════════════════════════ /** * 保存所有 Job 到磁盘 * @private */ _persist() { if (!this.storePath) return; try { const dir = path.dirname(this.storePath); if (!fs.existsSync(dir)) fs.mkdirSync(dir, { recursive: true }); const data = []; for (const job of this.jobs.values()) { data.push(job); } fs.writeFileSync(this.storePath, JSON.stringify(data, null, 2), 'utf-8'); } catch (err) { console.warn(`[JobManager] Persist error:`, err.message); } } /** * 从磁盘恢复 Job * @private */ _loadFromDisk() { if (!this.storePath || !fs.existsSync(this.storePath)) return; try { const raw = fs.readFileSync(this.storePath, 'utf-8'); const data = JSON.parse(raw); if (!Array.isArray(data)) return; let loaded = 0; for (const job of data) { if (job.jobId) { // 将 running/hunting/digesting/creating 状态标记为中断 if (!['completed', 'failed', 'queued'].includes(job.status)) { job.status = 'failed'; job.progress = '服务器重启,任务中断'; job.error = 'Server restarted while job was running'; job.updatedAt = new Date().toISOString(); } this.jobs.set(job.jobId, job); loaded++; } } if (loaded > 0) { console.log(`[JobManager] Restored ${loaded} jobs from ${this.storePath}`); } } catch (err) { console.warn(`[JobManager] Load error:`, err.message); } } // ══════════════════════════════════════ // Webhook & Cleanup // ══════════════════════════════════════ /** * 发送 Webhook 回调 * @param {Job} job * @private */ async _sendWebhook(job) { if (!job.webhookUrl) return; try { const response = await fetch(job.webhookUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ jobId: job.jobId, status: job.status, skillName: job.skillName, progress: job.progress, error: job.error, updatedAt: job.updatedAt, }), signal: AbortSignal.timeout(10000), }); console.log(`[JobManager] Webhook sent for ${job.jobId}: ${response.status}`); } catch (err) { console.warn(`[JobManager] Webhook error for ${job.jobId}:`, err.message); } } /** * 清理过期任务 * @private */ _cleanup() { const now = Date.now(); let cleaned = 0; for (const [id, job] of this.jobs) { if (job.status === 'completed' || job.status === 'failed') { const age = now - new Date(job.updatedAt).getTime(); if (age > JOB_TTL) { this.jobs.delete(id); cleaned++; } } } if (cleaned > 0) { this._persist(); console.log(`[JobManager] Cleaned ${cleaned} expired jobs`); } } /** * 关闭清理定时器 */ shutdown() { if (this._cleanupTimer) { clearInterval(this._cleanupTimer); this._cleanupTimer = null; } } } module.exports = { JobManager };