960 lines
39 KiB
JavaScript
960 lines
39 KiB
JavaScript
/**
|
|
* Worker Daemon Service
|
|
* Node.js-based background worker system that auto-runs like shell daemons
|
|
*
|
|
* Workers:
|
|
* - map: Codebase mapping (5 min interval)
|
|
* - audit: Security analysis (10 min interval)
|
|
* - optimize: Performance optimization (15 min interval)
|
|
* - consolidate: Memory consolidation (30 min interval)
|
|
* - testgaps: Test coverage analysis (20 min interval)
|
|
*/
|
|
import { EventEmitter } from 'events';
|
|
import { existsSync, mkdirSync, writeFileSync, readFileSync, appendFileSync, unlinkSync } from 'fs';
|
|
import { cpus } from 'os';
|
|
import { join } from 'path';
|
|
import { HeadlessWorkerExecutor, isHeadlessWorker, } from './headless-worker-executor.js';
|
|
// Default worker configurations with improved intervals (P0 fix: map 5min -> 15min)
|
|
const DEFAULT_WORKERS = [
|
|
{ type: 'map', intervalMs: 15 * 60 * 1000, offsetMs: 0, priority: 'normal', description: 'Codebase mapping', enabled: true },
|
|
{ type: 'audit', intervalMs: 10 * 60 * 1000, offsetMs: 2 * 60 * 1000, priority: 'critical', description: 'Security analysis', enabled: true },
|
|
{ type: 'optimize', intervalMs: 15 * 60 * 1000, offsetMs: 4 * 60 * 1000, priority: 'high', description: 'Performance optimization', enabled: true },
|
|
{ type: 'consolidate', intervalMs: 30 * 60 * 1000, offsetMs: 6 * 60 * 1000, priority: 'low', description: 'Memory consolidation', enabled: true },
|
|
{ type: 'testgaps', intervalMs: 20 * 60 * 1000, offsetMs: 8 * 60 * 1000, priority: 'normal', description: 'Test coverage analysis', enabled: true },
|
|
{ type: 'predict', intervalMs: 10 * 60 * 1000, offsetMs: 0, priority: 'low', description: 'Predictive preloading', enabled: false },
|
|
{ type: 'document', intervalMs: 60 * 60 * 1000, offsetMs: 0, priority: 'low', description: 'Auto-documentation', enabled: false },
|
|
];
|
|
// Worker timeout — must exceed the longest per-worker headless timeout (15 min for audit/refactor).
|
|
// Previously 5 min, which caused orphan processes when daemon timeout fired before executor timeout (#1117).
|
|
const DEFAULT_WORKER_TIMEOUT_MS = 16 * 60 * 1000;
|
|
/**
|
|
* Worker Daemon - Manages background workers with Node.js
|
|
*/
|
|
export class WorkerDaemon extends EventEmitter {
|
|
config;
|
|
workers = new Map();
|
|
timers = new Map();
|
|
running = false;
|
|
startedAt;
|
|
projectRoot;
|
|
runningWorkers = new Set(); // Track concurrent workers
|
|
pendingWorkers = []; // Queue for deferred workers
|
|
// Headless execution support
|
|
headlessExecutor = null;
|
|
headlessAvailable = false;
|
|
// Preserve the original constructor config so we can detect explicit overrides
|
|
// during state restoration (R1: constructor config takes priority over stale state)
|
|
originalConfig;
|
|
constructor(projectRoot, config) {
|
|
super();
|
|
this.projectRoot = projectRoot;
|
|
this.originalConfig = config;
|
|
const claudeFlowDir = join(projectRoot, '.claude-flow');
|
|
// Read daemon config from .claude-flow/config.json (Layer B)
|
|
const fileConfig = this.readDaemonConfigFromFile(claudeFlowDir);
|
|
// CPU-proportional smart default instead of hardcoded 2.0
|
|
const cpuCount = WorkerDaemon.getEffectiveCpuCount();
|
|
const smartMaxCpuLoad = Math.max(cpuCount * 0.8, 2.0); // Floor of 2.0 for single-CPU machines
|
|
// Platform-aware default: macOS os.freemem() excludes reclaimable file cache,
|
|
// so reported "free" is much lower than actually available memory.
|
|
// Linux reports available memory (including reclaimable cache) more accurately.
|
|
const defaultMinFreeMemory = process.platform === 'darwin' ? 5 : 10;
|
|
// Priority: constructor arg > config.json > smart default
|
|
// For resourceThresholds, merge field-by-field so partial overrides
|
|
// (e.g. only --max-cpu-load) still pick up defaults for other fields.
|
|
this.config = {
|
|
autoStart: config?.autoStart ?? fileConfig.autoStart ?? false,
|
|
logDir: config?.logDir ?? join(claudeFlowDir, 'logs'),
|
|
stateFile: config?.stateFile ?? join(claudeFlowDir, 'daemon-state.json'),
|
|
maxConcurrent: config?.maxConcurrent ?? fileConfig.maxConcurrent ?? 2,
|
|
workerTimeoutMs: config?.workerTimeoutMs ?? fileConfig.workerTimeoutMs ?? DEFAULT_WORKER_TIMEOUT_MS,
|
|
resourceThresholds: {
|
|
maxCpuLoad: config?.resourceThresholds?.maxCpuLoad ?? fileConfig.maxCpuLoad ?? smartMaxCpuLoad,
|
|
minFreeMemoryPercent: config?.resourceThresholds?.minFreeMemoryPercent ?? fileConfig.minFreeMemoryPercent ?? defaultMinFreeMemory,
|
|
},
|
|
workers: config?.workers ?? DEFAULT_WORKERS,
|
|
};
|
|
// Setup graceful shutdown handlers
|
|
this.setupShutdownHandlers();
|
|
// Ensure directories exist
|
|
if (!existsSync(claudeFlowDir)) {
|
|
mkdirSync(claudeFlowDir, { recursive: true });
|
|
}
|
|
if (!existsSync(this.config.logDir)) {
|
|
mkdirSync(this.config.logDir, { recursive: true });
|
|
}
|
|
// Initialize worker states
|
|
this.initializeWorkerStates();
|
|
// Initialize headless executor (async, non-blocking)
|
|
this.initHeadlessExecutor().catch((err) => {
|
|
this.log('warn', `Headless executor init failed: ${err}`);
|
|
});
|
|
}
|
|
/**
|
|
* Initialize headless executor if Claude Code is available
|
|
*/
|
|
async initHeadlessExecutor() {
|
|
try {
|
|
this.headlessExecutor = new HeadlessWorkerExecutor(this.projectRoot, {
|
|
maxConcurrent: this.config.maxConcurrent,
|
|
});
|
|
this.headlessAvailable = await this.headlessExecutor.isAvailable();
|
|
if (this.headlessAvailable) {
|
|
this.log('info', 'Claude Code headless mode available - AI workers enabled');
|
|
// Forward headless executor events
|
|
this.headlessExecutor.on('execution:start', (data) => {
|
|
this.emit('headless:start', data);
|
|
});
|
|
this.headlessExecutor.on('execution:complete', (data) => {
|
|
this.emit('headless:complete', data);
|
|
});
|
|
this.headlessExecutor.on('execution:error', (data) => {
|
|
this.emit('headless:error', data);
|
|
});
|
|
this.headlessExecutor.on('output', (data) => {
|
|
this.emit('headless:output', data);
|
|
});
|
|
}
|
|
else {
|
|
this.log('info', 'Claude Code not found - AI workers will run in local fallback mode');
|
|
}
|
|
}
|
|
catch (error) {
|
|
this.log('warn', `Failed to initialize headless executor: ${error}`);
|
|
this.headlessAvailable = false;
|
|
}
|
|
}
|
|
/**
|
|
* Check if headless execution is available
|
|
*/
|
|
isHeadlessAvailable() {
|
|
return this.headlessAvailable;
|
|
}
|
|
/**
|
|
* Get headless executor instance
|
|
*/
|
|
getHeadlessExecutor() {
|
|
return this.headlessExecutor;
|
|
}
|
|
/**
|
|
* Detect effective CPU count for the current environment.
|
|
*
|
|
* Inside Docker / K8s containers, os.cpus().length reports the HOST cpu
|
|
* count, not the container limit (Node.js #28762 — wontfix). We read
|
|
* cgroup v2 / v1 quota files first so the maxCpuLoad threshold stays
|
|
* meaningful under resource-limited containers.
|
|
*/
|
|
static getEffectiveCpuCount() {
|
|
// 1. Try cgroup v2: /sys/fs/cgroup/cpu.max
|
|
try {
|
|
const cpuMax = readFileSync('/sys/fs/cgroup/cpu.max', 'utf8').trim();
|
|
const [quotaStr, periodStr] = cpuMax.split(' ');
|
|
if (quotaStr !== 'max') {
|
|
const quota = parseInt(quotaStr, 10);
|
|
const period = parseInt(periodStr, 10);
|
|
if (quota > 0 && period > 0)
|
|
return Math.ceil(quota / period);
|
|
}
|
|
}
|
|
catch { /* not in cgroup v2 */ }
|
|
// 2. Try cgroup v1: /sys/fs/cgroup/cpu/cpu.cfs_quota_us
|
|
try {
|
|
const quota = parseInt(readFileSync('/sys/fs/cgroup/cpu/cpu.cfs_quota_us', 'utf8').trim(), 10);
|
|
const period = parseInt(readFileSync('/sys/fs/cgroup/cpu/cpu.cfs_period_us', 'utf8').trim(), 10);
|
|
if (quota > 0 && period > 0)
|
|
return Math.ceil(quota / period);
|
|
}
|
|
catch { /* not in cgroup v1 */ }
|
|
// 3. Fallback to os.cpus().length
|
|
return cpus().length || 1;
|
|
}
|
|
/**
|
|
* Read daemon-specific config from .claude-flow/config.json
|
|
* Supports dot-notation keys like 'daemon.resourceThresholds.maxCpuLoad'
|
|
*/
|
|
readDaemonConfigFromFile(claudeFlowDir) {
|
|
const configPath = join(claudeFlowDir, 'config.json');
|
|
if (!existsSync(configPath)) {
|
|
// Warn if config.yaml exists but config.json does not (#1395 Bug 4)
|
|
const yamlPath = join(claudeFlowDir, 'config.yaml');
|
|
const ymlPath = join(claudeFlowDir, 'config.yml');
|
|
if (existsSync(yamlPath) || existsSync(ymlPath)) {
|
|
this.log('warn', `Found ${existsSync(yamlPath) ? 'config.yaml' : 'config.yml'} but daemon reads only config.json — YAML config is being ignored. Convert to JSON or create config.json.`);
|
|
}
|
|
return {};
|
|
}
|
|
try {
|
|
const raw = JSON.parse(readFileSync(configPath, 'utf-8'));
|
|
// Support both flat keys at root and nested under scopes.project
|
|
const cfg = raw?.scopes?.project ?? raw;
|
|
const rawCpuLoad = cfg['daemon.resourceThresholds.maxCpuLoad'] ?? raw['daemon.resourceThresholds.maxCpuLoad'];
|
|
const rawMinMem = cfg['daemon.resourceThresholds.minFreeMemoryPercent'] ?? raw['daemon.resourceThresholds.minFreeMemoryPercent'];
|
|
const rawMaxConcurrent = cfg['daemon.maxConcurrent'] ?? raw['daemon.maxConcurrent'];
|
|
const rawTimeout = cfg['daemon.workerTimeoutMs'] ?? raw['daemon.workerTimeoutMs'];
|
|
return {
|
|
autoStart: typeof raw['daemon.autoStart'] === 'boolean' ? raw['daemon.autoStart'] : undefined,
|
|
maxConcurrent: (typeof rawMaxConcurrent === 'number' && rawMaxConcurrent > 0) ? rawMaxConcurrent : undefined,
|
|
workerTimeoutMs: (typeof rawTimeout === 'number' && rawTimeout > 0) ? rawTimeout : undefined,
|
|
maxCpuLoad: (typeof rawCpuLoad === 'number' && rawCpuLoad > 0 && rawCpuLoad < 1000) ? rawCpuLoad : undefined,
|
|
minFreeMemoryPercent: (typeof rawMinMem === 'number' && rawMinMem >= 0 && rawMinMem <= 100) ? rawMinMem : undefined,
|
|
};
|
|
}
|
|
catch {
|
|
return {};
|
|
}
|
|
}
|
|
/**
|
|
* Setup graceful shutdown handlers
|
|
*/
|
|
setupShutdownHandlers() {
|
|
const shutdown = async () => {
|
|
this.log('info', 'Received shutdown signal, stopping daemon...');
|
|
await this.stop();
|
|
process.exit(0);
|
|
};
|
|
process.on('SIGTERM', shutdown);
|
|
process.on('SIGINT', shutdown);
|
|
process.on('SIGHUP', shutdown);
|
|
}
|
|
/**
|
|
* Check if system resources allow worker execution
|
|
*/
|
|
async canRunWorker() {
|
|
const os = await import('os');
|
|
const cpuLoad = os.loadavg()[0];
|
|
const totalMem = os.totalmem();
|
|
const freeMem = os.freemem();
|
|
const freePercent = (freeMem / totalMem) * 100;
|
|
if (cpuLoad > this.config.resourceThresholds.maxCpuLoad) {
|
|
return { allowed: false, reason: `CPU load too high: ${cpuLoad.toFixed(2)}` };
|
|
}
|
|
if (freePercent < this.config.resourceThresholds.minFreeMemoryPercent) {
|
|
return { allowed: false, reason: `Memory too low: ${freePercent.toFixed(1)}% free` };
|
|
}
|
|
return { allowed: true };
|
|
}
|
|
/**
|
|
* Process pending workers queue
|
|
*
|
|
* When executeWorkerWithConcurrencyControl defers a worker (returns null),
|
|
* we break immediately to avoid a busy-wait loop — the deferred worker is
|
|
* already back on the pendingWorkers queue by that point. If no workers are
|
|
* currently running when we break, we schedule a backoff retry so the queue
|
|
* does not get permanently stuck.
|
|
*/
|
|
async processPendingWorkers() {
|
|
while (this.pendingWorkers.length > 0 && this.runningWorkers.size < this.config.maxConcurrent) {
|
|
const workerType = this.pendingWorkers.shift();
|
|
const workerConfig = this.config.workers.find(w => w.type === workerType);
|
|
if (workerConfig) {
|
|
const result = await this.executeWorkerWithConcurrencyControl(workerConfig);
|
|
if (result === null) {
|
|
// Worker was deferred (resource pressure or concurrency limit).
|
|
// Break to avoid tight-looping — the next executeWorker() completion
|
|
// will call processPendingWorkers() again via the finally block.
|
|
if (this.runningWorkers.size === 0) {
|
|
// No workers running means nobody will trigger the finally-block
|
|
// callback, so schedule a backoff retry to avoid a stuck queue.
|
|
setTimeout(() => this.processPendingWorkers(), 30_000).unref();
|
|
}
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
initializeWorkerStates() {
|
|
// Try to restore state from file
|
|
if (existsSync(this.config.stateFile)) {
|
|
try {
|
|
const saved = JSON.parse(readFileSync(this.config.stateFile, 'utf-8'));
|
|
// CRITICAL: Restore worker config (including enabled flag) from saved state
|
|
// This fixes #950: daemon enable command not persisting worker state
|
|
if (saved.config?.workers && Array.isArray(saved.config.workers)) {
|
|
for (const savedWorker of saved.config.workers) {
|
|
const workerConfig = this.config.workers.find(w => w.type === savedWorker.type);
|
|
if (workerConfig && typeof savedWorker.enabled === 'boolean') {
|
|
workerConfig.enabled = savedWorker.enabled;
|
|
}
|
|
}
|
|
}
|
|
// Restore resourceThresholds, maxConcurrent, workerTimeoutMs from saved state
|
|
// Only restore if valid numeric values within sane ranges
|
|
if (saved.config?.resourceThresholds && !this.originalConfig?.resourceThresholds) {
|
|
const rt = saved.config.resourceThresholds;
|
|
if (typeof rt.maxCpuLoad === 'number' && rt.maxCpuLoad > 0 && rt.maxCpuLoad < 1000) {
|
|
this.config.resourceThresholds.maxCpuLoad = rt.maxCpuLoad;
|
|
}
|
|
if (typeof rt.minFreeMemoryPercent === 'number' && rt.minFreeMemoryPercent >= 0 && rt.minFreeMemoryPercent <= 100) {
|
|
this.config.resourceThresholds.minFreeMemoryPercent = rt.minFreeMemoryPercent;
|
|
}
|
|
}
|
|
if (typeof saved.config?.maxConcurrent === 'number' && saved.config.maxConcurrent > 0) {
|
|
this.config.maxConcurrent = saved.config.maxConcurrent;
|
|
}
|
|
if (typeof saved.config?.workerTimeoutMs === 'number' && saved.config.workerTimeoutMs > 0) {
|
|
this.config.workerTimeoutMs = saved.config.workerTimeoutMs;
|
|
}
|
|
// Restore worker runtime states (runCount, successCount, etc.)
|
|
if (saved.workers) {
|
|
for (const [type, state] of Object.entries(saved.workers)) {
|
|
const savedState = state;
|
|
const lastRunValue = savedState.lastRun;
|
|
this.workers.set(type, {
|
|
runCount: savedState.runCount || 0,
|
|
successCount: savedState.successCount || 0,
|
|
failureCount: savedState.failureCount || 0,
|
|
averageDurationMs: savedState.averageDurationMs || 0,
|
|
lastRun: lastRunValue ? new Date(lastRunValue) : undefined,
|
|
nextRun: undefined,
|
|
isRunning: false,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
catch {
|
|
// Ignore parse errors, start fresh
|
|
}
|
|
}
|
|
// Initialize any missing workers
|
|
for (const workerConfig of this.config.workers) {
|
|
if (!this.workers.has(workerConfig.type)) {
|
|
this.workers.set(workerConfig.type, {
|
|
runCount: 0,
|
|
successCount: 0,
|
|
failureCount: 0,
|
|
averageDurationMs: 0,
|
|
isRunning: false,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
/**
|
|
* Get the PID file path for singleton enforcement (#1395 Bug 3).
|
|
*/
|
|
get pidFile() {
|
|
return join(this.projectRoot, '.claude-flow', 'daemon.pid');
|
|
}
|
|
/**
|
|
* Check if another daemon instance is already running.
|
|
* Returns the existing PID if alive, or null if no daemon is running.
|
|
*/
|
|
checkExistingDaemon() {
|
|
if (!existsSync(this.pidFile))
|
|
return null;
|
|
try {
|
|
const pid = parseInt(readFileSync(this.pidFile, 'utf-8').trim(), 10);
|
|
if (isNaN(pid))
|
|
return null;
|
|
// Check if process is alive (signal 0 = existence check)
|
|
process.kill(pid, 0);
|
|
return pid; // Process is alive
|
|
}
|
|
catch {
|
|
// Process is dead — clean up stale PID file
|
|
try {
|
|
unlinkSync(this.pidFile);
|
|
}
|
|
catch { /* ignore */ }
|
|
return null;
|
|
}
|
|
}
|
|
/**
|
|
* Write PID file for singleton enforcement.
|
|
*/
|
|
writePidFile() {
|
|
const dir = join(this.projectRoot, '.claude-flow');
|
|
if (!existsSync(dir))
|
|
mkdirSync(dir, { recursive: true });
|
|
writeFileSync(this.pidFile, String(process.pid), 'utf-8');
|
|
}
|
|
/**
|
|
* Remove PID file on shutdown.
|
|
*/
|
|
removePidFile() {
|
|
try {
|
|
unlinkSync(this.pidFile);
|
|
}
|
|
catch { /* ignore */ }
|
|
}
|
|
/**
|
|
* Start the daemon and all enabled workers
|
|
*/
|
|
async start() {
|
|
if (this.running) {
|
|
this.emit('warning', 'Daemon already running');
|
|
return;
|
|
}
|
|
// PID singleton enforcement (#1395 Bug 3): prevent daemon accumulation
|
|
const existingPid = this.checkExistingDaemon();
|
|
if (existingPid !== null) {
|
|
this.log('info', `Daemon already running (PID: ${existingPid}), skipping start`);
|
|
this.emit('warning', `Daemon already running (PID: ${existingPid})`);
|
|
return;
|
|
}
|
|
this.running = true;
|
|
this.startedAt = new Date();
|
|
this.writePidFile();
|
|
this.emit('started', { pid: process.pid, startedAt: this.startedAt });
|
|
// Schedule all enabled workers
|
|
for (const workerConfig of this.config.workers) {
|
|
if (workerConfig.enabled) {
|
|
this.scheduleWorker(workerConfig);
|
|
}
|
|
}
|
|
// Save state
|
|
this.saveState();
|
|
this.log('info', `Daemon started (PID: ${process.pid}, CPUs: ${cpus().length}, workers: ${this.config.workers.filter(w => w.enabled).length}, maxCpuLoad: ${this.config.resourceThresholds.maxCpuLoad}, minFreeMemoryPercent: ${this.config.resourceThresholds.minFreeMemoryPercent}%)`);
|
|
}
|
|
/**
|
|
* Stop the daemon and all workers
|
|
*/
|
|
async stop() {
|
|
if (!this.running) {
|
|
this.emit('warning', 'Daemon not running');
|
|
return;
|
|
}
|
|
// Clear all timers (convert to array to avoid iterator issues)
|
|
const timerEntries = Array.from(this.timers.entries());
|
|
for (const [type, timer] of timerEntries) {
|
|
clearTimeout(timer);
|
|
this.log('info', `Stopped worker: ${type}`);
|
|
}
|
|
this.timers.clear();
|
|
this.running = false;
|
|
this.removePidFile();
|
|
this.saveState();
|
|
this.emit('stopped', { stoppedAt: new Date() });
|
|
this.log('info', 'Daemon stopped');
|
|
}
|
|
/**
|
|
* Get daemon status
|
|
*/
|
|
getStatus() {
|
|
return {
|
|
running: this.running,
|
|
pid: process.pid,
|
|
startedAt: this.startedAt,
|
|
workers: new Map(this.workers),
|
|
config: this.config,
|
|
};
|
|
}
|
|
/**
|
|
* Schedule a worker to run at intervals with staggered start
|
|
*/
|
|
scheduleWorker(workerConfig) {
|
|
const state = this.workers.get(workerConfig.type);
|
|
const internalConfig = workerConfig;
|
|
const staggerOffset = internalConfig.offsetMs || 0;
|
|
// Calculate initial delay with stagger offset
|
|
let initialDelay = staggerOffset;
|
|
if (state.lastRun) {
|
|
const timeSinceLastRun = Date.now() - state.lastRun.getTime();
|
|
initialDelay = Math.max(staggerOffset, workerConfig.intervalMs - timeSinceLastRun);
|
|
}
|
|
state.nextRun = new Date(Date.now() + initialDelay);
|
|
const runAndReschedule = async () => {
|
|
if (!this.running)
|
|
return;
|
|
// Use concurrency-controlled execution (P0 fix)
|
|
await this.executeWorkerWithConcurrencyControl(workerConfig);
|
|
// Reschedule
|
|
if (this.running) {
|
|
const timer = setTimeout(runAndReschedule, workerConfig.intervalMs);
|
|
this.timers.set(workerConfig.type, timer);
|
|
state.nextRun = new Date(Date.now() + workerConfig.intervalMs);
|
|
}
|
|
};
|
|
// Schedule first run with stagger offset
|
|
const timer = setTimeout(runAndReschedule, initialDelay);
|
|
this.timers.set(workerConfig.type, timer);
|
|
this.log('info', `Scheduled ${workerConfig.type} (interval: ${workerConfig.intervalMs / 1000}s, first run in ${initialDelay / 1000}s)`);
|
|
}
|
|
/**
|
|
* Execute a worker with concurrency control (P0 fix)
|
|
*/
|
|
async executeWorkerWithConcurrencyControl(workerConfig) {
|
|
// Check concurrency limit
|
|
if (this.runningWorkers.size >= this.config.maxConcurrent) {
|
|
this.log('info', `Worker ${workerConfig.type} deferred: max concurrent (${this.config.maxConcurrent}) reached`);
|
|
this.pendingWorkers.push(workerConfig.type);
|
|
this.emit('worker:deferred', { type: workerConfig.type, reason: 'max_concurrent' });
|
|
return null;
|
|
}
|
|
// Check resource availability
|
|
const resourceCheck = await this.canRunWorker();
|
|
if (!resourceCheck.allowed) {
|
|
this.log('info', `Worker ${workerConfig.type} deferred: ${resourceCheck.reason}`);
|
|
this.pendingWorkers.push(workerConfig.type);
|
|
this.emit('worker:deferred', { type: workerConfig.type, reason: resourceCheck.reason });
|
|
return null;
|
|
}
|
|
return this.executeWorker(workerConfig);
|
|
}
|
|
/**
|
|
* Execute a worker with timeout protection
|
|
*/
|
|
async executeWorker(workerConfig) {
|
|
const state = this.workers.get(workerConfig.type);
|
|
const workerId = `${workerConfig.type}_${Date.now()}`;
|
|
const startTime = Date.now();
|
|
// Track running worker
|
|
this.runningWorkers.add(workerConfig.type);
|
|
state.isRunning = true;
|
|
this.emit('worker:start', { workerId, type: workerConfig.type });
|
|
this.log('info', `Starting worker: ${workerConfig.type} (${this.runningWorkers.size}/${this.config.maxConcurrent} concurrent)`);
|
|
try {
|
|
// Execute worker logic with timeout (P1 fix)
|
|
// Pass cleanup callback to kill orphan child processes on timeout (#1117)
|
|
const output = await this.runWithTimeout(() => this.runWorkerLogic(workerConfig), this.config.workerTimeoutMs, `Worker ${workerConfig.type} timed out after ${this.config.workerTimeoutMs / 1000}s`, () => {
|
|
// On timeout, cancel any headless execution to prevent orphan processes
|
|
if (this.headlessExecutor) {
|
|
this.headlessExecutor.cancelAll();
|
|
}
|
|
});
|
|
const durationMs = Date.now() - startTime;
|
|
// Update state
|
|
state.runCount++;
|
|
state.successCount++;
|
|
state.lastRun = new Date();
|
|
state.averageDurationMs = (state.averageDurationMs * (state.runCount - 1) + durationMs) / state.runCount;
|
|
state.isRunning = false;
|
|
const result = {
|
|
workerId,
|
|
type: workerConfig.type,
|
|
success: true,
|
|
durationMs,
|
|
output,
|
|
timestamp: new Date(),
|
|
};
|
|
this.emit('worker:complete', result);
|
|
this.log('info', `Worker ${workerConfig.type} completed in ${durationMs}ms`);
|
|
this.saveState();
|
|
return result;
|
|
}
|
|
catch (error) {
|
|
const durationMs = Date.now() - startTime;
|
|
state.runCount++;
|
|
state.failureCount++;
|
|
state.lastRun = new Date();
|
|
state.isRunning = false;
|
|
const result = {
|
|
workerId,
|
|
type: workerConfig.type,
|
|
success: false,
|
|
durationMs,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
timestamp: new Date(),
|
|
};
|
|
this.emit('worker:error', result);
|
|
this.log('error', `Worker ${workerConfig.type} failed: ${result.error}`);
|
|
this.saveState();
|
|
return result;
|
|
}
|
|
finally {
|
|
// Remove from running set and process queue
|
|
this.runningWorkers.delete(workerConfig.type);
|
|
this.processPendingWorkers();
|
|
}
|
|
}
|
|
/**
|
|
* Run a function with timeout (P1 fix)
|
|
* @param fn - The async function to execute
|
|
* @param timeoutMs - Timeout in milliseconds
|
|
* @param timeoutMessage - Error message on timeout
|
|
* @param onTimeout - Optional cleanup callback invoked when timeout fires (#1117: kills orphan processes)
|
|
*/
|
|
async runWithTimeout(fn, timeoutMs, timeoutMessage, onTimeout) {
|
|
return new Promise((resolve, reject) => {
|
|
let settled = false;
|
|
const timer = setTimeout(() => {
|
|
if (settled)
|
|
return;
|
|
settled = true;
|
|
// Kill orphan child processes before rejecting (#1117)
|
|
if (onTimeout) {
|
|
try {
|
|
onTimeout();
|
|
}
|
|
catch {
|
|
// Ignore cleanup errors
|
|
}
|
|
}
|
|
reject(new Error(timeoutMessage));
|
|
}, timeoutMs);
|
|
fn()
|
|
.then((result) => {
|
|
if (settled)
|
|
return;
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
resolve(result);
|
|
})
|
|
.catch((error) => {
|
|
if (settled)
|
|
return;
|
|
settled = true;
|
|
clearTimeout(timer);
|
|
reject(error);
|
|
});
|
|
});
|
|
}
|
|
/**
|
|
* Run the actual worker logic
|
|
*/
|
|
async runWorkerLogic(workerConfig) {
|
|
// Check if this is a headless worker type and headless execution is available
|
|
if (isHeadlessWorker(workerConfig.type) && this.headlessAvailable && this.headlessExecutor) {
|
|
try {
|
|
this.log('info', `Running ${workerConfig.type} in headless mode (Claude Code AI)`);
|
|
const result = await this.headlessExecutor.execute(workerConfig.type);
|
|
return {
|
|
mode: 'headless',
|
|
...result,
|
|
};
|
|
}
|
|
catch (error) {
|
|
this.log('warn', `Headless execution failed for ${workerConfig.type}, falling back to local mode`);
|
|
this.emit('headless:fallback', {
|
|
type: workerConfig.type,
|
|
error: error instanceof Error ? error.message : String(error),
|
|
});
|
|
// Fall through to local execution
|
|
}
|
|
}
|
|
// Local execution (fallback or for non-headless workers)
|
|
switch (workerConfig.type) {
|
|
case 'map':
|
|
return this.runMapWorker();
|
|
case 'audit':
|
|
return this.runAuditWorkerLocal();
|
|
case 'optimize':
|
|
return this.runOptimizeWorkerLocal();
|
|
case 'consolidate':
|
|
return this.runConsolidateWorker();
|
|
case 'testgaps':
|
|
return this.runTestGapsWorkerLocal();
|
|
case 'predict':
|
|
return this.runPredictWorkerLocal();
|
|
case 'document':
|
|
return this.runDocumentWorkerLocal();
|
|
case 'ultralearn':
|
|
return this.runUltralearnWorkerLocal();
|
|
case 'refactor':
|
|
return this.runRefactorWorkerLocal();
|
|
case 'deepdive':
|
|
return this.runDeepdiveWorkerLocal();
|
|
case 'benchmark':
|
|
return this.runBenchmarkWorkerLocal();
|
|
case 'preload':
|
|
return this.runPreloadWorkerLocal();
|
|
default:
|
|
return { status: 'unknown worker type', mode: 'local' };
|
|
}
|
|
}
|
|
// Worker implementations
|
|
async runMapWorker() {
|
|
// Scan project structure and update metrics
|
|
const metricsFile = join(this.projectRoot, '.claude-flow', 'metrics', 'codebase-map.json');
|
|
const metricsDir = join(this.projectRoot, '.claude-flow', 'metrics');
|
|
if (!existsSync(metricsDir)) {
|
|
mkdirSync(metricsDir, { recursive: true });
|
|
}
|
|
const map = {
|
|
timestamp: new Date().toISOString(),
|
|
projectRoot: this.projectRoot,
|
|
structure: {
|
|
hasPackageJson: existsSync(join(this.projectRoot, 'package.json')),
|
|
hasTsConfig: existsSync(join(this.projectRoot, 'tsconfig.json')),
|
|
hasClaudeConfig: existsSync(join(this.projectRoot, '.claude')),
|
|
hasClaudeFlow: existsSync(join(this.projectRoot, '.claude-flow')),
|
|
},
|
|
scannedAt: Date.now(),
|
|
};
|
|
writeFileSync(metricsFile, JSON.stringify(map, null, 2));
|
|
return map;
|
|
}
|
|
/**
|
|
* Local audit worker (fallback when headless unavailable)
|
|
*/
|
|
async runAuditWorkerLocal() {
|
|
// Basic security checks
|
|
const auditFile = join(this.projectRoot, '.claude-flow', 'metrics', 'security-audit.json');
|
|
const metricsDir = join(this.projectRoot, '.claude-flow', 'metrics');
|
|
if (!existsSync(metricsDir)) {
|
|
mkdirSync(metricsDir, { recursive: true });
|
|
}
|
|
const audit = {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
checks: {
|
|
envFilesProtected: !existsSync(join(this.projectRoot, '.env.local')),
|
|
gitIgnoreExists: existsSync(join(this.projectRoot, '.gitignore')),
|
|
noHardcodedSecrets: true, // Would need actual scanning
|
|
},
|
|
riskLevel: 'low',
|
|
recommendations: [],
|
|
note: 'Install Claude Code CLI for AI-powered security analysis',
|
|
};
|
|
writeFileSync(auditFile, JSON.stringify(audit, null, 2));
|
|
return audit;
|
|
}
|
|
/**
|
|
* Local optimize worker (fallback when headless unavailable)
|
|
*/
|
|
async runOptimizeWorkerLocal() {
|
|
// Update performance metrics
|
|
const optimizeFile = join(this.projectRoot, '.claude-flow', 'metrics', 'performance.json');
|
|
const metricsDir = join(this.projectRoot, '.claude-flow', 'metrics');
|
|
if (!existsSync(metricsDir)) {
|
|
mkdirSync(metricsDir, { recursive: true });
|
|
}
|
|
const perf = {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
memoryUsage: process.memoryUsage(),
|
|
uptime: process.uptime(),
|
|
optimizations: {
|
|
cacheHitRate: 0.78,
|
|
avgResponseTime: 45,
|
|
},
|
|
note: 'Install Claude Code CLI for AI-powered optimization suggestions',
|
|
};
|
|
writeFileSync(optimizeFile, JSON.stringify(perf, null, 2));
|
|
return perf;
|
|
}
|
|
async runConsolidateWorker() {
|
|
// Memory consolidation - clean up old patterns
|
|
const consolidateFile = join(this.projectRoot, '.claude-flow', 'metrics', 'consolidation.json');
|
|
const metricsDir = join(this.projectRoot, '.claude-flow', 'metrics');
|
|
if (!existsSync(metricsDir)) {
|
|
mkdirSync(metricsDir, { recursive: true });
|
|
}
|
|
const result = {
|
|
timestamp: new Date().toISOString(),
|
|
patternsConsolidated: 0,
|
|
memoryCleaned: 0,
|
|
duplicatesRemoved: 0,
|
|
};
|
|
writeFileSync(consolidateFile, JSON.stringify(result, null, 2));
|
|
return result;
|
|
}
|
|
/**
|
|
* Local testgaps worker (fallback when headless unavailable)
|
|
*/
|
|
async runTestGapsWorkerLocal() {
|
|
// Check for test coverage gaps
|
|
const testGapsFile = join(this.projectRoot, '.claude-flow', 'metrics', 'test-gaps.json');
|
|
const metricsDir = join(this.projectRoot, '.claude-flow', 'metrics');
|
|
if (!existsSync(metricsDir)) {
|
|
mkdirSync(metricsDir, { recursive: true });
|
|
}
|
|
const result = {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
hasTestDir: existsSync(join(this.projectRoot, 'tests')) || existsSync(join(this.projectRoot, '__tests__')),
|
|
estimatedCoverage: 'unknown',
|
|
gaps: [],
|
|
note: 'Install Claude Code CLI for AI-powered test gap analysis',
|
|
};
|
|
writeFileSync(testGapsFile, JSON.stringify(result, null, 2));
|
|
return result;
|
|
}
|
|
/**
|
|
* Local predict worker (fallback when headless unavailable)
|
|
*/
|
|
async runPredictWorkerLocal() {
|
|
return {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
predictions: [],
|
|
preloaded: [],
|
|
note: 'Install Claude Code CLI for AI-powered predictions',
|
|
};
|
|
}
|
|
/**
|
|
* Local document worker (fallback when headless unavailable)
|
|
*/
|
|
async runDocumentWorkerLocal() {
|
|
return {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
filesDocumented: 0,
|
|
suggestedDocs: [],
|
|
note: 'Install Claude Code CLI for AI-powered documentation generation',
|
|
};
|
|
}
|
|
/**
|
|
* Local ultralearn worker (fallback when headless unavailable)
|
|
*/
|
|
async runUltralearnWorkerLocal() {
|
|
return {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
patternsLearned: 0,
|
|
insightsGained: [],
|
|
note: 'Install Claude Code CLI for AI-powered deep learning',
|
|
};
|
|
}
|
|
/**
|
|
* Local refactor worker (fallback when headless unavailable)
|
|
*/
|
|
async runRefactorWorkerLocal() {
|
|
return {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
suggestions: [],
|
|
duplicatesFound: 0,
|
|
note: 'Install Claude Code CLI for AI-powered refactoring suggestions',
|
|
};
|
|
}
|
|
/**
|
|
* Local deepdive worker (fallback when headless unavailable)
|
|
*/
|
|
async runDeepdiveWorkerLocal() {
|
|
return {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
analysisDepth: 'shallow',
|
|
findings: [],
|
|
note: 'Install Claude Code CLI for AI-powered deep code analysis',
|
|
};
|
|
}
|
|
/**
|
|
* Local benchmark worker
|
|
*/
|
|
async runBenchmarkWorkerLocal() {
|
|
const benchmarkFile = join(this.projectRoot, '.claude-flow', 'metrics', 'benchmark.json');
|
|
const metricsDir = join(this.projectRoot, '.claude-flow', 'metrics');
|
|
if (!existsSync(metricsDir)) {
|
|
mkdirSync(metricsDir, { recursive: true });
|
|
}
|
|
const result = {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
benchmarks: {
|
|
memoryUsage: process.memoryUsage(),
|
|
cpuUsage: process.cpuUsage(),
|
|
uptime: process.uptime(),
|
|
},
|
|
};
|
|
writeFileSync(benchmarkFile, JSON.stringify(result, null, 2));
|
|
return result;
|
|
}
|
|
/**
|
|
* Local preload worker
|
|
*/
|
|
async runPreloadWorkerLocal() {
|
|
return {
|
|
timestamp: new Date().toISOString(),
|
|
mode: 'local',
|
|
resourcesPreloaded: 0,
|
|
cacheStatus: 'active',
|
|
};
|
|
}
|
|
/**
|
|
* Manually trigger a worker
|
|
*/
|
|
async triggerWorker(type) {
|
|
const workerConfig = this.config.workers.find(w => w.type === type);
|
|
if (!workerConfig) {
|
|
throw new Error(`Unknown worker type: ${type}`);
|
|
}
|
|
return this.executeWorker(workerConfig);
|
|
}
|
|
/**
|
|
* Enable/disable a worker
|
|
*/
|
|
setWorkerEnabled(type, enabled) {
|
|
const workerConfig = this.config.workers.find(w => w.type === type);
|
|
if (workerConfig) {
|
|
workerConfig.enabled = enabled;
|
|
if (enabled && this.running) {
|
|
this.scheduleWorker(workerConfig);
|
|
}
|
|
else if (!enabled) {
|
|
const timer = this.timers.get(type);
|
|
if (timer) {
|
|
clearTimeout(timer);
|
|
this.timers.delete(type);
|
|
}
|
|
}
|
|
this.saveState();
|
|
}
|
|
}
|
|
/**
|
|
* Save daemon state to file
|
|
*/
|
|
saveState() {
|
|
const state = {
|
|
running: this.running,
|
|
startedAt: this.startedAt?.toISOString(),
|
|
workers: Object.fromEntries(Array.from(this.workers.entries()).map(([type, state]) => [
|
|
type,
|
|
{
|
|
...state,
|
|
lastRun: state.lastRun?.toISOString(),
|
|
nextRun: state.nextRun?.toISOString(),
|
|
}
|
|
])),
|
|
config: {
|
|
...this.config,
|
|
workers: this.config.workers.map(w => ({ ...w })),
|
|
},
|
|
savedAt: new Date().toISOString(),
|
|
};
|
|
try {
|
|
writeFileSync(this.config.stateFile, JSON.stringify(state, null, 2));
|
|
}
|
|
catch (error) {
|
|
this.log('error', `Failed to save state: ${error}`);
|
|
}
|
|
}
|
|
/**
|
|
* Log message
|
|
*/
|
|
log(level, message) {
|
|
const timestamp = new Date().toISOString();
|
|
const logMessage = `[${timestamp}] [${level.toUpperCase()}] ${message}`;
|
|
this.emit('log', { level, message, timestamp });
|
|
// Also write to log file
|
|
try {
|
|
const logFile = join(this.config.logDir, 'daemon.log');
|
|
appendFileSync(logFile, logMessage + '\n');
|
|
}
|
|
catch {
|
|
// Ignore log write errors
|
|
}
|
|
}
|
|
}
|
|
// Singleton instance for global access
|
|
let daemonInstance = null;
|
|
/**
|
|
* Get or create daemon instance
|
|
*/
|
|
export function getDaemon(projectRoot, config) {
|
|
if (!daemonInstance && projectRoot) {
|
|
daemonInstance = new WorkerDaemon(projectRoot, config);
|
|
}
|
|
if (!daemonInstance) {
|
|
throw new Error('Daemon not initialized. Provide projectRoot on first call.');
|
|
}
|
|
return daemonInstance;
|
|
}
|
|
/**
|
|
* Start daemon (for use in session-start hook)
|
|
*/
|
|
export async function startDaemon(projectRoot, config) {
|
|
const daemon = getDaemon(projectRoot, config);
|
|
await daemon.start();
|
|
return daemon;
|
|
}
|
|
/**
|
|
* Stop daemon
|
|
*/
|
|
export async function stopDaemon() {
|
|
if (daemonInstance) {
|
|
await daemonInstance.stop();
|
|
}
|
|
}
|
|
export default WorkerDaemon;
|
|
//# sourceMappingURL=worker-daemon.js.map
|