Sync: Deployed Dynamic Agent Onboarding logic to Space

logos/agent_dispatcher.py

@@ -347,6 +347,7 @@ class LogosSwarm:
     """
     Protocol 21: The Async Interference Bus.
     Agents pulse in parallel; the solution is the geometric intersection.
+    Protocol 26: Dynamic Agent Onboarding (Matroska Router Connection).
     """
     def __init__(self, base_url="http://localhost:1234/v1", model="google/gemma-3-4b"):
         self.connector = get_connector('local', base_url=base_url, model=model)
@@ -359,6 +360,39 @@ class LogosSwarm:
         self.oversight = DolphinOversight(self.state)
         self.metadata = GemmaMetadata()
         self.routing = RNJ1Routing(self.state, self.connector)
+        
+        # Dynamic Agent Registry
+        self.agents: Dict[str, Any] = {}
+        self.discover_agents()
+
+    def discover_agents(self):
+        """
+        Onboards agents from logos/agents/ that implement BaseAgent.
+        """
+        import pkgutil
+        import importlib
+        import inspect
+        import logos.agents
+        from logos.agents.base_agent import BaseAgent
+
+        package = logos.agents
+        prefix = package.__name__ + "."
+
+        print("🔍 [SWARM] Scanning for Matroska Agents...")
+        for _, name, _ in pkgutil.iter_modules(package.__path__, prefix):
+            try:
+                module = importlib.import_module(name)
+                for name_cls, cls in inspect.getmembers(module, inspect.isclass):
+                    if issubclass(cls, BaseAgent) and cls is not BaseAgent:
+                        try:
+                            instance = cls()
+                            self.agents[instance.name.lower()] = instance
+                            print(f"   ✅ Onboarded: {instance.name} (Trigger: {instance.triggers})")
+                        except Exception as e:
+                            print(f"   ⚠️ Failed to init {name_cls}: {e}")
+            except Exception as e:
+                print(f"   ⚠️ Error loading module {name}: {e}")
+
 
     async def process(self, raw_input: str):
         print("\n🌊 [HOLOGRAPHIC BUS] PULSING WAVE FUNCTIONS...")
@@ -368,27 +402,41 @@ class LogosSwarm:
         user_packet = {"content": raw_input, "type": "text_command"}
         action, packet = self.oversight.ingest(user_packet)
         
-        # 2. PARALLEL INTERFERENCE: Gemma (Semantic), RNJ-1 (Logic), and Gemini (Reasoning)
+        # 2. PARALLEL INTERFERENCE: Gemma, RNJ-1, Reasoning, AND Dynamic Agents
         # They "interfere" to find the true deterministic coordinate
         gemma_task = asyncio.create_task(self.metadata.process(packet))
         rnj1_task = asyncio.create_task(self.routing.calculate_position(packet))
         
-        # reasoning_task: Pulse to LLM for high-level topological insights
+        # reasoning_task: Pulse to LLM
         reasoning_task = asyncio.create_task(self.routing.connector.chat_async(
             f"Analyze this packet for topological resonance: {raw_input}",
             system_prompt="You are the SWARM_REASONER. Identify the high-level intent and potential manifold collisions."
         ))
-        
-        # Wait for the High-Frequency Triple Wave to align
+
+        # Dynamic Agent Task (if applicable)
+        agent_tasks = []
+        for agent_name, agent in self.agents.items():
+            # Check triggers or intent
+            if any(t in raw_input.lower() for t in agent.triggers):
+                print(f"🚀 [SWARM] Triggering Agent: {agent.name}")
+                agent_tasks.append(asyncio.create_task(agent.process(packet)))
+
+        # Wait for Core Triple Wave
         await asyncio.gather(gemma_task, rnj1_task, reasoning_task)
         
         gemma_status, mass = gemma_task.result()
         res = rnj1_task.result()
         reasoning, logprobs = reasoning_task.result()
         
-        # Update Entropy Kill Switch with reasoning wave telemetry
+        # Process Agent Results
+        agent_results = []
+        if agent_tasks:
+            await asyncio.gather(*agent_tasks)
+            for t in agent_tasks:
+                agent_results.append(t.result())
+        
+        # Update Entropy Kill Switch
         if logprobs:
-            # logprobs from OpenAI usually contain 'content' if it's chat/completion
             log_content = logprobs.get('content', [])
             self.oversight.kill_switch.monitor_bulk(log_content)
         
@@ -472,15 +520,47 @@ class LogosSwarm:
             })
         except: pass
 
