diff options
Diffstat (limited to 'modules/core/module-orchestrator.js')
| -rw-r--r-- | modules/core/module-orchestrator.js | 132 |
1 files changed, 96 insertions, 36 deletions
diff --git a/modules/core/module-orchestrator.js b/modules/core/module-orchestrator.js index 5cff89e..ad9f020 100644 --- a/modules/core/module-orchestrator.js +++ b/modules/core/module-orchestrator.js @@ -8,6 +8,10 @@ class OrchestratorModule { this.terminal = null; this._agents = {}; this._SPAWN_LIMIT = 5; + // Boot-time agents — stable per-role ids so a reload re-adopts the existing VFS dir + // instead of minting a new timestamp id every boot (that was the /agents/ accumulation bug). + this._DEFAULT_ROLES = ['auto-learner', 'kernel-architect', 'searcher']; + this._BOOT_DELAY = 2000; // let other modules init before the boot team spawns (tests set 0) this._SPAWN_STRATEGY = 'eager'; this._HEARTBEAT_TIMEOUT = 15000; this._IDLE_TIMEOUT = 120000; @@ -174,8 +178,10 @@ class OrchestratorModule { if (registry) { registry.registerAgent(['register', 'orchestrator-0', 'delegate', 'spawn', 'monitor', 'purge']); } - // Reanimate sleeping agents from VFS - this._reanimateFromVFS(); + // Reconcile + pre-spawn the default team. VirtualFileSystem.loadFromStorage() is ASYNC, + // so reconciliation MUST wait for it — running it synchronously saw an empty root and + // silently did nothing, which let duplicate agents accumulate on every reload. + this._bootAgents(); // A2A mailbox polling loop (500ms) if (this._A2A_POLLING) { this._a2aPollTimer = this.terminal.moduleManager._setInterval(() => this._pollAgentMailboxes(), this._A2A_POLL_INTERVAL); @@ -187,8 +193,6 @@ class OrchestratorModule { this.terminal.print('🧠 Orchestrator loaded — A2A mailbox transport active', 'success'); // Load recurring schedules from VFS this._loadSchedules(); - // Pre-spawn default agents after a short delay (let other modules init first) - setTimeout(() => this._preSpawnDefaultAgents(), 2000); // Subscribe to event bus for in-memory A2A delivery (instant, no VFS poll) this._on('a2a:message', (msg) => { if (msg.type === 'heartbeat' && msg.from && this._agents[msg.from]) { @@ -334,7 +338,7 @@ class OrchestratorModule { _sanitizeId(name) { return (name || 'worker').replace(/[^a-zA-Z0-9_-]/g, '-').replace(/-+/g, '-').replace(/^-|-$/g, '').substring(0, 32) || 'worker'; } - spawnAgent(role, capabilities, modelName = null) { + spawnAgent(role, capabilities, modelName = null, idHint = null) { if (this._spawnLock) { this.terminal.print('⚠️ Spawn in progress — queued', 'warning'); return null; @@ -342,7 +346,12 @@ class OrchestratorModule { this._spawnLock = true; try { const safeRole = this._sanitizeId(role); - const id = `agent-${Date.now().toString(36)}-${Math.random().toString(36).substr(2, 4)}-${safeRole}`; + const id = idHint || `agent-${Date.now().toString(36)}-${Math.random().toString(36).substr(2, 4)}-${safeRole}`; + // Idempotent: a stable-id agent that is already live is reused, never duplicated + if (this._agents[id]) { + this.terminal.print(`♻️ Agent '${id}' already live — reusing`, 'info'); + return this._agents[id]; + } if (Object.keys(this._agents).length >= this._SPAWN_LIMIT) { this.terminal.print(`❌ Agent spawn limit (${this._SPAWN_LIMIT}) reached`, 'error'); return null; @@ -400,8 +409,10 @@ class OrchestratorModule { registry.a2aUnregister(id); } } - // Clean up VFS - this.terminal.fs.deleteFile('agents/' + id); + // Clean up VFS — rm() dispatches file|dir → trash (deleteFile() is file-only + // and silently returns false for directories, which leaked /agents/<id>/) + const removed = this.terminal.fs.rm('agents/' + id); + if (!removed) this.terminal.print(`⚠️ VFS cleanup failed for '${id}' — /agents/${id}/ still present`, 'warning'); delete this._agents[id]; delete this._restartBackoffs[id]; this.terminal.print(`💀 Agent '${id}' killed`, 'warning'); @@ -439,31 +450,79 @@ class OrchestratorModule { this.terminal.print(`🔄 Agent '${id}' reanimated from VFS sleep`, 'success'); } + async _bootAgents() { + // Boot sequence for /agents/ — order is load-bearing: + // 1. await VFS hydration (loadFromStorage is async; a sync reconcile sees an empty root) + // 2. reconcile /agents/ (adopt what is already there, purge duplicates/orphans) + // 3. only then spawn any missing default agent + try { + await Promise.resolve(this.terminal.fs._ready); + } catch (e) { + this.terminal.print('⚠️ VFS hydration failed — agent reconciliation skipped', 'warning'); + return this._agents; + } + this._reanimateFromVFS(); + if (this._BOOT_DELAY > 0) await new Promise(r => setTimeout(r, this._BOOT_DELAY)); + this._preSpawnDefaultAgents(); + return this._agents; + } + _reanimateFromVFS() { + // Boot reconciliation for /agents/: + // 1. sleep.json → reanimate the agent (previous behaviour) + // 2. status.json + default role → ADOPT it (a reload left it behind, still wanted) + // 3. no usable state / duplicate → orphan → fs.rm(), so a reload cannot grow /agents/ + // Only `agent-*` dirs are touched; /agents/register/ belongs to AgentRegistry. const fs = this.terminal.fs; const agentsDir = fs.readDir('/agents/'); if (!agentsDir) return; - let reanimated = 0; + + // Stable ids (agent-<role>) beat timestamped ones; among timestamped ids, newest wins. + const tsOf = (id) => { + const m = /^agent-([0-9a-z]+)-[0-9a-z]{4}-/.exec(id); + return m ? (parseInt(m[1], 36) || 0) : Number.MAX_SAFE_INTEGER; + }; + const keep = new Map(); // role:<role> for default agents (singleton) | id:<id> for workers + const orphans = []; + for (const entry of agentsDir) { - if (entry.type !== 'dir') continue; - const sleepData = fs.fsReadJSON(`/agents/${entry.name}/sleep.json`); - if (sleepData) { - const agent = { - ...sleepData, - status: 'idle', - currentTask: null, - leaseId: null, - lastHeartbeat: Date.now() - }; - delete agent._sleepTime; - this._agents[entry.name] = agent; - fs.writeFile(`agents/${entry.name}/status.json`, JSON.stringify({ ...agent, status: 'idle' })); - reanimated++; + if (entry.type !== 'dir' || !entry.name.startsWith('agent-')) continue; + const id = entry.name; + if (this._agents[id]) continue; + const sleepData = fs.fsReadJSON(`/agents/${id}/sleep.json`); + const statusData = fs.fsReadJSON(`/agents/${id}/status.json`); + const state = sleepData || null; + const role = (state || statusData || {}).role; + const wanted = !!state || (statusData && this._DEFAULT_ROLES.includes(role)); + if (!wanted) { orphans.push(id); continue; } + // Default roles are singletons (one dir per role); worker agents stay unique per id. + const key = this._DEFAULT_ROLES.includes(role) ? 'role:' + role : 'id:' + id; + const prev = keep.get(key); + if (!prev || tsOf(id) > tsOf(prev.id)) { + if (prev) orphans.push(prev.id); + keep.set(key, { id, state: state || statusData }); + } else { + orphans.push(id); } } - if (reanimated > 0) { - this.terminal.print(`🔄 Reanimated ${reanimated} agent(s) from VFS sleep`, 'info'); + + for (const { id, state } of keep.values()) { + const agent = { + ...state, + status: 'idle', + currentTask: null, + leaseId: null, + lastHeartbeat: Date.now() + }; + delete agent._sleepTime; + this._agents[id] = agent; + fs.writeFile(`agents/${id}/status.json`, JSON.stringify({ ...agent, status: 'idle' })); + if (fs.readFile(`/agents/${id}/sleep.json`)) fs.rm(`agents/${id}/sleep.json`); } + for (const id of orphans) fs.rm('agents/' + id); // rm() dispatches dir → trashDir + + if (keep.size > 0) this.terminal.print(`🔄 Restored ${keep.size} agent(s) from VFS`, 'info'); + if (orphans.length > 0) this.terminal.print(`🧹 Purged ${orphans.length} orphaned agent dir(s)`, 'warning'); } // ─── Restart Policies ─── @@ -1284,22 +1343,23 @@ class OrchestratorModule { } _preSpawnDefaultAgents() { - // Spawn 3 role-specific agents at boot + // Spawn the default team ONCE per browser profile. Stable ids (`agent-<role>`) mean a + // reload re-adopts the existing dir in _reanimateFromVFS() instead of writing a new one. var roles = [ - { role: 'auto-learner', caps: ['list_files', 'read_file', 'write_file', 'view_module'], task: 'audit skills' }, - { role: 'kernel-architect', caps: ['list_files', 'read_file', 'write_file', 'run_command'], task: 'index VFS' }, - { role: 'searcher', caps: ['list_files', 'read_file', 'web_search'], task: 'check stale data' } + { role: 'auto-learner', caps: ['list_files', 'read_file', 'write_file', 'view_module'] }, + { role: 'kernel-architect', caps: ['list_files', 'read_file', 'write_file', 'run_command'] }, + { role: 'searcher', caps: ['list_files', 'read_file', 'web_search'] } ]; for (var i = 0; i < roles.length; i++) { var r = roles[i]; - // Check if an agent with this role already exists + // Already live (fresh spawn this boot, or adopted from VFS) → just keep it alive var existing = Object.values(this._agents).find(function(a) { return a.role === r.role; }); - if (!existing) { - var agent = this.spawnAgent(r.role, r.caps); - if (agent) { - agent.restartPolicy = 'always'; - this.terminal.print('🤖 Pre-spawned agent: ' + r.role + ' (' + r.caps.length + ' caps)', 'info'); - } + if (existing) { existing.restartPolicy = 'always'; continue; } + var id = 'agent-' + this._sanitizeId(r.role); + var agent = this.spawnAgent(r.role, r.caps, null, id); + if (agent) { + agent.restartPolicy = 'always'; + this.terminal.print('🤖 Pre-spawned agent: ' + r.role + ' (' + r.caps.length + ' caps)', 'info'); } } } |