-ROUTER_PROMPT = """
+
+def get_onboarded_agents() -> Dict[str, Any]:
+    """Helper to discover agents dynamically."""
+    import pkgutil
+    import importlib
+    import inspect
+    import logos.agents
+    from logos.agents.base_agent import BaseAgent
+    
+    agents = {}
+    package = logos.agents
+    prefix = package.__name__ + "."
+    
+    for _, name, _ in pkgutil.iter_modules(package.__path__, prefix):
+        try:
+            module = importlib.import_module(name)
+            for _, cls in inspect.getmembers(module, inspect.isclass):
+                if issubclass(cls, BaseAgent) and cls is not BaseAgent:
+                    try:
+                        instance = cls()
+                        agents[instance.name.lower()] = instance
+                    except: pass
+        except: pass
+    return agents
+
+# Dynamic Prompt Construction
+_onboarded = get_onboarded_agents()
+_agent_lines = []
+for name, agent in _onboarded.items():
+    _agent_lines.append(f"- '{name}': {agent.description}")
+
+ROUTER_PROMPT = f"""
 Analyze the following user input and classify it into one of these categories:
 - 'fractal_architect': For code, debugging, storage, sharding, or technical implementation.
 - 'prime_navigator': For math, primes, geometry, calculations, or physics.
 - 'creative': For descriptions, stories, or documentation style.
 - 'general': For greetings or anything else.
+{chr(10).join(_agent_lines)}
 
 Reply with ONLY a JSON object in this format:
-{"intent": "category_name", "confidence": 0.9}
+{{"intent": "category_name", "confidence": 0.9}}
 Do not write any introductory text.
 """
 

logos/agents/base_agent.py

@@ -0,0 +1,38 @@
+
+from abc import ABC, abstractmethod
+from typing import Dict, Any, List, Optional
+
+class BaseAgent(ABC):
+    """
+    Protocol 26: Standard Agent Interface for Matroska Swarm.
+    All agents must implement this to be auto-discovered by the Router.
+    """
+    
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Unique name of the agent (e.g., 'VideoAtomizer')"""
+        pass
+        
+    @property
+    @abstractmethod
+    def description(self) -> str:
+        """Short description for the Router's system prompt"""
+        pass
+        
+    @property
+    @abstractmethod
+    def triggers(self) -> List[str]:
+        """List of keywords or regex patterns that trigger this agent"""
+        pass
+        
+    @abstractmethod
+    async def process(self, task: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Execute the agent's logic.
+        Args:
+            task: The input packet (e.g., {'content': '...', 'context': {}})
+        Returns:
+            Dict containing 'status', 'result', and 'tensor_updates'
+        """
+        pass

logos/agents/video_atomizer.py

@@ -1,15 +1,43 @@
 import re
 import asyncio
 from youtube_transcript_api import YouTubeTranscriptApi
+from logos.agents.base_agent import BaseAgent
 
-class VideoAtomizer:
+class VideoAtomizer(BaseAgent):
     """
     Role: V-NODE (Video Ingest)
     Function: Rips semantic atoms from video streams and collides them
               with the existing Project Manifold.
     """
+    @property
+    def name(self) -> str:
+        return "VideoAtomizer"
+
+    @property
+    def description(self) -> str:
+        return "Ingests YouTube videos/playlists, extracts transcripts, and atomizes content into semantic tensors."
+
+    @property
+    def triggers(self) -> list:
+        return ["youtube", "playlist", "video", "transcript", "watch?v="]
+
+    async def process(self, task: dict) -> dict:
+        content = task.get('content', '')
+        project_dna = task.get('context', {}).get('dna', {})
+        
+        # Extract URLs from content
+        urls = re.findall(r'(https?://(?:www\.|m\.)?youtube\.com/watch\?v=[\w-]+|https?://youtu\.be/[\w-]+)', content)
+        
+        results = []
+        for url in urls:
+            res = await self.ingest_and_align(url, project_dna)
+            results.append(res)
+            
+        return {"status": "COMPLETE", "result": results}
+
     def __init__(self):
-        self.name = "VideoAtomizer"
+        self._name = "VideoAtomizer" # Internal generic
+
 
     def extract_video_id(self, url):
         # Extracts 'R9czY1uVq_k' from the URL
@@ -31,10 +59,30 @@ class VideoAtomizer:
         
         # 1. FETCH TRANSCRIPT (The Raw Atoms)
         try:
-            # We use a thread pool for the blocking API call
+            # Instantiate the API wrapper (Local Environment Quirk)
+            yt_api = YouTubeTranscriptApi()
+            
+            # Use 'fetch' method in thread executor
             loop = asyncio.get_event_loop()
-            transcript_list = await loop.run_in_executor(None, YouTubeTranscriptApi.get_transcript, video_id)
-            full_text = " ".join([t['text'] for t in transcript_list])
+            transcript_list = await loop.run_in_executor(None, yt_api.fetch, video_id)
+            
+            # Handle return types
+            if isinstance(transcript_list, dict):
+                full_text = str(transcript_list)
+            elif isinstance(transcript_list, list):
+                if len(transcript_list) > 0:
+                    first = transcript_list[0]
+                    if isinstance(first, str):
+                        full_text = " ".join(transcript_list)
+                    elif isinstance(first, dict):
+                        full_text = " ".join([t.get('text', '') for t in transcript_list])
+                    else:
+                        full_text = str(transcript_list)
+                else:
+                    full_text = ""
+            else:
+                full_text = str(transcript_list)
+                
         except Exception as e:
             return {"error": f"Signal Lost: {e}"}