diff --git a/AgentSystem/modules/__init__.py b/AgentSystem/modules/__init__.py index 74d90ce..4b24f15 100644 --- a/AgentSystem/modules/__init__.py +++ b/AgentSystem/modules/__init__.py @@ -4,15 +4,11 @@ Contains optional modules that extend agent capabilities """ -from .browser import BrowserModule -from .code_editor import CodeEditor -from .continuous_learning import ContinuousLearningModule -from .email import EmailModule -from .knowledge_graph import KnowledgeGraphModule -from .sensory_input import SensoryInputModule -from .system_interface import SystemInterfaceModule - __all__ = [ + 'AgentForgeRegistry', + 'AgentForgeSDK', + 'KnowledgeExchange', + 'ModuleDescriptor', 'BrowserModule', 'CodeEditor', 'ContinuousLearningModule', @@ -21,3 +17,40 @@ 'SensoryInputModule', 'SystemInterfaceModule' ] + + +def __getattr__(name): # pragma: no cover - thin import helper + if name == 'AgentForgeRegistry': + from .agent_forge import AgentForgeRegistry + return AgentForgeRegistry + if name == 'AgentForgeSDK': + from .agent_forge import AgentForgeSDK + return AgentForgeSDK + if name == 'KnowledgeExchange': + from .agent_forge import KnowledgeExchange + return KnowledgeExchange + if name == 'ModuleDescriptor': + from .agent_forge import ModuleDescriptor + return ModuleDescriptor + if name == 'BrowserModule': + from .browser import BrowserModule + return BrowserModule + if name == 'CodeEditor': + from .code_editor import CodeEditor + return CodeEditor + if name == 'ContinuousLearningModule': + from .continuous_learning import ContinuousLearningModule + return ContinuousLearningModule + if name == 'EmailModule': + from .email import EmailModule + return EmailModule + if name == 'KnowledgeGraphModule': + from .knowledge_graph import KnowledgeGraphModule + return KnowledgeGraphModule + if name == 'SensoryInputModule': + from .sensory_input import SensoryInputModule + return SensoryInputModule + if name == 'SystemInterfaceModule': + from .system_interface import SystemInterfaceModule + return SystemInterfaceModule + raise AttributeError(name) diff --git a/AgentSystem/modules/agent_forge.py b/AgentSystem/modules/agent_forge.py new file mode 100644 index 0000000..690af57 --- /dev/null +++ b/AgentSystem/modules/agent_forge.py @@ -0,0 +1,214 @@ +"""AgentForge Developer Framework. + +Provides registry and SDK helpers so developers can share and reuse +modules, datasets, and knowledge packages across AgentSystem deployments. +""" + +from __future__ import annotations + +import json +import threading +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Sequence + +from AgentSystem.utils.logger import get_logger + +logger = get_logger("modules.agent_forge") + + +@dataclass(frozen=True) +class ModuleDescriptor: + """Metadata describing a distributable AgentForge module.""" + + name: str + version: str + summary: str + author: str = "unknown" + capabilities: Sequence[str] = () + tags: Sequence[str] = () + + def key(self) -> str: + return f"{self.name}:{self.version}" + + def __post_init__(self) -> None: + object.__setattr__(self, "capabilities", tuple(self.capabilities)) + object.__setattr__(self, "tags", tuple(self.tags)) + + +class AgentForgeRegistry: + """Lightweight registry for AgentForge modules. + + The registry persists module descriptors to disk so deployments can + synchronise available capabilities across environments without + requiring an external service. + """ + + def __init__(self, storage_path: Optional[Path] = None) -> None: + self._lock = threading.Lock() + self.storage_path = storage_path or Path(".agentforge") + self.storage_path.mkdir(parents=True, exist_ok=True) + self._registry_file = self.storage_path / "registry.json" + self._modules: Dict[str, ModuleDescriptor] = {} + self._load() + + def _load(self) -> None: + if not self._registry_file.exists(): + return + try: + data = json.loads(self._registry_file.read_text()) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("Failed to load AgentForge registry: %s", exc) + return + for payload in data.get("modules", []): + try: + descriptor = ModuleDescriptor(**payload) + except TypeError as exc: + logger.debug("Skipping invalid module payload %s: %s", payload, exc) + continue + self._modules[descriptor.key()] = descriptor + + def _persist(self) -> None: + payload = {"modules": [asdict(item) for item in self._modules.values()]} + try: + self._registry_file.write_text(json.dumps(payload, indent=2, sort_keys=True)) + except OSError as exc: + logger.error("Failed to persist AgentForge registry: %s", exc) + + def register(self, descriptor: ModuleDescriptor, *, overwrite: bool = False) -> ModuleDescriptor: + """Register a module in the registry. + + Args: + descriptor: Module metadata to persist. + overwrite: Allow replacement when the key already exists. + """ + + key = descriptor.key() + with self._lock: + if not overwrite and key in self._modules: + raise ValueError(f"Module {key} already registered") + self._modules[key] = descriptor + self._persist() + logger.debug("Registered AgentForge module %s", key) + return descriptor + + def list_modules(self, *, tag: Optional[str] = None) -> List[ModuleDescriptor]: + modules = list(self._modules.values()) + if tag is None: + return modules + return [item for item in modules if tag in item.tags] + + def get(self, name: str, version: Optional[str] = None) -> Optional[ModuleDescriptor]: + if version: + return self._modules.get(f"{name}:{version}") + # Return latest version lexicographically if not specified + candidates = [item for item in self._modules.values() if item.name == name] + if not candidates: + return None + return sorted(candidates, key=lambda item: item.version)[-1] + + +class KnowledgeExchange: + """Knowledge sharing surface for AgentForge deployments.""" + + def __init__(self, storage_path: Optional[Path] = None) -> None: + self._lock = threading.Lock() + self.storage_path = storage_path or Path(".agentforge") + self.storage_path.mkdir(parents=True, exist_ok=True) + self._exchange_file = self.storage_path / "knowledge.json" + self._entries: List[Dict[str, Any]] = [] + self._load() + + def _load(self) -> None: + if not self._exchange_file.exists(): + return + try: + data = json.loads(self._exchange_file.read_text()) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("Failed to load AgentForge knowledge exchange: %s", exc) + return + self._entries = [entry for entry in data.get("entries", []) if isinstance(entry, dict)] + + def _persist(self) -> None: + payload = {"entries": self._entries} + try: + self._exchange_file.write_text(json.dumps(payload, indent=2, sort_keys=True)) + except OSError as exc: + logger.error("Failed to persist AgentForge knowledge exchange: %s", exc) + + def publish(self, *, title: str, content: str, authors: Sequence[str], tags: Sequence[str] = (), + metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + record = { + "title": title, + "content": content, + "authors": list(authors), + "tags": list(tags), + "metadata": dict(metadata or {}), + } + with self._lock: + self._entries.append(record) + self._persist() + logger.debug("Published knowledge artifact %s", title) + return record + + def query(self, *, tag: Optional[str] = None, limit: Optional[int] = None) -> List[Dict[str, Any]]: + matches = self._entries if tag is None else [entry for entry in self._entries if tag in entry["tags"]] + if limit is None: + return list(matches) + return list(matches[:limit]) + + +class AgentForgeSDK: + """High-level helper combining registry and knowledge exchange.""" + + def __init__( + self, + *, + registry: Optional[AgentForgeRegistry] = None, + exchange: Optional[KnowledgeExchange] = None, + knowledge_manager: Optional[Any] = None, + ) -> None: + self.registry = registry or AgentForgeRegistry() + self.exchange = exchange or KnowledgeExchange() + self.knowledge_manager = knowledge_manager + + def publish_module(self, descriptor: ModuleDescriptor, *, overwrite: bool = False) -> ModuleDescriptor: + return self.registry.register(descriptor, overwrite=overwrite) + + def share_knowledge( + self, + title: str, + content: str, + *, + authors: Sequence[str], + tags: Sequence[str] = (), + metadata: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + entry = self.exchange.publish( + title=title, + content=content, + authors=authors, + tags=tags, + metadata=metadata, + ) + if self.knowledge_manager and tags: + for tag in tags: + fact = f"Knowledge artifact '{title}' tagged with {tag}" + try: + self.knowledge_manager.add_fact(fact, source="agentforge", category=tag) + except Exception as exc: # pragma: no cover - defensive logging + logger.warning("Failed to sync knowledge artifact %s into knowledge base: %s", title, exc) + return entry + + def bootstrap(self, modules: Iterable[ModuleDescriptor]) -> None: + for descriptor in modules: + try: + self.registry.register(descriptor, overwrite=False) + except ValueError: + continue + + def fetch_modules(self, *, tag: Optional[str] = None) -> List[ModuleDescriptor]: + return self.registry.list_modules(tag=tag) + + def retrieve_knowledge(self, *, tag: Optional[str] = None, limit: Optional[int] = None) -> List[Dict[str, Any]]: + return self.exchange.query(tag=tag, limit=limit) diff --git a/AgentSystem/modules/knowledge_manager.py b/AgentSystem/modules/knowledge_manager.py index d0e96a4..d506ac3 100644 --- a/AgentSystem/modules/knowledge_manager.py +++ b/AgentSystem/modules/knowledge_manager.py @@ -13,8 +13,14 @@ import sqlite3 import os import time -import numpy as np +import json +from pathlib import Path from typing import Dict, List, Any, Optional + +try: + import numpy as np # type: ignore +except ImportError: + np = None # type: ignore from datetime import datetime, timedelta # Optional PostgreSQL support with graceful fallback try: @@ -26,7 +32,10 @@ RealDictCursor = None POSTGRES_AVAILABLE = False # RealDictCursor import moved to conditional block above -import psutil +try: + import psutil # type: ignore +except ImportError: + psutil = None # type: ignore import subprocess from AgentSystem.utils.logger import get_logger @@ -58,6 +67,10 @@ def __init__(self, db_path: Optional[str] = None, use_postgres: bool = False, self.max_retries = 3 self.retry_delay = 1.0 self.ram_limit = 20.0 # GB - Increased for testing + # Memory system tuning + self.memory_decay_half_life = 60 * 60 * 24 # one day default + self.minimum_salience = 0.05 + self.consolidation_threshold = 0.8 self.init_database() def monitor_ram(self, max_usage: float = None) -> bool: @@ -65,7 +78,10 @@ def monitor_ram(self, max_usage: float = None) -> bool: if max_usage is None: max_usage = self.ram_limit - ram_usage = psutil.virtual_memory().used / (1024 ** 3) # GB + if psutil: + ram_usage = psutil.virtual_memory().used / (1024 ** 3) # GB + else: + ram_usage = 0.0 if ram_usage > max_usage: logger.warning(f"RAM usage {ram_usage:.2f}GB exceeds limit {max_usage}GB") return False @@ -115,7 +131,7 @@ def init_database(self) -> None: embedding BYTEA ) ''') - + cursor.execute(''' CREATE TABLE IF NOT EXISTS documents ( id SERIAL PRIMARY KEY, @@ -128,12 +144,38 @@ def init_database(self) -> None: last_accessed TIMESTAMP ) ''') - + + cursor.execute(''' + CREATE TABLE IF NOT EXISTS episodic_memories ( + id SERIAL PRIMARY KEY, + event TEXT NOT NULL, + outcome TEXT, + emotion TEXT, + salience REAL DEFAULT 0.5, + context JSONB, + timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + embedding BYTEA, + consolidated BOOLEAN DEFAULT FALSE + ) + ''') + + cursor.execute(''' + CREATE TABLE IF NOT EXISTS source_trust ( + source TEXT PRIMARY KEY, + score REAL DEFAULT 0.5, + success_count INTEGER DEFAULT 0, + failure_count INTEGER DEFAULT 0, + last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + ''') + # Create indexes for performance cursor.execute('CREATE INDEX IF NOT EXISTS idx_facts_category ON facts(category)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_facts_timestamp ON facts(timestamp)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_facts_confidence ON facts(confidence)') - + cursor.execute('CREATE INDEX IF NOT EXISTS idx_epi_salience ON episodic_memories(salience)') + cursor.execute('CREATE INDEX IF NOT EXISTS idx_epi_timestamp ON episodic_memories(timestamp)') + else: # Local SQLite for caching on Pi 5 self.conn = sqlite3.connect(self.db_path) @@ -167,11 +209,37 @@ def init_database(self) -> None: last_accessed DATETIME ) ''') - + + cursor.execute(''' + CREATE TABLE IF NOT EXISTS episodic_memories ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + event TEXT NOT NULL, + outcome TEXT, + emotion TEXT, + salience REAL DEFAULT 0.5, + context TEXT, + timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, + embedding BLOB, + consolidated INTEGER DEFAULT 0 + ) + ''') + + cursor.execute(''' + CREATE TABLE IF NOT EXISTS source_trust ( + source TEXT PRIMARY KEY, + score REAL DEFAULT 0.5, + success_count INTEGER DEFAULT 0, + failure_count INTEGER DEFAULT 0, + last_updated DATETIME DEFAULT CURRENT_TIMESTAMP + ) + ''') + # Create indexes cursor.execute('CREATE INDEX IF NOT EXISTS idx_facts_category ON facts(category)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_facts_access ON facts(access_count)') - + cursor.execute('CREATE INDEX IF NOT EXISTS idx_epi_salience ON episodic_memories(salience)') + cursor.execute('CREATE INDEX IF NOT EXISTS idx_epi_timestamp ON episodic_memories(timestamp)') + self.conn.commit() logger.info(f"Initialized {'PostgreSQL' if self.use_postgres else 'SQLite'} database") @@ -180,7 +248,488 @@ def init_database(self) -> None: if self.conn: self.conn.close() self.conn = None - + + # ------------------------------------------------------------------ + # Episodic memory interface + # ------------------------------------------------------------------ + def add_episode( + self, + event: str, + outcome: Optional[str] = None, + emotion: Optional[str] = None, + salience: float = 0.5, + context: Optional[Dict[str, Any]] = None, + embedding: Optional[bytes] = None, + ) -> int: + """Persist an episodic memory with an associated salience score.""" + if not self.conn: + self.init_database() + + normalized_salience = max(self.minimum_salience, min(salience, 1.0)) + context_payload: Optional[str] + if context is None: + context_payload = None + else: + context_payload = json.dumps(context, default=str) + + try: + cursor = self.conn.cursor() + if self.use_postgres: + cursor.execute( + """ + INSERT INTO episodic_memories (event, outcome, emotion, salience, context, embedding) + VALUES (%s, %s, %s, %s, %s::jsonb, %s) + RETURNING id + """, + (event, outcome, emotion, float(normalized_salience), context_payload, embedding), + ) + episode_id = cursor.fetchone()[0] + else: + cursor.execute( + """ + INSERT INTO episodic_memories (event, outcome, emotion, salience, context, embedding) + VALUES (?, ?, ?, ?, ?, ?) + """, + (event, outcome, emotion, float(normalized_salience), context_payload, embedding), + ) + episode_id = cursor.lastrowid + + self.conn.commit() + return int(episode_id) + except (sqlite3.Error, psycopg2.Error) as exc: # type: ignore[arg-type] + logger.error("Failed to persist episodic memory: %s", exc) + return -1 + + def decay_memories(self, half_life: Optional[float] = None) -> None: + """Apply exponential decay to episodic salience, pruning stale entries.""" + if not self.conn: + self.init_database() + + half_life = half_life or self.memory_decay_half_life + if half_life <= 0: + return + + decay_factor = 0.5 ** (self.retry_delay / half_life) + + try: + cursor = self.conn.cursor() + if self.use_postgres: + cursor.execute( + """ + UPDATE episodic_memories + SET salience = GREATEST(%s, salience * %s) + """, + (self.minimum_salience, decay_factor), + ) + cursor.execute( + "DELETE FROM episodic_memories WHERE salience <= %s", + (self.minimum_salience,), + ) + else: + cursor.execute( + """ + UPDATE episodic_memories + SET salience = MAX(?, salience * ?) + """, + (self.minimum_salience, decay_factor), + ) + cursor.execute( + "DELETE FROM episodic_memories WHERE salience <= ?", + (self.minimum_salience,), + ) + self.conn.commit() + except (sqlite3.Error, psycopg2.Error) as exc: # type: ignore[arg-type] + logger.error("Failed to decay episodic memories: %s", exc) + + def consolidate_memories(self, limit: int = 10) -> List[int]: + """Elevate highly salient episodes into long-term factual knowledge.""" + if not self.conn: + self.init_database() + + promoted: List[int] = [] + try: + cursor = self.conn.cursor() + if self.use_postgres: + cursor.execute( + """ + SELECT id, event, outcome, emotion + FROM episodic_memories + WHERE consolidated = FALSE AND salience >= %s + ORDER BY salience DESC, timestamp DESC + LIMIT %s + """, + (self.consolidation_threshold, limit), + ) + rows = cursor.fetchall() + else: + cursor.execute( + """ + SELECT id, event, outcome, emotion + FROM episodic_memories + WHERE consolidated = 0 AND salience >= ? + ORDER BY salience DESC, timestamp DESC + LIMIT ? + """, + (self.consolidation_threshold, limit), + ) + rows = cursor.fetchall() + + for row in rows: + episode_id, event, outcome, emotion = row + summary_parts = [event] + if outcome: + summary_parts.append(f"Outcome: {outcome}") + if emotion: + summary_parts.append(f"Emotion: {emotion}") + fact_text = " | ".join(summary_parts) + fact_id = self.add_fact( + content=fact_text, + source="episodic_memory", + confidence=0.9, + category="experience", + ) + promoted.append(int(fact_id)) + if self.use_postgres: + cursor.execute( + "UPDATE episodic_memories SET consolidated = TRUE WHERE id = %s", + (episode_id,), + ) + else: + cursor.execute( + "UPDATE episodic_memories SET consolidated = 1 WHERE id = ?", + (episode_id,), + ) + + self.conn.commit() + except (sqlite3.Error, psycopg2.Error) as exc: # type: ignore[arg-type] + logger.error("Failed to consolidate episodic memories: %s", exc) + + return promoted + + def contextual_recall(self, cue: str, limit: int = 5) -> List[Dict[str, Any]]: + """Retrieve episodic memories related to the provided cue.""" + if not self.conn: + self.init_database() + + try: + cursor = self.conn.cursor() + if self.use_postgres: + cursor.execute( + """ + SELECT id, event, outcome, emotion, salience, context::text, timestamp + FROM episodic_memories + WHERE event ILIKE %s OR outcome ILIKE %s + ORDER BY salience DESC, timestamp DESC + LIMIT %s + """, + (f"%{cue}%", f"%{cue}%", limit), + ) + else: + cursor.execute( + """ + SELECT id, event, outcome, emotion, salience, context, timestamp + FROM episodic_memories + WHERE event LIKE ? OR outcome LIKE ? + ORDER BY salience DESC, timestamp DESC + LIMIT ? + """, + (f"%{cue}%", f"%{cue}%", limit), + ) + rows = cursor.fetchall() + memories: List[Dict[str, Any]] = [] + for row in rows: + context_blob = row[5] + parsed_context = None + if context_blob: + try: + parsed_context = json.loads(context_blob) + except json.JSONDecodeError: + parsed_context = {"raw": context_blob} + memories.append( + { + "id": row[0], + "event": row[1], + "outcome": row[2], + "emotion": row[3], + "salience": row[4], + "context": parsed_context, + "timestamp": row[6], + } + ) + return memories + except (sqlite3.Error, psycopg2.Error) as exc: # type: ignore[arg-type] + logger.error("Failed contextual recall: %s", exc) + return [] + + def fusion_search(self, query: str, limit: int = 10) -> Dict[str, List[Dict[str, Any]]]: + """Combine structured facts and episodic memories for richer recall.""" + facts = self.search_facts(query, limit=limit) + episodes = self.contextual_recall(query, limit=max(1, limit // 2)) + return {"facts": facts, "episodes": episodes} + + def synthesize_knowledge(self, topic: str, limit: int = 15) -> Dict[str, Any]: + """Build a lightweight semantic graph for the requested topic.""" + nodes: Dict[str, Dict[str, Any]] = {} + edges: List[Dict[str, Any]] = [] + facts = self.search_facts(topic, limit=limit) + episodes = self.contextual_recall(topic, limit=max(3, limit // 3)) + + for fact in facts: + node_id = f"fact-{fact['id']}" + nodes[node_id] = { + "id": node_id, + "label": fact["content"], + "type": "fact", + "confidence": fact.get("confidence", 1.0), + } + + for episode in episodes: + node_id = f"episode-{episode['id']}" + nodes[node_id] = { + "id": node_id, + "label": episode["event"], + "type": "episode", + "salience": episode.get("salience"), + } + + combined = list(nodes.values()) + for idx, source in enumerate(combined): + for target in combined[idx + 1 : idx + 4]: + edges.append( + { + "source": source["id"], + "target": target["id"], + "weight": 0.5, + "relation": "related", + } + ) + + return {"topic": topic, "nodes": list(nodes.values()), "edges": edges} + + def generate_hypotheses(self, topic: str, limit: int = 5) -> List[str]: + """Draft simple hypotheses based on existing facts and episodes.""" + fused = self.fusion_search(topic, limit=limit * 2) + hypotheses: List[str] = [] + for fact in fused["facts"][:limit]: + hypotheses.append(f"If {fact['content']}, then exploring more about {topic} may reveal deeper causes.") + for episode in fused["episodes"][:limit]: + hypotheses.append( + f"When {episode['event']} occurs, outcome {episode.get('outcome')} could influence future {topic} tasks." + ) + return hypotheses[:limit] + + def verify_claim(self, claim: str, min_sources: int = 2) -> Dict[str, Any]: + """Cross-check a claim across multiple stored sources.""" + facts = self.search_facts(claim, limit=10) + supporting = [fact for fact in facts if claim.lower() in fact["content"].lower()] + trust_scores: List[float] = [] + for fact in supporting: + source = fact.get("source") + if not source: + continue + trust = self.get_source_trust(str(source)) + trust_scores.append(trust["score"]) + # Reinforce trust for sources that consistently support claims + self.update_source_trust(str(source), success=True, weight=0.2) + + average_trust = sum(trust_scores) / len(trust_scores) if trust_scores else 0.0 + + verdict = "unknown" + if len(supporting) >= min_sources and average_trust >= 0.6: + verdict = "supported" + elif supporting and average_trust >= 0.4: + verdict = "partial" + elif supporting: + verdict = "unknown" + + return { + "claim": claim, + "verdict": verdict, + "sources": supporting[:min_sources], + "average_trust": average_trust, + } + + def integrity_check(self) -> Dict[str, Any]: + """Validate database health and surface detected issues.""" + + if not self.conn: + self.init_database() + + engine = "postgres" if self.use_postgres else "sqlite" + result: Dict[str, Any] = {"engine": engine} + + try: + cursor = self.conn.cursor() + if self.use_postgres: + cursor.execute("SELECT 1") + cursor.fetchone() + result["status"] = "ok" + else: + cursor.execute("PRAGMA integrity_check") + rows = cursor.fetchall() + findings = [row[0] if isinstance(row, (list, tuple)) else row for row in rows] + status = findings[0] if findings else "unknown" + result["status"] = "ok" if status == "ok" else "error" + if status != "ok": + result["details"] = findings + return result + except (sqlite3.Error, psycopg2.Error) as exc: # type: ignore[arg-type] + logger.error("Integrity check failed: %s", exc) + result["status"] = "error" + result["error"] = str(exc) + return result + + def recover_integrity(self) -> Dict[str, Any]: + """Attempt to recover from detected integrity issues safely.""" + + engine = "postgres" if self.use_postgres else "sqlite" + outcome: Dict[str, Any] = {"engine": engine} + + try: + if self.conn: + self.conn.close() + except Exception as exc: # pragma: no cover - defensive cleanup + logger.warning("Failed to close connection during recovery: %s", exc) + finally: + self.conn = None + + backup_path: Optional[Path] = None + if not self.use_postgres and self.db_path not in (None, ":memory:"): + candidate = Path(str(self.db_path)) + if candidate.exists(): + backup_path = candidate.with_name(f"{candidate.name}.corrupt.{int(time.time())}") + try: + candidate.rename(backup_path) + except OSError as exc: # pragma: no cover - filesystem edge case + logger.error("Failed to back up corrupt database: %s", exc) + backup_path = None + + self.init_database() + + outcome["status"] = "reset" + if backup_path: + outcome["backup_path"] = str(backup_path) + return outcome + + def update_source_trust(self, source: str, success: bool, weight: float = 1.0) -> None: + """Adjust trust metrics for a given information source.""" + if not source: + return + + if not self.conn: + self.init_database() + + delta = max(weight, 0.0) * (0.05 if success else -0.05) + + try: + cursor = self.conn.cursor() + if self.use_postgres: + cursor.execute( + "SELECT score, success_count, failure_count FROM source_trust WHERE source = %s", + (source,), + ) + else: + cursor.execute( + "SELECT score, success_count, failure_count FROM source_trust WHERE source = ?", + (source,), + ) + row = cursor.fetchone() + + if row: + current_score = row[0] if row[0] is not None else 0.5 + success_count = int(row[1] or 0) + (1 if success else 0) + failure_count = int(row[2] or 0) + (0 if success else 1) + new_score = max(0.0, min(1.0, current_score + delta)) + if self.use_postgres: + cursor.execute( + """ + UPDATE source_trust + SET score = %s, success_count = %s, failure_count = %s, + last_updated = CURRENT_TIMESTAMP + WHERE source = %s + """, + (new_score, success_count, failure_count, source), + ) + else: + cursor.execute( + """ + UPDATE source_trust + SET score = ?, success_count = ?, failure_count = ?, + last_updated = CURRENT_TIMESTAMP + WHERE source = ? + """, + (new_score, success_count, failure_count, source), + ) + else: + base_score = 0.55 if success else 0.45 + success_count = 1 if success else 0 + failure_count = 0 if success else 1 + if self.use_postgres: + cursor.execute( + """ + INSERT INTO source_trust (source, score, success_count, failure_count, last_updated) + VALUES (%s, %s, %s, %s, CURRENT_TIMESTAMP) + """, + (source, base_score, success_count, failure_count), + ) + else: + cursor.execute( + """ + INSERT INTO source_trust (source, score, success_count, failure_count, last_updated) + VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) + """, + (source, base_score, success_count, failure_count), + ) + + self.conn.commit() + except (sqlite3.Error, psycopg2.Error) as exc: # type: ignore[arg-type] + logger.error("Failed to update trust for %s: %s", source, exc) + if self.conn: + self.conn.rollback() + + def get_source_trust(self, source: str) -> Dict[str, Any]: + """Retrieve trust metrics for the provided source.""" + default = { + "source": source, + "score": 0.5, + "success_count": 0, + "failure_count": 0, + "last_updated": None, + } + + if not source: + return default + + if not self.conn: + self.init_database() + + try: + cursor = self.conn.cursor() + if self.use_postgres: + cursor.execute( + "SELECT score, success_count, failure_count, last_updated FROM source_trust WHERE source = %s", + (source,), + ) + else: + cursor.execute( + "SELECT score, success_count, failure_count, last_updated FROM source_trust WHERE source = ?", + (source,), + ) + row = cursor.fetchone() + if not row: + return default + + return { + "source": source, + "score": row[0] if row[0] is not None else 0.5, + "success_count": int(row[1] or 0), + "failure_count": int(row[2] or 0), + "last_updated": row[3], + } + except (sqlite3.Error, psycopg2.Error) as exc: # type: ignore[arg-type] + logger.error("Failed to read trust for %s: %s", source, exc) + return default + def add_fact(self, content: str, source: Optional[str] = None, confidence: float = 1.0, category: Optional[str] = None, embedding: Optional[bytes] = None) -> int: diff --git a/AgentSystem/modules/learning_agent.py b/AgentSystem/modules/learning_agent.py index d68f718..ee31b4f 100644 --- a/AgentSystem/modules/learning_agent.py +++ b/AgentSystem/modules/learning_agent.py @@ -14,18 +14,564 @@ import threading import queue import time -from typing import Dict, List, Any, Optional +import random +import json +from collections import deque +from dataclasses import dataclass +from typing import Dict, List, Any, Optional, Callable, Iterable from pathlib import Path from AgentSystem.utils.logger import get_logger -from AgentSystem.modules.knowledge_manager import KnowledgeManager -from AgentSystem.modules.web_researcher import WebResearcher -from AgentSystem.modules.code_modifier import CodeModifier + +try: + from AgentSystem.modules.knowledge_manager import KnowledgeManager + from AgentSystem.modules.web_researcher import WebResearcher + from AgentSystem.modules.code_modifier import CodeModifier +except ImportError: + import importlib.util + + MODULE_DIR = Path(__file__).resolve().parent + + def _fallback_import(module_name: str): + module_path = MODULE_DIR / f"{module_name}.py" + spec = importlib.util.spec_from_file_location(f"learning_agent_{module_name}", module_path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + try: + spec.loader.exec_module(module) + except Exception: + return None + return module + + km_module = _fallback_import("knowledge_manager") + if km_module and hasattr(km_module, "KnowledgeManager"): + KnowledgeManager = km_module.KnowledgeManager # type: ignore[attr-defined] + else: # pragma: no cover - knowledge manager is required for core operation + raise + + wr_module = _fallback_import("web_researcher") + if wr_module and hasattr(wr_module, "WebResearcher"): + WebResearcher = wr_module.WebResearcher # type: ignore[attr-defined] + else: + class WebResearcher: # type: ignore[no-redef] + """Minimal stub used when web research dependencies are unavailable.""" + + def __init__(self, knowledge_manager: Any, *_, **__): + self.knowledge_manager = knowledge_manager + + def research_topic(self, topic: str, depth: int = 1) -> List[Dict[str, Any]]: + return [] + + cm_module = _fallback_import("code_modifier") + if cm_module and hasattr(cm_module, "CodeModifier"): + CodeModifier = cm_module.CodeModifier # type: ignore[attr-defined] + else: + class CodeModifier: # type: ignore[no-redef] + """Minimal stub used when code modification dependencies are unavailable.""" + + def __init__(self, *_, **__): + pass + + def analyze_code(self, file_path: str) -> Dict[str, Any]: + return {} + + def suggest_improvements(self, file_path: str) -> List[Dict[str, Any]]: + return [] + + def modify_code(self, file_path: str, changes: Dict[str, Any]) -> bool: + return False logger = get_logger("modules.learning_agent") + +@dataclass +class ReflexRule: + """Map fast sensory events to immediate actions.""" + + trigger: str + action: Callable[[Dict[str, Any]], None] + priority: int = 0 + + +class ReflexLayer: + """Fast response layer for immediate reactions.""" + + def __init__(self) -> None: + self._rules: List[ReflexRule] = [] + + def register_rule(self, rule: ReflexRule) -> None: + self._rules.append(rule) + self._rules.sort(key=lambda r: r.priority, reverse=True) + + def process(self, event: Dict[str, Any]) -> bool: + signal = event.get("type") or event.get("signal") + for rule in self._rules: + if rule.trigger == signal: + logger.debug("ReflexLayer matched rule %s for event %s", rule.trigger, event) + rule.action(event) + return True + return False + + +class DeliberativeLayer: + """Planning layer using lightweight tree search heuristics.""" + + def __init__(self, evaluator: Optional[Callable[[Dict[str, Any]], float]] = None) -> None: + self._evaluator = evaluator or (lambda plan: float(plan.get("expected_reward", 0))) + + def plan(self, goal: str, options: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Pick a plan by scoring candidate actions.""" + best_score = float("-inf") + best_plan: List[Dict[str, Any]] = [] + for idx, option in enumerate(options): + option = dict(option) + option.setdefault("steps", [option.get("action", goal)]) + option.setdefault("expected_reward", 0.0) + score = self._evaluator(option) - idx * 0.01 # light tie-breaker + if score > best_score: + best_score = score + best_plan = list(option["steps"]) + logger.debug("DeliberativeLayer chose plan %s for goal %s", best_plan, goal) + return best_plan + + +class MetaCognitiveLayer: + """Monitors performance and updates strategies.""" + + def __init__(self) -> None: + self._history: deque = deque(maxlen=200) + self._active_goals: deque = deque(maxlen=20) + + def record_outcome(self, feedback: Dict[str, Any]) -> None: + self._history.append(feedback) + + def track_goal(self, goal: str) -> None: + if goal not in self._active_goals: + self._active_goals.append(goal) + + def review(self) -> Dict[str, Any]: + if not self._history: + return {"status": "insufficient-data"} + recent_rewards = [item.get("reward", 0.0) for item in list(self._history)[-10:]] + average = sum(recent_rewards) / max(len(recent_rewards), 1) + recommendation = "maintain" if average >= 0 else "adjust-prompts" + return { + "status": "ok" if average >= 0 else "needs-adjustment", + "recent_average_reward": average, + "tracked_goals": list(self._active_goals), + "recommendation": recommendation, + } + + +class CausalInferencer: + """Track simple cause-effect pairs to move beyond correlation.""" + + def __init__(self) -> None: + self._counts: Dict[tuple, Dict[str, int]] = {} + + def observe(self, cause: str, effect: str, success: bool) -> None: + bucket = self._counts.setdefault((cause, effect), {"success": 0, "failure": 0}) + bucket["success" if success else "failure"] += 1 + + def infer(self, cause: str) -> Optional[str]: + best_effect = None + best_ratio = 0.0 + for (observed_cause, effect), stats in self._counts.items(): + if observed_cause != cause: + continue + total = stats["success"] + stats["failure"] + if not total: + continue + ratio = stats["success"] / total + if ratio > best_ratio: + best_ratio = ratio + best_effect = effect + return best_effect + + +class ReActReasoner: + """Blend reasoning traces with acting hooks and memory introspection.""" + + def __init__(self, knowledge_manager: "KnowledgeManager") -> None: + self.knowledge_manager = knowledge_manager + + def reason(self, query: str) -> Dict[str, Any]: + trace: List[str] = [] + trace.append(f"Thought: need information about {query}") + memory = self.knowledge_manager.fusion_search(query, limit=5) + trace.append(f"Retrieved {len(memory['facts'])} facts and {len(memory['episodes'])} episodes") + action = "act:consult_knowledge_base" if memory["facts"] else "act:web_research" + return {"trace": trace, "action": action, "memory": memory} + + +class DistributedAgentMesh: + """Lightweight federated mesh for specialised agents.""" + + def __init__(self) -> None: + self._subscribers: Dict[str, Callable[[Dict[str, Any]], None]] = {} + self._shared_state: Dict[str, Any] = {} + self._weights: Dict[str, float] = {} + + def register( + self, + role: str, + callback: Callable[[Dict[str, Any]], Any], + *, + weight: float = 1.0, + ) -> None: + self._subscribers[role] = callback + try: + self._weights[role] = max(float(weight), 0.0) + except (TypeError, ValueError): + self._weights[role] = 1.0 + + def broadcast(self, message: Dict[str, Any]) -> None: + for role, callback in self._subscribers.items(): + try: + callback(dict(message, target_role=role)) + except Exception as exc: + logger.warning("Mesh callback for %s failed: %s", role, exc) + + def update_shared_state(self, key: str, value: Any) -> None: + self._shared_state[key] = value + + def get_shared_state(self, key: str, default: Any = None) -> Any: + return self._shared_state.get(key, default) + + def request_consensus( + self, + question: Dict[str, Any], + *, + quorum: Optional[float] = None, + ) -> Dict[str, Any]: + """Request proposals from subscribers and return an aggregated decision.""" + + votes: List[Dict[str, Any]] = [] + total_weight = 0.0 + for role, callback in self._subscribers.items(): + payload = dict(question, target_role=role, kind="consensus_request") + try: + response = callback(payload) + except Exception as exc: + logger.warning("Consensus callback for %s failed: %s", role, exc) + continue + + if response is None: + continue + + if not isinstance(response, dict): + response = {"vote": response} + + if "vote" not in response: + continue + + response = dict(response) + response.setdefault("role", role) + weight = response.get("weight", self._weights.get(role, 1.0)) + try: + weight_value = float(weight) + except (TypeError, ValueError): + weight_value = 1.0 + if weight_value < 0: + weight_value = 0.0 + response["weight"] = weight_value + votes.append(response) + total_weight += weight_value + + if not votes: + return { + "decision": None, + "votes": [], + "passed": False, + "total_weight": 0.0, + "decision_weight": 0.0, + } + + tallies: Dict[Any, float] = {} + for entry in votes: + vote_key = entry.get("vote") + tallies[vote_key] = tallies.get(vote_key, 0.0) + entry["weight"] + + decision, decision_weight = max(tallies.items(), key=lambda item: item[1]) + threshold = quorum if quorum is not None else (total_weight / 2.0) + passed = decision_weight >= threshold if total_weight else False + + return { + "decision": decision, + "votes": votes, + "passed": passed, + "total_weight": total_weight, + "decision_weight": decision_weight, + "threshold": threshold, + } + + +class ResilienceManager: + """Monitor agent health and auto-heal where possible.""" + + def __init__(self) -> None: + self._restart_hooks: List[Callable[[], None]] = [] + self._health_checks: Dict[str, Dict[str, Any]] = {} + self._failure_counts: Dict[str, int] = {} + + def register_restart(self, hook: Callable[[], None]) -> None: + self._restart_hooks.append(hook) + + def ensure_thread(self, thread: Optional[threading.Thread], starter: Callable[[], None]) -> None: + if thread and thread.is_alive(): + return + logger.warning("Detected inactive thread; attempting automatic restart") + for hook in self._restart_hooks: + try: + hook() + except Exception as exc: + logger.error("Restart hook failed: %s", exc) + starter() + + def register_health_check( + self, + name: str, + check: Callable[[], Any], + *, + recover: Optional[Callable[[], Any]] = None, + threshold: int = 3, + ) -> None: + self._health_checks[name] = { + "check": check, + "recover": recover, + "threshold": max(1, int(threshold)) if threshold else 3, + } + + def run_health_checks(self) -> Dict[str, Dict[str, Any]]: + outcomes: Dict[str, Dict[str, Any]] = {} + for name, config in self._health_checks.items(): + check_callable = config.get("check") + try: + status = check_callable() if callable(check_callable) else None + if isinstance(status, dict): + result = dict(status) + result.setdefault("status", "ok") + else: + result = { + "status": "ok" if status in (None, True, "ok") else "unknown", + "result": status, + } + except Exception as exc: # pragma: no cover - exercised via tests with recoveries + logger.error("Health check %s failed: %s", name, exc) + result = {"status": "error", "error": str(exc)} + status = None + outcomes[name] = result + + status_value = result.get("status") + if status_value not in {"ok", "pass"}: + recover_callable = config.get("recover") + if recover_callable: + try: + recover_callable() + result["recovered"] = True + except Exception as recover_exc: # pragma: no cover - defensive logging + logger.error("Recovery for %s failed: %s", name, recover_exc) + result["recovered"] = False + return outcomes + + def record_failure(self, component: str, *, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + config = self._health_checks.get(component, {}) + count = self._failure_counts.get(component, 0) + 1 + self._failure_counts[component] = count + threshold = config.get("threshold", 3) + triggered = count >= threshold + if triggered: + self._failure_counts[component] = 0 + recover_callable = config.get("recover") + if recover_callable: + try: + recover_callable() + except Exception as exc: # pragma: no cover - defensive logging + logger.error("Failure recovery for %s failed: %s", component, exc) + triggered = False + summary = { + "component": component, + "count": count, + "threshold": threshold, + "triggered": triggered, + } + if context: + summary["context"] = context + return summary + + def record_success(self, component: str) -> None: + if component in self._failure_counts: + self._failure_counts[component] = 0 + + def get_failure_count(self, component: str) -> int: + return self._failure_counts.get(component, 0) + + +class InferenceRouter: + """Select between local and cloud inference pathways.""" + + def __init__(self) -> None: + self.local_available = False + self.cloud_available = True + self.cached_prompts: Dict[str, Dict[str, Any]] = {} + self._result_cache: Dict[str, Dict[str, Any]] = {} + self.local_handler: Optional[Callable[[str, Dict[str, Any]], Any]] = None + self.cloud_handler: Optional[Callable[[str, Dict[str, Any]], Any]] = None + self.statistics: Dict[str, Dict[str, float]] = { + "local": {"success": 0, "failure": 0}, + "cloud": {"success": 0, "failure": 0}, + "cached": {"hits": 0, "misses": 0}, + } + + def register_local( + self, available: bool, handler: Optional[Callable[[str, Dict[str, Any]], Any]] = None + ) -> None: + if handler is not None: + self.local_handler = handler + self.local_available = bool(available and self.local_handler) + + def register_cloud( + self, available: bool, handler: Optional[Callable[[str, Dict[str, Any]], Any]] = None + ) -> None: + if handler is not None: + self.cloud_handler = handler + self.cloud_available = bool(available and self.cloud_handler) + + def choose(self, task: str, prefer_local: bool = True) -> str: + if prefer_local and self.local_handler and self.local_available: + return "local" + if self.cloud_handler and self.cloud_available: + return "cloud" + if self._result_cache: + return "cached" + return "unavailable" + + def cache_prompt(self, key: str, payload: Dict[str, Any]) -> None: + self.cached_prompts[key] = payload + + def cache_result(self, key: str, payload: Dict[str, Any]) -> None: + self._result_cache[key] = dict(payload, timestamp=time.time()) + + def get_cached_result(self, key: str) -> Optional[Dict[str, Any]]: + return self._result_cache.get(key) + + def _record_stat(self, channel: str, success: bool) -> None: + stats = self.statistics.setdefault(channel, {"success": 0, "failure": 0}) + key = "success" if success else "failure" + stats[key] = stats.get(key, 0) + 1 + + def run( + self, + task: str, + payload: Dict[str, Any], + *, + cache_key: Optional[str] = None, + prefer_local: bool = True, + ) -> Dict[str, Any]: + order = ["local", "cloud"] if prefer_local else ["cloud", "local"] + errors: List[Dict[str, Any]] = [] + + for channel in order: + handler: Optional[Callable[[str, Dict[str, Any]], Any]] + available: bool + if channel == "local": + handler = self.local_handler + available = self.local_available and handler is not None + else: + handler = self.cloud_handler + available = self.cloud_available and handler is not None + + if not available or handler is None: + continue + + try: + result = handler(task, dict(payload)) + except Exception as exc: # pragma: no cover - defensive logging + logger.warning("%s inference path failed: %s", channel, exc) + self._record_stat(channel, success=False) + if channel == "local": + self.local_available = False + else: + self.cloud_available = False + errors.append({"path": channel, "error": str(exc)}) + continue + + self._record_stat(channel, success=True) + response = { + "result": result, + "path": channel, + "cached": False, + "errors": errors, + } + if cache_key: + self.cache_result(cache_key, {"result": result, "path": channel, "task": task}) + return response + + cached_payload: Optional[Dict[str, Any]] = None + if cache_key: + cached_payload = self.get_cached_result(cache_key) + if cached_payload is None and self._result_cache: + cached_payload = next(iter(self._result_cache.values())) + + if cached_payload is not None: + stats = self.statistics.setdefault("cached", {"hits": 0, "misses": 0}) + stats["hits"] = stats.get("hits", 0) + 1 + return { + "result": cached_payload.get("result"), + "path": "cached", + "cached": True, + "errors": errors, + } + + stats = self.statistics.setdefault("cached", {"hits": 0, "misses": 0}) + stats["misses"] = stats.get("misses", 0) + 1 + return {"result": None, "path": "unavailable", "cached": False, "errors": errors} + + +class SocialIntelligenceLayer: + """Minimal affective computing helper.""" + + POSITIVE_WORDS = {"great", "good", "excellent", "awesome", "thanks"} + NEGATIVE_WORDS = {"bad", "terrible", "angry", "upset", "frustrated"} + + def analyse(self, text: str) -> Dict[str, Any]: + words = {w.strip(".,!?" ).lower() for w in text.split()} + positivity = len(words & self.POSITIVE_WORDS) + negativity = len(words & self.NEGATIVE_WORDS) + sentiment = "neutral" + if positivity > negativity: + sentiment = "positive" + elif negativity > positivity: + sentiment = "negative" + return {"sentiment": sentiment, "positivity": positivity, "negativity": negativity} + + def adapt_response(self, text: str, sentiment: str) -> str: + if sentiment == "positive": + return f"I'm glad to hear that! {text}" + if sentiment == "negative": + return f"I understand the concern. {text}" + return text + + +class CognitionStack: + """Aggregate reflex, deliberative, and meta-cognitive layers.""" + + def __init__(self, meta_layer: MetaCognitiveLayer) -> None: + self.reflex = ReflexLayer() + self.deliberative = DeliberativeLayer() + self.meta = meta_layer + + def handle_event(self, event: Dict[str, Any], planner: Callable[[str], List[str]]) -> Dict[str, Any]: + handled = self.reflex.process(event) + response: Dict[str, Any] = {"handled": handled} + if not handled and event.get("goal"): + plan = planner(event["goal"]) + response["plan"] = plan + self.meta.record_outcome({"reward": event.get("reward", 0.0)}) + return response + class LearningAgent: - def __init__(self, + def __init__(self, knowledge_base_path: Optional[str] = None, backup_dir: Optional[str] = None): """ @@ -45,7 +591,43 @@ def __init__(self, self.learning_thread = None self.learning_active = False self.learning_lock = threading.Lock() # Protect learning_active flag - + + # Reward tracking + self._reward_history: deque = deque(maxlen=100) + self._cumulative_reward: float = 0.0 + self._task_outcomes = {"total": 0, "success": 0, "failure": 0} + self._last_feedback: Optional[Dict[str, Any]] = None + + # Cognitive layers + self.meta_layer = MetaCognitiveLayer() + self.cognition = CognitionStack(self.meta_layer) + self.causal_inferencer = CausalInferencer() + self.reasoner = ReActReasoner(self.knowledge_manager) + self.mesh = DistributedAgentMesh() + self.resilience = ResilienceManager() + self.inference_router = InferenceRouter() + self.social_layer = SocialIntelligenceLayer() + self._prompt_versions: Dict[str, Dict[str, Any]] = {} + self._self_play_log: deque = deque(maxlen=50) + self._distillation_buffer: deque = deque(maxlen=200) + + self.mesh.register("Observer", lambda msg: logger.debug("Observer received %s", msg)) + self.resilience.register_restart(self.start_learning) + self.resilience.register_health_check( + "knowledge_base", + self.knowledge_manager.integrity_check, + recover=self.knowledge_manager.recover_integrity, + threshold=1, + ) + self.resilience.register_health_check( + "learning_task", + lambda: { + "status": "active" if self.learning_active else "idle", + "queue_size": self.learning_queue.qsize(), + }, + threshold=3, + ) + def start_learning(self) -> None: """Start background learning thread""" with self.learning_lock: @@ -61,7 +643,8 @@ def start_learning(self) -> None: ) self.learning_thread.start() logger.info("Started background learning thread with thread-safe protection") - + self.meta_layer.track_goal("background_learning") + def stop_learning(self) -> None: """Stop background learning thread""" with self.learning_lock: @@ -74,11 +657,12 @@ def stop_learning(self) -> None: self.learning_thread.join(timeout=5.0) if self.learning_thread.is_alive(): logger.warning("Learning thread did not terminate within timeout") + self.resilience.ensure_thread(self.learning_thread, self.start_learning) else: logger.debug("Learning thread terminated successfully") self.learning_thread = None logger.info("Stopped background learning with thread-safe protection") - + def _learning_loop(self) -> None: """Background learning thread main loop""" while True: @@ -100,36 +684,77 @@ def _learning_loop(self) -> None: # Process task with enhanced error recovery task_type = task.get("type") task_success = False - + task_details: Dict[str, Any] = {} + try: if task_type == "research": topic = task["topic"] depth = task.get("depth", 1) logger.info(f"Starting research task: {topic} (depth={depth})") results = self.research_topic(topic, depth) + sources = sorted( + { + entry.get("source") + for entry in results + if isinstance(entry, dict) and entry.get("source") + } + ) logger.info(f"Completed research: {topic} - found {len(results)} results") + task_details = { + "result_count": len(results), + "topic": topic, + "sources": sources, + } task_success = True - + elif task_type == "improve_code": file_path = task["file"] logger.info(f"Starting code improvement: {file_path}") improvements = self.improve_code(file_path) logger.info(f"Completed improvement: {file_path} - made {len(improvements)} improvements") + task_details = { + "change_count": len(improvements), + "file_path": file_path + } task_success = True - + else: logger.warning(f"Unknown task type: {task_type}") - + except Exception as task_error: logger.error(f"Task processing failed for {task_type}: {task_error}") logger.exception("Task processing error details:") # Continue processing other tasks despite this failure - + self.learning_queue.task_done() processing_time = time.time() - start_time status = "SUCCESS" if task_success else "FAILED" logger.info(f"Task {status}: {task_type} in {processing_time:.2f}s | Queue: {self.learning_queue.qsize()} pending") - + + reward = self._calculate_reward(task_type, task_success, processing_time, task_details) + self._record_reward(task_type, reward, task_success, processing_time, task_details) + if task_type: + self.causal_inferencer.observe(task_type, status, task_success) + self.meta_layer.record_outcome({"reward": reward, "task": task_type, "success": task_success}) + if task_type == "research" and task_success: + self.mesh.broadcast({"event": "research_complete", "details": task_details}) + if task_success: + self.resilience.record_success("learning_task") + else: + failure_state = self.resilience.record_failure( + "learning_task", + context={ + "task": task_type, + "details": task_details, + "processing_time": processing_time, + }, + ) + if failure_state.get("triggered"): + logger.warning( + "Repeated learning task failures triggered recovery: %s", + failure_state, + ) + except Exception as e: logger.error(f"Critical error in learning loop: {e}") logger.exception("Learning loop critical error details:") @@ -149,11 +774,12 @@ def queue_research(self, topic: str, depth: int = 1) -> None: "topic": topic, "depth": depth }) - + self.meta_layer.track_goal(f"research:{topic}") + def queue_code_improvement(self, file_path: str) -> None: """ Queue a code improvement task - + Args: file_path: Path to file to improve """ @@ -161,6 +787,7 @@ def queue_code_improvement(self, file_path: str) -> None: "type": "improve_code", "file": file_path }) + self.meta_layer.track_goal(f"improve:{file_path}") def research_topic(self, topic: str, depth: int = 1) -> List[Dict[str, Any]]: """ @@ -248,21 +875,534 @@ def improve_code(self, file_path: str) -> List[Dict[str, Any]]: "type": suggestion["type"], "description": suggestion["description"] } - + # Try to apply changes if self.code_modifier.modify_code(file_path, changes): improvements.append(changes) - + return improvements - + + # ------------------------------------------------------------------ + # Advanced cognition helpers + # ------------------------------------------------------------------ + def process_event(self, event: Dict[str, Any]) -> Dict[str, Any]: + """Route sensory or system events through the cognition stack.""" + + def planner(goal: str) -> List[str]: + options = ( + { + "action": "research", + "steps": [f"Research {goal}", "Summarise findings"], + "expected_reward": 0.6, + }, + { + "action": "reflect", + "steps": [f"Consult memories about {goal}", "Draft reflection"], + "expected_reward": 0.5, + }, + ) + return self.cognition.deliberative.plan(goal, options) + + response = self.cognition.handle_event(event, planner) + if not response.get("handled") and event.get("goal"): + self.learning_queue.put({"type": "research", "topic": event["goal"], "depth": 1}) + return response + + def deliberate(self, goal: str) -> List[str]: + """Perform a deliberate planning pass for the provided goal.""" + options = [ + {"action": "research", "steps": [f"Investigate {goal}"], "expected_reward": 0.5}, + {"action": "consult", "steps": [f"Recall experiences about {goal}", "Synthesize learnings"], "expected_reward": 0.55}, + ] + plan = self.cognition.deliberative.plan(goal, options) + self.meta_layer.track_goal(goal) + return plan + + def meta_review(self) -> Dict[str, Any]: + """Expose meta-cognitive review data.""" + return self.meta_layer.review() + + def react_reason(self, query: str) -> Dict[str, Any]: + """Run a ReAct-style reasoning cycle.""" + return self.reasoner.reason(query) + + # ------------------------------------------------------------------ + # Dynamic learning loop extensions + # ------------------------------------------------------------------ + def simulate_self_play(self, scenario: str) -> Dict[str, Any]: + """Run a lightweight self-play scenario to gather experience.""" + agent_score = random.uniform(-0.5, 1.0) + outcome = "win" if agent_score > 0 else "loss" + record = { + "scenario": scenario, + "score": agent_score, + "outcome": outcome, + } + self._self_play_log.append(record) + self.knowledge_manager.add_episode( + event=f"Self-play {scenario}", + outcome=outcome, + emotion="positive" if agent_score > 0 else "frustrated", + salience=min(1.0, max(0.1, 0.5 + agent_score / 2)), + context={"score": agent_score}, + ) + return record + + def register_prompt( + self, + key: str, + template: str, + metadata: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """Register a prompt template for evolution and tracking.""" + + state = self._prompt_versions.setdefault( + key, + { + "version": 1, + "template": template, + "history": [], + "stats": { + "success": 0, + "failure": 0, + "success_streak": 0, + "failure_streak": 0, + }, + "metadata": metadata.copy() if metadata else {}, + "adjustments": [], + }, + ) + if not state.get("template"): + state["template"] = template + if metadata: + state.setdefault("metadata", {}).update(metadata) + return state + + def get_prompt_template(self, key: str) -> Optional[str]: + """Return the currently active template for a prompt identifier.""" + + state = self._prompt_versions.get(key) + return state.get("template") if state else None + + def evolve_prompts( + self, + key: str, + success: bool, + notes: Optional[str] = None, + reward: Optional[float] = None, + template: Optional[str] = None, + ) -> Dict[str, Any]: + """Adapt prompts based on performance history and feedback.""" + + state = self._prompt_versions.setdefault( + key, + { + "version": 1, + "template": template or "", + "history": [], + "stats": { + "success": 0, + "failure": 0, + "success_streak": 0, + "failure_streak": 0, + }, + "metadata": {}, + "adjustments": [], + }, + ) + if template and not state.get("template"): + state["template"] = template + + stats = state.setdefault( + "stats", + { + "success": 0, + "failure": 0, + "success_streak": 0, + "failure_streak": 0, + }, + ) + + entry = { + "success": success, + "notes": notes, + "reward": reward, + "version": state["version"], + "timestamp": time.time(), + } + state.setdefault("history", []).append(entry) + + if success: + stats["success"] = stats.get("success", 0) + 1 + stats["success_streak"] = stats.get("success_streak", 0) + 1 + stats["failure_streak"] = 0 + else: + stats["failure"] = stats.get("failure", 0) + 1 + stats["failure_streak"] = stats.get("failure_streak", 0) + 1 + stats["success_streak"] = 0 + + mutated = False + if not success and stats["failure_streak"] >= 2: + new_template = self._generate_prompt_variant( + state.get("template", ""), + notes, + state["version"] + 1, + state.setdefault("adjustments", []), + ) + if new_template != state.get("template"): + state["template"] = new_template + state["version"] += 1 + stats["failure_streak"] = 0 + state.setdefault("adjustments", []).append( + {"note": notes or "", "version": state["version"], "timestamp": time.time()} + ) + mutated = True + elif success and stats.get("success_streak", 0) >= 3: + state.setdefault("metadata", {})["stabilised"] = True + + cache_payload = { + "template": state.get("template"), + "version": state["version"], + "mutated": mutated, + "success": stats.get("success", 0), + "failure": stats.get("failure", 0), + } + self.inference_router.cache_prompt(key, cache_payload) + return state + + def auto_evolve_prompts(self) -> List[str]: + """Automatically evolve prompts when the meta review indicates issues.""" + + review = self.meta_layer.review() + evolved: List[str] = [] + if review.get("status") == "needs-adjustment": + for key, state in self._prompt_versions.items(): + stats = state.get("stats", {}) + if stats.get("failure", 0) > stats.get("success", 0): + self.evolve_prompts(key, success=False, notes="Meta-review requested refinement.") + evolved.append(key) + return evolved + + def log_interaction(self, prompt_id: str, prompt: str, response: str, reward: float) -> None: + """Record an interaction for later offline distillation.""" + + record = { + "prompt_id": prompt_id, + "prompt": prompt, + "response": response, + "reward": reward, + "timestamp": time.time(), + } + self._distillation_buffer.append(record) + + def distill_model(self, output_path: Optional[str] = None) -> Dict[str, Any]: + """Produce a distilled dataset from recent interactions for offline tuning.""" + + samples = list(self._distillation_buffer) + if not samples: + return {"status": "no-data", "samples": 0} + + average_reward = sum(sample["reward"] for sample in samples) / len(samples) + if output_path: + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + for sample in samples: + handle.write(json.dumps(sample) + "\n") + + summary = { + "status": "distilled", + "samples": len(samples), + "average_reward": average_reward, + "prompt_ids": sorted({sample["prompt_id"] for sample in samples}), + } + return summary + + def _generate_prompt_variant( + self, + template: str, + notes: Optional[str], + version: int, + adjustments: List[Dict[str, Any]], + ) -> str: + """Create a deterministic prompt variation based on prior adjustments.""" + + base = template.strip() if template else "You are an adaptive research assistant." + guidance_segments: List[str] = [] + if notes: + guidance_segments.append(f"Feedback: {notes.strip()}") + if not adjustments: + guidance_segments.append("Incorporate verification and reflection before final answers.") + else: + guidance_segments.append("Provide deeper analysis and cite relevant memories when possible.") + guidance = " ".join(segment for segment in guidance_segments if segment).strip() + if guidance: + base = f"{base}\n\n[Adjustment v{version}]: {guidance}" + return base + + def reinforcement_feedback(self, module: str, reward: float) -> None: + """Record reinforcement signal for the specified module.""" + self._record_reward(module, reward, reward >= 0, processing_time=0.0, details={"module": module}) + + def register_observation(self, description: str, success: bool, emotion: Optional[str] = None) -> None: + """Log an observation into episodic memory and cognition layers.""" + salience = 0.9 if success else 0.4 + self.knowledge_manager.add_episode( + event=description, + outcome="success" if success else "failure", + emotion=emotion, + salience=salience, + ) + self.meta_layer.record_outcome({"reward": 1.0 if success else -0.5, "event": description}) + + def contextual_recall(self, cue: str) -> Dict[str, List[Dict[str, Any]]]: + """Fetch related facts and episodes for the provided cue.""" + return self.knowledge_manager.fusion_search(cue) + + def choose_inference_path(self, task: str, prefer_local: bool = True) -> str: + """Select the inference strategy (local/cloud/cached).""" + return self.inference_router.choose(task, prefer_local=prefer_local) + + def register_local_inference( + self, handler: Optional[Callable[[str, Dict[str, Any]], Any]] + ) -> None: + """Register a local inference handler and mark availability.""" + + self.inference_router.register_local(bool(handler), handler=handler) + + def register_cloud_inference( + self, handler: Optional[Callable[[str, Dict[str, Any]], Any]] + ) -> None: + """Register a cloud inference handler and mark availability.""" + + self.inference_router.register_cloud(bool(handler), handler=handler) + + def generate_inference( + self, + task: str, + prompt: str, + *, + context: Optional[Dict[str, Any]] = None, + prefer_local: bool = True, + cache_key: Optional[str] = None, + ) -> Dict[str, Any]: + """Execute an inference request using the configured hybrid strategy.""" + + payload = {"prompt": prompt, "context": dict(context or {})} + key = cache_key or f"{task}:{hash(prompt) & 0xFFFFFFFF:x}" + outcome = self.inference_router.run( + task, payload, cache_key=key, prefer_local=prefer_local + ) + + result = outcome.get("result") + if result is None: + return outcome + + if not isinstance(result, dict): + result = {"response": str(result)} + outcome["result"] = result + + reward_value = result.get("reward", 0.0) + try: + reward = float(reward_value) + except (TypeError, ValueError): + reward = 0.0 + + if outcome.get("path") in {"local", "cloud"}: + self.log_interaction(task, prompt, result.get("response", ""), reward) + + return outcome + + def adapt_response_tone(self, text: str) -> Dict[str, Any]: + """Analyse sentiment and adapt the response tone.""" + analysis = self.social_layer.analyse(text) + adapted = self.social_layer.adapt_response(text, analysis["sentiment"]) + return {"analysis": analysis, "response": adapted} + + def join_mesh( + self, + role: str, + callback: Callable[[Dict[str, Any]], Any], + *, + weight: float = 1.0, + ) -> None: + """Register a specialised agent callback within the distributed mesh.""" + self.mesh.register(role, callback, weight=weight) + + def share_mesh_state(self, key: str, value: Any) -> None: + """Publish shared state for other agents in the mesh.""" + self.mesh.update_shared_state(key, value) + + def get_mesh_state(self, key: str, default: Any = None) -> Any: + """Retrieve shared mesh state.""" + return self.mesh.get_shared_state(key, default) + + def consensus_plan( + self, + goal: str, + options: List[Dict[str, Any]], + *, + quorum: Optional[float] = None, + ) -> Dict[str, Any]: + """Seek a consensus-backed plan with a deliberative fallback.""" + + consensus = self.mesh.request_consensus( + {"goal": goal, "options": options}, quorum=quorum + ) + + plan_steps = self.cognition.deliberative.plan(goal, options) + + decision_key = consensus.get("decision") + if decision_key is not None: + chosen_option = next( + ( + opt + for opt in options + if opt.get("id") == decision_key + or opt.get("action") == decision_key + ), + None, + ) + if chosen_option and consensus.get("passed"): + plan_steps = list( + chosen_option.get( + "steps", [chosen_option.get("action", goal)] + ) + ) + + return {"plan": plan_steps, "consensus": consensus} + + def _calculate_reward( + self, + task_type: Optional[str], + success: bool, + processing_time: float, + details: Dict[str, Any], + ) -> float: + """Compute a heuristic reward score for a completed task.""" + base_reward = 1.0 if success else -1.0 + + if success: + if task_type == "research": + result_count = details.get("result_count", 0) + base_reward += min(result_count, 5) * 0.1 + elif task_type == "improve_code": + change_count = details.get("change_count", 0) + base_reward += min(change_count, 5) * 0.2 + + # Penalize long running tasks slightly to encourage efficiency + base_reward -= min(processing_time / 60.0, 0.5) + return base_reward + + def _record_reward( + self, + task_type: Optional[str], + reward: float, + success: bool, + processing_time: float, + details: Dict[str, Any], + ) -> None: + """Persist reward information for later introspection.""" + entry = { + "task_type": task_type, + "reward": reward, + "success": success, + "processing_time": processing_time, + "details": details, + "timestamp": time.time(), + } + self._reward_history.append(entry) + self._cumulative_reward += reward + self._task_outcomes["total"] += 1 + if success: + self._task_outcomes["success"] += 1 + else: + self._task_outcomes["failure"] += 1 + self._last_feedback = entry + logger.info( + "Recorded reward %.2f for task %s (success=%s, duration=%.2fs)", + reward, + task_type, + success, + processing_time, + ) + sources_field = None + if isinstance(details, dict): + sources_field = details.get("sources") + if sources_field: + if isinstance(sources_field, (list, tuple, set)): + candidates = [str(src) for src in sources_field if src] + else: + candidates = [str(sources_field)] + for source in candidates: + try: + self.knowledge_manager.update_source_trust(source, success) + except Exception as exc: # pragma: no cover - defensive safeguard + logger.debug("Skipping trust update for %s: %s", source, exc) + self.meta_layer.record_outcome(entry) + + def get_learning_feedback(self) -> Dict[str, Any]: + """Return aggregate reward metrics for the learning loop.""" + recent = list(self._reward_history) + recent_average = ( + sum(item["reward"] for item in recent) / len(recent) + if recent + else 0.0 + ) + success_rate = ( + self._task_outcomes["success"] / self._task_outcomes["total"] + if self._task_outcomes["total"] + else 0.0 + ) + return { + "cumulative_reward": self._cumulative_reward, + "recent_average_reward": recent_average, + "total_tasks": self._task_outcomes["total"], + "success_rate": success_rate, + "last_feedback": self._last_feedback, + "meta_review": self.meta_layer.review(), + } + + def submit_feedback( + self, + score: float, + note: Optional[str] = None, + task_type: str = "external_feedback", + ) -> None: + """Allow external systems to provide manual reward signals.""" + feedback_entry = { + "task_type": task_type, + "reward": score, + "success": score >= 0, + "processing_time": 0.0, + "details": {"note": note} if note else {}, + "timestamp": time.time(), + } + self._reward_history.append(feedback_entry) + self._cumulative_reward += score + self._task_outcomes["total"] += 1 + if score >= 0: + self._task_outcomes["success"] += 1 + else: + self._task_outcomes["failure"] += 1 + self._last_feedback = feedback_entry + logger.info("Manual feedback recorded with reward %.2f (%s)", score, note or "no note") + def get_knowledge_stats(self) -> Dict[str, Any]: """Get statistics about acquired knowledge""" - return { + stats = { "facts": len(self.knowledge_manager.search_facts("")), "categories": len(set(f["category"] for f in self.knowledge_manager.search_facts(""))), "queue_size": self.learning_queue.qsize(), "is_learning": bool(self.learning_thread and self.learning_thread.is_alive()) } + stats["performance"] = self.get_learning_feedback() + return stats + + def perform_health_checks(self) -> Dict[str, Any]: + """Run registered resilience health checks.""" + + return self.resilience.run_health_checks() def shutdown(self) -> None: """Cleanup and shutdown agent""" diff --git a/AgentSystem/modules/sensory_input.py b/AgentSystem/modules/sensory_input.py index 4d35fac..763cbb9 100644 --- a/AgentSystem/modules/sensory_input.py +++ b/AgentSystem/modules/sensory_input.py @@ -10,7 +10,8 @@ import queue import json import base64 -from typing import Dict, List, Any, Optional, Callable, Union +from collections import deque +from typing import Dict, List, Any, Optional, Callable, Union, Iterable from datetime import datetime # Local imports @@ -20,21 +21,312 @@ logger = get_logger("modules.sensory_input") try: - import cv2 - import numpy as np - import pyaudio - import speech_recognition as sr - from PIL import Image + import cv2 # type: ignore + import numpy as np # type: ignore + import pyaudio # type: ignore + import speech_recognition as sr # type: ignore + from PIL import Image # type: ignore SENSORY_IMPORTS_AVAILABLE = True except ImportError: - logger.warning("Sensory input dependencies not available. Install with: pip install opencv-python numpy pyaudio SpeechRecognition pillow") + cv2 = None # type: ignore + np = None # type: ignore + pyaudio = None # type: ignore + sr = None # type: ignore + Image = None # type: ignore + logger.warning( + "Sensory input dependencies not available. Install with: pip install opencv-python numpy pyaudio SpeechRecognition pillow" + ) SENSORY_IMPORTS_AVAILABLE = False +class AudioCaptureBackend: + """Interface for audio capture implementations.""" + + @property + def is_available(self) -> bool: + return False + + def start(self, sample_rate: int, chunk_size: int, device_index: Optional[int] = None) -> bool: + """Start the capture stream.""" + return False + + def read_chunk(self, chunk_size: int) -> bytes: + """Read a chunk of audio data.""" + return b"" + + def stop(self) -> None: + """Stop the capture stream.""" + + def list_devices(self) -> List[Dict[str, Any]]: + """Return available capture devices.""" + return [] + + def shutdown(self) -> None: + """Release backend resources.""" + self.stop() + + +class VideoCaptureBackend: + """Interface for video capture implementations.""" + + @property + def is_available(self) -> bool: + return False + + def start(self, camera_index: int, width: int, height: int, fps: int) -> bool: + """Open the capture stream.""" + return False + + def read_frame(self) -> Optional[Any]: + """Read a frame from the stream.""" + return None + + def stop(self) -> None: + """Stop the capture stream.""" + + def list_cameras(self) -> List[Dict[str, Any]]: + """Return available camera descriptions.""" + return [] + + +class SyntheticAudioBackend(AudioCaptureBackend): + """Feed audio data from a predefined iterable for simulations.""" + + def __init__(self, chunks: Optional[Iterable[bytes]] = None) -> None: + self._chunks = deque(chunks or []) + self._active = False + + @property + def is_available(self) -> bool: + return True + + def start(self, sample_rate: int, chunk_size: int, device_index: Optional[int] = None) -> bool: + self._active = True + return True + + def read_chunk(self, chunk_size: int) -> bytes: + if not self._active or not self._chunks: + return b"" + return self._chunks.popleft() + + def stop(self) -> None: + self._active = False + + +class SyntheticVideoBackend(VideoCaptureBackend): + """Provide synthetic frames for testing or simulation feeds.""" + + def __init__(self, frames: Optional[Iterable[Any]] = None) -> None: + self._frames = deque(frames or []) + self._active = False + + @property + def is_available(self) -> bool: + return True + + def start(self, camera_index: int, width: int, height: int, fps: int) -> bool: + self._active = True + return True + + def read_frame(self) -> Optional[Any]: # type: ignore[override] + if not self._active or not self._frames: + return None + return self._frames.popleft() + + def stop(self) -> None: + self._active = False + + +class MultimodalFusionEngine: + """Fuse vision, audio, and optional text into a shared embedding.""" + + def __init__(self) -> None: + self._history: deque = deque(maxlen=100) + + def fuse( + self, + audio_event: Optional[Dict[str, Any]] = None, + video_event: Optional[Dict[str, Any]] = None, + text: Optional[str] = None, + ) -> Dict[str, Any]: + features: Dict[str, Any] = {"timestamp": time.time()} + if audio_event: + raw = audio_event.get("raw_data") + features["audio_energy"] = len(raw) if isinstance(raw, (bytes, bytearray)) else 0 + features["audio_label"] = audio_event.get("type") + if video_event: + features["visual_objects"] = video_event.get("objects", []) + features["frame_shape"] = video_event.get("frame_shape") + if text: + features["text"] = text + features["embedding"] = self._build_signature(features) + self._history.append(features) + return features + + def _build_signature(self, features: Dict[str, Any]) -> List[float]: + signature = [0.0, 0.0, 0.0] + if "audio_energy" in features: + signature[0] = min(1.0, features["audio_energy"] / 10000.0) + if features.get("visual_objects"): + signature[1] = min(1.0, len(features["visual_objects"])) + if features.get("text"): + signature[2] = min(1.0, len(str(features["text"])) / 200.0) + return signature + + def recent_history(self) -> List[Dict[str, Any]]: + return list(self._history) + + +class CrossModalReasoner: + """Perform lightweight reasoning across fused sensory channels.""" + + def __init__(self, fusion_engine: MultimodalFusionEngine) -> None: + self.fusion_engine = fusion_engine + + def infer_context(self, fused_event: Dict[str, Any]) -> Dict[str, Any]: + audio_energy = fused_event.get("audio_energy", 0) + objects = fused_event.get("visual_objects", []) + context = "unknown" + if audio_energy and objects: + if "water" in objects or "pool" in objects: + context = "water_scene" if audio_energy > 0 else "still_water" + elif "person" in objects: + context = "conversation" if audio_energy > 0 else "observation" + elif objects: + context = "visual_only" + elif audio_energy: + context = "audio_only" + return {"context": context, "confidence": 0.6 if context != "unknown" else 0.2} + + +class PyAudioCaptureBackend(AudioCaptureBackend): + """Hardware audio capture implementation using PyAudio.""" + + def __init__(self) -> None: + self._audio = pyaudio.PyAudio() if SENSORY_IMPORTS_AVAILABLE and pyaudio else None + self._stream = None + + @property + def is_available(self) -> bool: + return bool(self._audio) + + def start(self, sample_rate: int, chunk_size: int, device_index: Optional[int] = None) -> bool: + if not self._audio: + return False + self._stream = self._audio.open( + format=pyaudio.paInt16, # type: ignore[attr-defined] + channels=1, + rate=sample_rate, + input=True, + frames_per_buffer=chunk_size, + input_device_index=device_index, + ) + return True + + def read_chunk(self, chunk_size: int) -> bytes: + if not self._stream: + return b"" + return self._stream.read(chunk_size, exception_on_overflow=False) + + def stop(self) -> None: + if self._stream: + try: + self._stream.stop_stream() + self._stream.close() + finally: + self._stream = None + + def list_devices(self) -> List[Dict[str, Any]]: + devices: List[Dict[str, Any]] = [] + if not self._audio: + return devices + + for i in range(self._audio.get_device_count()): + device_info = self._audio.get_device_info_by_index(i) + if device_info.get('maxInputChannels', 0) > 0: + devices.append({ + 'index': i, + 'name': device_info.get('name'), + 'channels': device_info.get('maxInputChannels'), + 'sample_rate': int(device_info.get('defaultSampleRate', 0)), + }) + return devices + + def shutdown(self) -> None: + self.stop() + if self._audio: + self._audio.terminate() + self._audio = None + + +class OpenCVCaptureBackend(VideoCaptureBackend): + """Hardware video capture implementation using OpenCV.""" + + def __init__(self) -> None: + self._cap = None + + @property + def is_available(self) -> bool: + return cv2 is not None + + def start(self, camera_index: int, width: int, height: int, fps: int) -> bool: + if cv2 is None: + return False + + self._cap = cv2.VideoCapture(camera_index) + if not self._cap or not self._cap.isOpened(): + self._cap = None + return False + + self._cap.set(cv2.CAP_PROP_FRAME_WIDTH, width) + self._cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height) + if fps > 0: + self._cap.set(cv2.CAP_PROP_FPS, fps) + return True + + def read_frame(self) -> Optional[Any]: + if not self._cap: + return None + ret, frame = self._cap.read() + if not ret: + return None + return frame + + def stop(self) -> None: + if self._cap: + self._cap.release() + self._cap = None + + def list_cameras(self) -> List[Dict[str, Any]]: + cameras: List[Dict[str, Any]] = [] + if cv2 is None: + return cameras + + for i in range(10): + cap = cv2.VideoCapture(i) + try: + if cap.isOpened(): + cameras.append({ + 'index': i, + 'name': f"Camera {i}", + 'width': int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)), + 'height': int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)), + 'fps': int(cap.get(cv2.CAP_PROP_FPS)), + }) + finally: + cap.release() + return cameras + + class AudioProcessor: """Processes audio input from microphone""" - - def __init__(self, sample_rate: int = 16000, chunk_size: int = 1024): + + def __init__( + self, + sample_rate: int = 16000, + chunk_size: int = 1024, + capture_backend: Optional[AudioCaptureBackend] = None, + ): """ Initialize the audio processor @@ -47,13 +339,29 @@ def __init__(self, sample_rate: int = 16000, chunk_size: int = 1024): self.audio_queue = queue.Queue() self.is_recording = False self.recording_thread = None - + # Speech recognition - self.recognizer = sr.Recognizer() if SENSORY_IMPORTS_AVAILABLE else None - - # Audio device info - self.audio = pyaudio.PyAudio() if SENSORY_IMPORTS_AVAILABLE else None - self.devices = self._get_audio_devices() if SENSORY_IMPORTS_AVAILABLE else [] + self.recognizer = sr.Recognizer() if SENSORY_IMPORTS_AVAILABLE and sr else None + + self.capture_backend: Optional[AudioCaptureBackend] = capture_backend + if self.capture_backend is None and SENSORY_IMPORTS_AVAILABLE and pyaudio: + self.capture_backend = PyAudioCaptureBackend() + + @property + def is_available(self) -> bool: + return bool(self.capture_backend and self.capture_backend.is_available) + + @property + def devices(self) -> List[Dict[str, Any]]: + if not self.capture_backend: + return [] + return self.capture_backend.list_devices() + + def use_backend(self, backend: Optional[AudioCaptureBackend]) -> None: + """Swap the capture backend implementation.""" + if self.capture_backend and self.capture_backend is not backend: + self.capture_backend.shutdown() + self.capture_backend = backend def _get_audio_devices(self) -> List[Dict[str, Any]]: """Get available audio input devices""" @@ -82,14 +390,14 @@ def start_recording(self, device_index: Optional[int] = None) -> bool: Returns: Success flag """ - if not SENSORY_IMPORTS_AVAILABLE: - logger.error("Audio processing dependencies not available") + if not self.is_available: + logger.error("No audio capture backend configured") return False - + if self.is_recording: logger.warning("Already recording") return False - + try: self.is_recording = True self.recording_thread = threading.Thread( @@ -120,7 +428,10 @@ def stop_recording(self) -> bool: if self.recording_thread: self.recording_thread.join(timeout=2.0) self.recording_thread = None - + + if self.capture_backend: + self.capture_backend.stop() + logger.info("Stopped audio recording") return True @@ -131,67 +442,69 @@ def _record_audio_thread(self, device_index: Optional[int] = None) -> None: Args: device_index: Index of audio device to use """ - stream = self.audio.open( - format=pyaudio.paInt16, - channels=1, - rate=self.sample_rate, - input=True, - frames_per_buffer=self.chunk_size, - input_device_index=device_index - ) - + if not self.capture_backend: + logger.error("No audio capture backend available") + self.is_recording = False + return + + if not self.capture_backend.start(self.sample_rate, self.chunk_size, device_index=device_index): + logger.error("Failed to start audio backend stream") + self.is_recording = False + return + # Buffer to accumulate audio chunks - audio_buffer = [] - buffer_duration_sec = 0 + audio_buffer: List[bytes] = [] + buffer_duration_sec = 0.0 target_duration_sec = 3 # Process in 3-second chunks - - while self.is_recording: - try: - # Read audio chunk - data = stream.read(self.chunk_size, exception_on_overflow=False) - audio_buffer.append(data) - - # Calculate buffer duration - buffer_duration_sec += self.chunk_size / self.sample_rate - - # If buffer reaches target duration, process it - if buffer_duration_sec >= target_duration_sec: - # Process buffer (in a separate thread to avoid blocking) - threading.Thread( - target=self._process_audio_chunk, - args=(b''.join(audio_buffer),), - daemon=True - ).start() - - # Clear buffer - audio_buffer = [] - buffer_duration_sec = 0 - - except Exception as e: - logger.error(f"Error recording audio: {e}") - time.sleep(0.1) # Prevent tight loop on errors - - # Clean up - stream.stop_stream() - stream.close() + + try: + while self.is_recording: + try: + data = self.capture_backend.read_chunk(self.chunk_size) + if not data: + time.sleep(0.01) + continue + + audio_buffer.append(data) + buffer_duration_sec += self.chunk_size / self.sample_rate + + if buffer_duration_sec >= target_duration_sec: + threading.Thread( + target=self._process_audio_chunk, + args=(b''.join(audio_buffer),), + daemon=True, + ).start() + + audio_buffer = [] + buffer_duration_sec = 0.0 + + except Exception as e: + logger.error(f"Error recording audio: {e}") + time.sleep(0.1) + finally: + self.capture_backend.stop() def _process_audio_chunk(self, audio_data: bytes) -> None: """ Process an audio chunk for speech recognition - + Args: audio_data: Raw audio data """ try: + if not self.recognizer or sr is None: + self._queue_raw_audio_event(audio_data) + return + # Convert audio data to AudioData for speech recognition audio = sr.AudioData(audio_data, self.sample_rate, 2) # 2 bytes per sample (16-bit) - + # Try to recognize speech try: text = self.recognizer.recognize_google(audio) if text: logger.debug(f"Recognized speech: {text}") - + # Add to queue self.audio_queue.put({ 'type': 'speech', @@ -204,21 +517,27 @@ def _process_audio_chunk(self, audio_data: bytes) -> None: self._extract_audio_features(audio_data) except Exception as e: logger.error(f"Speech recognition error: {e}") - + self._queue_raw_audio_event(audio_data) + except Exception as e: logger.error(f"Error processing audio chunk: {e}") - + self._queue_raw_audio_event(audio_data) + def _extract_audio_features(self, audio_data: bytes) -> None: """ Extract features from audio data when speech isn't detected - + Args: audio_data: Raw audio data """ try: + if np is None: + self._queue_raw_audio_event(audio_data) + return + # Convert to numpy array for processing audio_np = np.frombuffer(audio_data, dtype=np.int16) - + # Calculate basic audio features if len(audio_np) > 0: rms = np.sqrt(np.mean(np.square(audio_np.astype(np.float32)))) @@ -238,6 +557,17 @@ def _extract_audio_features(self, audio_data: bytes) -> None: }) except Exception as e: logger.error(f"Error extracting audio features: {e}") + self._queue_raw_audio_event(audio_data) + + def _queue_raw_audio_event(self, audio_data: bytes) -> None: + """Add a raw audio event to the queue for downstream processing.""" + snippet = base64.b64encode(audio_data[: min(len(audio_data), self.chunk_size * 2)]).decode('utf-8') if audio_data else "" + self.audio_queue.put({ + 'type': 'audio_chunk', + 'timestamp': datetime.now().isoformat(), + 'sample_rate': self.sample_rate, + 'preview': snippet, + }) def get_next_audio_event(self, timeout: Optional[float] = 0.1) -> Optional[Dict[str, Any]]: """ @@ -254,11 +584,22 @@ def get_next_audio_event(self, timeout: Optional[float] = 0.1) -> Optional[Dict[ except queue.Empty: return None + def shutdown(self) -> None: + """Release backend resources.""" + if self.capture_backend: + self.capture_backend.shutdown() + class VideoProcessor: """Processes video input from webcam or other sources""" - - def __init__(self, width: int = 640, height: int = 480, fps: int = 5): + + def __init__( + self, + width: int = 640, + height: int = 480, + fps: int = 5, + capture_backend: Optional[VideoCaptureBackend] = None, + ): """ Initialize the video processor @@ -276,12 +617,13 @@ def __init__(self, width: int = 640, height: int = 480, fps: int = 5): self.is_capturing = False self.capture_thread = None - # OpenCV capture object - self.cap = None - + self.capture_backend: Optional[VideoCaptureBackend] = capture_backend + if self.capture_backend is None and SENSORY_IMPORTS_AVAILABLE and cv2 is not None: + self.capture_backend = OpenCVCaptureBackend() + # Initialize face detection if OpenCV is available self.face_cascade = None - if SENSORY_IMPORTS_AVAILABLE: + if SENSORY_IMPORTS_AVAILABLE and cv2 is not None: try: # Load the pre-trained face cascade classifier self.face_cascade = cv2.CascadeClassifier( @@ -297,25 +639,10 @@ def list_cameras(self) -> List[Dict[str, Any]]: Returns: List of camera information dictionaries """ - if not SENSORY_IMPORTS_AVAILABLE: + if not self.capture_backend: return [] - - cameras = [] - for i in range(10): # Try up to 10 camera indices - try: - cap = cv2.VideoCapture(i) - if cap.isOpened(): - cameras.append({ - 'index': i, - 'name': f"Camera {i}", - 'width': int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)), - 'height': int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)), - 'fps': int(cap.get(cv2.CAP_PROP_FPS)) - }) - cap.release() - except Exception: - pass - return cameras + + return self.capture_backend.list_cameras() def start_capture(self, camera_index: int = 0) -> bool: """ @@ -327,25 +654,20 @@ def start_capture(self, camera_index: int = 0) -> bool: Returns: Success flag """ - if not SENSORY_IMPORTS_AVAILABLE: - logger.error("Video processing dependencies not available") + if not self.capture_backend or not self.capture_backend.is_available: + logger.error("No video capture backend configured") return False - + if self.is_capturing: logger.warning("Already capturing video") return False - + try: # Initialize camera - self.cap = cv2.VideoCapture(camera_index) - if not self.cap.isOpened(): + if not self.capture_backend.start(camera_index, self.width, self.height, self.fps): logger.error(f"Failed to open camera {camera_index}") return False - - # Set resolution - self.cap.set(cv2.CAP_PROP_FRAME_WIDTH, self.width) - self.cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self.height) - + # Start capture thread self.is_capturing = True self.capture_thread = threading.Thread( @@ -359,9 +681,8 @@ def start_capture(self, camera_index: int = 0) -> bool: except Exception as e: logger.error(f"Error starting video capture: {e}") - if self.cap: - self.cap.release() - self.cap = None + if self.capture_backend: + self.capture_backend.stop() self.is_capturing = False return False @@ -381,9 +702,8 @@ def stop_capture(self) -> bool: self.capture_thread.join(timeout=2.0) self.capture_thread = None - if self.cap: - self.cap.release() - self.cap = None + if self.capture_backend: + self.capture_backend.stop() logger.info("Stopped video capture") return True @@ -392,17 +712,21 @@ def _capture_video_thread(self) -> None: """Thread function for continuous video capture""" last_frame_time = 0 - while self.is_capturing and self.cap and self.cap.isOpened(): + while self.is_capturing: try: # Maintain target frame rate current_time = time.time() if current_time - last_frame_time < self.frame_interval: time.sleep(0.001) # Short sleep to prevent CPU spin continue - + # Capture frame - ret, frame = self.cap.read() - if not ret: + if not self.capture_backend: + logger.error("No video capture backend available during capture") + break + + frame = self.capture_backend.read_frame() + if frame is None: logger.error("Failed to capture frame") time.sleep(0.1) # Prevent tight loop on errors continue @@ -410,7 +734,7 @@ def _capture_video_thread(self) -> None: # Process the frame in a separate thread threading.Thread( target=self._process_frame, - args=(frame.copy(),), # Copy to prevent race conditions + args=(frame.copy() if hasattr(frame, 'copy') else frame,), # Copy to prevent race conditions daemon=True ).start() @@ -419,73 +743,74 @@ def _capture_video_thread(self) -> None: except Exception as e: logger.error(f"Error in video capture: {e}") time.sleep(0.1) # Prevent tight loop on errors - + + if self.capture_backend: + self.capture_backend.stop() + def _process_frame(self, frame: Any) -> None: """ Process a captured video frame - + Args: frame: Video frame to process """ try: - # Detect faces - faces = [] - if self.face_cascade is not None: - # Convert to grayscale for face detection - gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) - - # Detect faces - detected_faces = self.face_cascade.detectMultiScale( - gray, - scaleFactor=1.1, - minNeighbors=5, - minSize=(30, 30) - ) - - # Process detected faces - for (x, y, w, h) in detected_faces: - faces.append({ - 'x': int(x), - 'y': int(y), - 'width': int(w), - 'height': int(h) - }) - - # Extract basic image features - # Calculate average brightness - brightness = np.mean(frame) - - # Calculate color distribution - if frame.shape[2] == 3: # Check if frame has 3 color channels - color_means = [ - float(np.mean(frame[:, :, 0])), # Blue - float(np.mean(frame[:, :, 1])), # Green - float(np.mean(frame[:, :, 2])) # Red - ] + event: Dict[str, Any] + if np is not None and cv2 is not None and hasattr(frame, "shape"): + faces = [] + if self.face_cascade is not None: + gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) + detected_faces = self.face_cascade.detectMultiScale( + gray, + scaleFactor=1.1, + minNeighbors=5, + minSize=(30, 30) + ) + + for (x, y, w, h) in detected_faces: + faces.append({ + 'x': int(x), + 'y': int(y), + 'width': int(w), + 'height': int(h) + }) + + brightness = float(np.mean(frame)) + if len(getattr(frame, "shape", [])) >= 3 and frame.shape[2] == 3: + color_means = [ + float(np.mean(frame[:, :, 0])), + float(np.mean(frame[:, :, 1])), + float(np.mean(frame[:, :, 2])) + ] + else: + color_means = [brightness] + + thumbnail = cv2.resize(frame, (160, 120)) + _, jpeg_data = cv2.imencode('.jpg', thumbnail, [cv2.IMWRITE_JPEG_QUALITY, 70]) + thumbnail_b64 = base64.b64encode(jpeg_data).decode('utf-8') + resolution = { + 'width': int(frame.shape[1]), + 'height': int(frame.shape[0]) + } else: - color_means = [float(brightness)] - - # Create a thumbnail for visualizing - thumbnail = cv2.resize(frame, (160, 120)) - _, jpeg_data = cv2.imencode('.jpg', thumbnail, [cv2.IMWRITE_JPEG_QUALITY, 70]) - thumbnail_b64 = base64.b64encode(jpeg_data).decode('utf-8') - - # Create event object + faces = [] + color_means = [] + brightness = 0.0 + thumbnail_b64 = None + resolution = {} + event = { 'type': 'video_frame', 'timestamp': datetime.now().isoformat(), 'features': { - 'brightness': float(brightness), + 'brightness': brightness, 'color_means': color_means, 'faces': faces, - 'resolution': { - 'width': frame.shape[1], - 'height': frame.shape[0] - } + 'resolution': resolution }, 'thumbnail': thumbnail_b64 } - + # Add to queue (non-blocking to prevent slowdowns) try: self.video_queue.put(event, block=False) @@ -503,7 +828,7 @@ def _process_frame(self, frame: Any) -> None: def get_next_video_event(self, timeout: Optional[float] = 0.1) -> Optional[Dict[str, Any]]: """ Get the next video event from the queue - + Args: timeout: Timeout in seconds (None to block indefinitely) @@ -515,14 +840,25 @@ def get_next_video_event(self, timeout: Optional[float] = 0.1) -> Optional[Dict[ except queue.Empty: return None + def use_backend(self, backend: Optional[VideoCaptureBackend]) -> None: + """Swap the capture backend implementation.""" + if self.capture_backend and self.capture_backend is not backend: + self.capture_backend.stop() + self.capture_backend = backend + + def shutdown(self) -> None: + """Release backend resources.""" + if self.capture_backend: + self.capture_backend.stop() + class SensoryInputModule: """Module for processing sensory inputs (audio, video, etc.)""" def __init__(self): """Initialize the sensory input module""" - self.audio_processor = AudioProcessor() if SENSORY_IMPORTS_AVAILABLE else None - self.video_processor = VideoProcessor() if SENSORY_IMPORTS_AVAILABLE else None + self.audio_processor = AudioProcessor() + self.video_processor = VideoProcessor() # Callbacks for processing events self.event_callbacks = [] @@ -530,12 +866,23 @@ def __init__(self): # Event processing thread self.processing_thread = None self.is_processing = False - + # Event buffer for batch processing self.event_buffer = [] self.buffer_lock = threading.Lock() + + # Multimodal grounding helpers + self.fusion_engine = MultimodalFusionEngine() + self.cross_modal_reasoner = CrossModalReasoner(self.fusion_engine) + self._latest_audio_event: Optional[Dict[str, Any]] = None + self._latest_video_event: Optional[Dict[str, Any]] = None + self._fused_events: deque = deque(maxlen=50) - logger.info(f"Initialized SensoryInputModule (dependencies available: {SENSORY_IMPORTS_AVAILABLE})") + logger.info( + "Initialized SensoryInputModule (audio backend available: %s, video backend available: %s)", + self.audio_processor.is_available, + bool(self.video_processor.capture_backend and self.video_processor.capture_backend.is_available), + ) def get_tools(self) -> Dict[str, Any]: """Get tools provided by this module""" @@ -628,6 +975,14 @@ def get_tools(self) -> Dict[str, Any]: "type": "object", "properties": {} } + }, + "get_multimodal_context": { + "description": "Fuse recent audio/video into a unified context", + "function": self.get_multimodal_context, + "parameters": { + "type": "object", + "properties": {} + } } } @@ -641,12 +996,12 @@ def start_audio_recording(self, device_index: Optional[int] = None) -> Dict[str, Returns: Dictionary with result information """ - if not SENSORY_IMPORTS_AVAILABLE or not self.audio_processor: + if not self.audio_processor.is_available: return { "success": False, - "error": "Audio processing dependencies not available" + "error": "Audio capture backend not available" } - + success = self.audio_processor.start_recording(device_index) return { "success": success, @@ -660,12 +1015,12 @@ def stop_audio_recording(self) -> Dict[str, Any]: Returns: Dictionary with result information """ - if not SENSORY_IMPORTS_AVAILABLE or not self.audio_processor: + if not self.audio_processor.is_available: return { "success": False, - "error": "Audio processing dependencies not available" + "error": "Audio capture backend not available" } - + success = self.audio_processor.stop_recording() return { "success": success, @@ -679,12 +1034,12 @@ def list_audio_devices(self) -> Dict[str, Any]: Returns: Dictionary with audio device information """ - if not SENSORY_IMPORTS_AVAILABLE or not self.audio_processor: + if not self.audio_processor.is_available: return { "success": False, - "error": "Audio processing dependencies not available" + "error": "Audio capture backend not available" } - + devices = self.audio_processor.devices return { "success": True, @@ -702,12 +1057,13 @@ def start_video_capture(self, camera_index: int = 0) -> Dict[str, Any]: Returns: Dictionary with result information """ - if not SENSORY_IMPORTS_AVAILABLE or not self.video_processor: + backend = self.video_processor.capture_backend + if not backend or not backend.is_available: return { "success": False, - "error": "Video processing dependencies not available" + "error": "Video capture backend not available" } - + success = self.video_processor.start_capture(camera_index) return { "success": success, @@ -721,12 +1077,13 @@ def stop_video_capture(self) -> Dict[str, Any]: Returns: Dictionary with result information """ - if not SENSORY_IMPORTS_AVAILABLE or not self.video_processor: + backend = self.video_processor.capture_backend + if not backend or not backend.is_available: return { "success": False, - "error": "Video processing dependencies not available" + "error": "Video capture backend not available" } - + success = self.video_processor.stop_capture() return { "success": success, @@ -736,22 +1093,72 @@ def stop_video_capture(self) -> Dict[str, Any]: def list_cameras(self) -> Dict[str, Any]: """ List available camera devices - + Returns: Dictionary with camera information """ - if not SENSORY_IMPORTS_AVAILABLE or not self.video_processor: + backend = self.video_processor.capture_backend + if not backend: return { "success": False, - "error": "Video processing dependencies not available" + "error": "Video capture backend not available" } - + cameras = self.video_processor.list_cameras() return { "success": True, "cameras": cameras, "count": len(cameras) } + + def configure_audio_backend(self, backend: Optional[AudioCaptureBackend]) -> Dict[str, Any]: + """Attach a new audio backend implementation.""" + self.audio_processor.use_backend(backend) + return { + "success": True, + "available": self.audio_processor.is_available + } + + def configure_video_backend(self, backend: Optional[VideoCaptureBackend]) -> Dict[str, Any]: + """Attach a new video backend implementation.""" + self.video_processor.use_backend(backend) + return { + "success": True, + "available": bool(self.video_processor.capture_backend and self.video_processor.capture_backend.is_available) + } + + def check_audio_availability(self) -> Dict[str, Any]: + """Return audio backend availability information.""" + devices = self.audio_processor.devices if self.audio_processor.is_available else [] + return { + "available": self.audio_processor.is_available, + "device_count": len(devices), + "devices": devices + } + + def check_video_availability(self) -> Dict[str, Any]: + """Return video backend availability information.""" + backend = self.video_processor.capture_backend + cameras = backend.list_cameras() if backend else [] + return { + "available": bool(backend and backend.is_available), + "camera_count": len(cameras), + "cameras": cameras + } + + def add_test_event(self, event: Dict[str, Any]) -> None: + """Inject a synthetic event for testing or simulation.""" + event.setdefault("timestamp", datetime.now().isoformat()) + with self.buffer_lock: + self.event_buffer.append(event) + if len(self.event_buffer) > 100: + self.event_buffer = self.event_buffer[-100:] + + for callback in list(self.event_callbacks): + try: + callback(event) + except Exception as exc: + logger.error(f"Error in injected event callback: {exc}") def register_event_callback(self, callback: Callable[[Dict[str, Any]], None]) -> None: """ @@ -791,13 +1198,7 @@ def start_event_processing(self) -> Dict[str, Any]: "success": False, "message": "Event processing already running" } - - if not SENSORY_IMPORTS_AVAILABLE: - return { - "success": False, - "error": "Sensory processing dependencies not available" - } - + self.is_processing = True self.processing_thread = threading.Thread( target=self._event_processing_thread, @@ -856,7 +1257,9 @@ def _event_processing_thread(self) -> None: # Limit buffer size if len(self.event_buffer) > 100: self.event_buffer = self.event_buffer[-100:] - + self._latest_audio_event = audio_event + self._try_fuse_events() + # Get video events if self.video_processor: video_event = self.video_processor.get_next_video_event(timeout=0.01) @@ -874,6 +1277,8 @@ def _event_processing_thread(self) -> None: # Limit buffer size if len(self.event_buffer) > 100: self.event_buffer = self.event_buffer[-100:] + self._latest_video_event = video_event + self._try_fuse_events() # Short sleep to prevent tight loop time.sleep(0.01) @@ -885,18 +1290,43 @@ def _event_processing_thread(self) -> None: def get_latest_sensory_events(self, max_events: int = 10) -> Dict[str, Any]: """ Get the latest sensory events (audio, video, etc.) - + Args: max_events: Maximum number of events to return - + Returns: Dictionary with sensory events """ with self.buffer_lock: events = self.event_buffer[-max_events:] if self.event_buffer else [] - + return { "success": True, "events": events, "count": len(events) } + + def get_multimodal_context(self) -> Dict[str, Any]: + """Return the latest fused sensory context and reasoning.""" + if not self._fused_events: + return {"success": False, "error": "Insufficient data for fusion"} + fused_event = self._fused_events[-1] + reasoning = self.cross_modal_reasoner.infer_context(fused_event) + return {"success": True, "fused": fused_event, "reasoning": reasoning} + + def _try_fuse_events(self) -> None: + if self._latest_audio_event is None and self._latest_video_event is None: + return + fused = self.fusion_engine.fuse( + audio_event=self._latest_audio_event, + video_event=self._latest_video_event, + ) + self._fused_events.append(fused) + + def shutdown(self) -> None: + """Release resources used by the sensory processors.""" + if self.audio_processor: + self.audio_processor.shutdown() + if self.video_processor: + self.video_processor.shutdown() + self._fused_events.clear() diff --git a/AgentSystem/tests/test_agent_forge.py b/AgentSystem/tests/test_agent_forge.py new file mode 100644 index 0000000..6f005cc --- /dev/null +++ b/AgentSystem/tests/test_agent_forge.py @@ -0,0 +1,156 @@ +"""Tests for AgentForge developer framework.""" + +import json +import shutil +import tempfile +import unittest +from pathlib import Path +from typing import Any, Dict, List + +from AgentSystem.modules.agent_forge import ( + AgentForgeRegistry, + AgentForgeSDK, + KnowledgeExchange, + ModuleDescriptor, +) + + +class _StubKnowledgeManager: + def __init__(self) -> None: + self.facts: List[Dict[str, Any]] = [] + + def add_fact(self, content: str, source: str = "", category: str = "") -> int: + self.facts.append({"content": content, "source": source, "category": category}) + return len(self.facts) + + +class TestAgentForgeRegistry(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.mkdtemp() + self.registry_path = Path(self.temp_dir) / "registry" + self.registry = AgentForgeRegistry(self.registry_path) + + def tearDown(self) -> None: + shutil.rmtree(self.temp_dir) + + def test_register_and_list_modules(self) -> None: + descriptor = ModuleDescriptor( + name="vision-enhancer", + version="1.0.0", + summary="Adds advanced vision filters", + author="tester", + capabilities=["vision"], + tags=["vision", "beta"], + ) + self.registry.register(descriptor) + + modules = self.registry.list_modules() + self.assertEqual(len(modules), 1) + self.assertEqual(modules[0].name, "vision-enhancer") + + tagged = self.registry.list_modules(tag="vision") + self.assertEqual(len(tagged), 1) + self.assertEqual(tagged[0].key(), descriptor.key()) + + self.assertEqual(self.registry.get("vision-enhancer"), descriptor) + + def test_register_duplicate_without_overwrite_errors(self) -> None: + descriptor = ModuleDescriptor( + name="vision-enhancer", + version="1.0.0", + summary="Adds advanced vision filters", + ) + self.registry.register(descriptor) + with self.assertRaises(ValueError): + self.registry.register(descriptor) + + def test_persistence_round_trip(self) -> None: + descriptor = ModuleDescriptor( + name="vision-enhancer", + version="1.0.0", + summary="Adds advanced vision filters", + ) + self.registry.register(descriptor) + raw = json.loads((self.registry_path / "registry.json").read_text()) + self.assertIn("modules", raw) + self.assertEqual(raw["modules"][0]["name"], "vision-enhancer") + + reloaded = AgentForgeRegistry(self.registry_path) + self.assertEqual(reloaded.get("vision-enhancer"), descriptor) + + +class TestKnowledgeExchange(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.mkdtemp() + self.exchange_path = Path(self.temp_dir) / "exchange" + self.exchange = KnowledgeExchange(self.exchange_path) + + def tearDown(self) -> None: + shutil.rmtree(self.temp_dir) + + def test_publish_and_query(self) -> None: + self.exchange.publish( + title="Vision tips", + content="Use histogram equalisation", + authors=["tester"], + tags=["vision"], + ) + self.exchange.publish( + title="Audio tips", + content="Use noise reduction", + authors=["tester"], + tags=["audio"], + ) + + all_entries = self.exchange.query() + self.assertEqual(len(all_entries), 2) + + vision_entries = self.exchange.query(tag="vision") + self.assertEqual(len(vision_entries), 1) + self.assertEqual(vision_entries[0]["title"], "Vision tips") + + limited = self.exchange.query(limit=1) + self.assertEqual(len(limited), 1) + + +class TestAgentForgeSDK(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.mkdtemp() + storage = Path(self.temp_dir) + registry = AgentForgeRegistry(storage / "registry") + exchange = KnowledgeExchange(storage / "exchange") + self.knowledge_manager = _StubKnowledgeManager() + self.sdk = AgentForgeSDK( + registry=registry, + exchange=exchange, + knowledge_manager=self.knowledge_manager, + ) + + def tearDown(self) -> None: + shutil.rmtree(self.temp_dir) + + def test_publish_module_and_share_knowledge(self) -> None: + descriptor = ModuleDescriptor( + name="research-booster", + version="0.1.0", + summary="Improves research heuristics", + tags=["research"], + ) + self.sdk.publish_module(descriptor) + fetched = self.sdk.fetch_modules() + self.assertEqual(len(fetched), 1) + self.assertEqual(fetched[0].name, "research-booster") + + self.sdk.share_knowledge( + "Research handbook", + "Always cite your sources", + authors=["mentor"], + tags=["research", "best-practices"], + ) + knowledge = self.sdk.retrieve_knowledge(tag="research") + self.assertEqual(len(knowledge), 1) + self.assertEqual(self.knowledge_manager.facts[0]["category"], "research") + + +if __name__ == "__main__": # pragma: no cover - manual execution + unittest.main() diff --git a/AgentSystem/tests/test_env_loader.py b/AgentSystem/tests/test_env_loader.py new file mode 100644 index 0000000..d38c24f --- /dev/null +++ b/AgentSystem/tests/test_env_loader.py @@ -0,0 +1,80 @@ +import os +from pathlib import Path + +import pytest + +from AgentSystem.utils import env_loader + + +@pytest.fixture(autouse=True) +def restore_environment(): + original_env = os.environ.copy() + yield + # Restore environment to avoid leaking state between tests + os.environ.clear() + os.environ.update(original_env) + + +@pytest.fixture +def temp_env_file(tmp_path: Path) -> Path: + env_file = tmp_path / ".env" + env_file.write_text( + "\n".join( + [ + "# comment line should be ignored", + "BASIC=value", + "INLINE=needs_comment # trailing comment should be stripped", + "WITH_COLON: colon value", + "QUOTED_DOUBLE=\"quoted # still value\"", + "QUOTED_SINGLE='single # still value'", + "WITH_ESCAPE=hash\\#should stay", + "WITH_INTERPOLATION=${BASIC}_suffix", + "DOUBLE_INTERPOLATION=\"${BASIC}_${INLINE}\"", + "SINGLE_INTERPOLATION='${BASIC}_${INLINE}'", + "ESCAPED_INTERPOLATION=\\${INLINE}", + "ESCAPED_DOLLAR=\"Cost is \\$5\"", + "export EXPORTED=from_export", + "PRESERVED=from_file", + "OVERWRITE=first", + "OVERWRITE=second", + ] + ) + ) + return env_file + + +def test_manual_loader_parses_various_patterns(temp_env_file: Path) -> None: + keys_to_check = { + "BASIC": "value", + "INLINE": "needs_comment", + "WITH_COLON": "colon value", + "QUOTED_DOUBLE": "quoted # still value", + "QUOTED_SINGLE": "single # still value", + "WITH_ESCAPE": "hash#should stay", + "WITH_INTERPOLATION": "value_suffix", + "DOUBLE_INTERPOLATION": "value_needs_comment", + "SINGLE_INTERPOLATION": "${BASIC}_${INLINE}", + "ESCAPED_INTERPOLATION": "${INLINE}", + "ESCAPED_DOLLAR": "Cost is $5", + "EXPORTED": "from_export", + "OVERWRITE": "second", + } + + os.environ["PRESERVED"] = "already set" + + original_load_dotenv = env_loader.load_dotenv + env_loader.load_dotenv = None + try: + loader = env_loader.EnvLoader(env_file=str(temp_env_file)) + finally: + env_loader.load_dotenv = original_load_dotenv + + for key, expected in keys_to_check.items(): + assert os.environ.get(key) == expected + assert loader.get(key) == expected + + # Ensure pre-existing environment variables are not clobbered + assert os.environ.get("PRESERVED") == "already set" + + # Accessing via loader should return preserved value + assert loader.get("PRESERVED") == "already set" diff --git a/AgentSystem/tests/test_learning_system.py b/AgentSystem/tests/test_learning_system.py index 200bf7e..6440e6b 100644 --- a/AgentSystem/tests/test_learning_system.py +++ b/AgentSystem/tests/test_learning_system.py @@ -4,16 +4,51 @@ Unit tests for the learning system components. """ +import json +import importlib.util import unittest import tempfile import shutil from pathlib import Path +from typing import Any, Dict, List, Optional from unittest.mock import Mock, patch -from AgentSystem.modules.knowledge_manager import KnowledgeManager -from AgentSystem.modules.web_researcher import WebResearcher -from AgentSystem.modules.code_modifier import CodeModifier -from AgentSystem.modules.learning_agent import LearningAgent +MODULE_DIR = Path(__file__).resolve().parents[1] / "modules" + + +def _load_module(module_name: str): + module_path = MODULE_DIR / f"{module_name}.py" + spec = importlib.util.spec_from_file_location(f"test_{module_name}", module_path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +knowledge_manager_module = _load_module("knowledge_manager") +learning_agent_module = _load_module("learning_agent") + +KnowledgeManager = knowledge_manager_module.KnowledgeManager +LearningAgent = learning_agent_module.LearningAgent +ReflexRule = learning_agent_module.ReflexRule + +try: + web_researcher_module = _load_module("web_researcher") + WebResearcher = web_researcher_module.WebResearcher +except Exception: # pragma: no cover - optional dependency path + WebResearcher = None + +WEB_IMPORTS_AVAILABLE = bool( + getattr(web_researcher_module, "WEB_IMPORTS_AVAILABLE", False) +) if 'web_researcher_module' in locals() else False + +try: + code_modifier_module = _load_module("code_modifier") + CodeModifier = code_modifier_module.CodeModifier +except Exception: # pragma: no cover - optional dependency path + CodeModifier = None + +CODE_MODIFIER_AVAILABLE = CodeModifier is not None class TestKnowledgeManager(unittest.TestCase): def setUp(self): @@ -59,13 +94,68 @@ def test_search_facts(self): content="JavaScript runs in browsers", category="programming" ) - + # Search results = self.knowledge_manager.search_facts("Python") - + self.assertEqual(len(results), 1) self.assertIn("Python", results[0]["content"]) + def test_episodic_memory_and_fusion(self): + episode_id = self.knowledge_manager.add_episode( + event="Investigated water behaviour", + outcome="understood flow", + emotion="curious", + salience=0.9, + context={"topic": "water"}, + ) + self.assertGreater(episode_id, 0) + + fused = self.knowledge_manager.fusion_search("water") + self.assertTrue(fused["episodes"]) + + promoted = self.knowledge_manager.consolidate_memories(limit=1) + self.assertIsInstance(promoted, list) + + def test_knowledge_synthesis_and_verification(self): + self.knowledge_manager.add_fact("Water flows downhill", category="science") + self.knowledge_manager.add_fact("Water flows through rivers", category="science") + + graph = self.knowledge_manager.synthesize_knowledge("water") + self.assertIn("nodes", graph) + self.assertTrue(graph["nodes"]) + + hypotheses = self.knowledge_manager.generate_hypotheses("water") + self.assertTrue(hypotheses) + + verdict = self.knowledge_manager.verify_claim("Water flows") + self.assertIn(verdict["verdict"], {"supported", "partial", "unknown"}) + + def test_source_trust_calibration(self): + source = "http://example.com" + baseline = self.knowledge_manager.get_source_trust(source) + self.assertEqual(baseline["score"], 0.5) + + self.knowledge_manager.update_source_trust(source, success=True) + increased = self.knowledge_manager.get_source_trust(source) + self.assertGreater(increased["score"], baseline["score"]) + self.assertEqual(increased["success_count"], 1) + + self.knowledge_manager.update_source_trust(source, success=False, weight=2.0) + reduced = self.knowledge_manager.get_source_trust(source) + self.assertLess(reduced["score"], increased["score"]) + self.assertEqual(reduced["failure_count"], 1) + + def test_integrity_check_and_recovery(self): + status = self.knowledge_manager.integrity_check() + self.assertEqual(status["status"], "ok") + + recovery = self.knowledge_manager.recover_integrity() + self.assertEqual(recovery["status"], "reset") + post = self.knowledge_manager.integrity_check() + self.assertEqual(post["status"], "ok") + +@unittest.skipIf(not WEB_IMPORTS_AVAILABLE, "Web researcher dependencies unavailable") class TestWebResearcher(unittest.TestCase): def setUp(self): """Set up test web researcher""" @@ -101,6 +191,7 @@ def test_search(self): self.assertEqual(results[0]["url"], "http://test.com") self.assertEqual(results[0]["snippet"], "Test snippet") +@unittest.skipIf(not CODE_MODIFIER_AVAILABLE, "Code modifier dependencies unavailable") class TestCodeModifier(unittest.TestCase): def setUp(self): """Set up test code modifier""" @@ -167,13 +258,231 @@ def test_background_learning(self): """Test background learning queue""" self.agent.start_learning() self.assertTrue(self.agent.learning_thread.is_alive()) - + self.agent.queue_research("test topic") self.assertEqual(self.agent.learning_queue.qsize(), 1) - + self.agent.stop_learning() self.assertFalse(self.agent.learning_active) + def test_reward_tracking_metrics(self): + """Ensure reward metrics update when feedback is recorded.""" + baseline = self.agent.get_learning_feedback() + self.assertEqual(baseline["cumulative_reward"], 0.0) + self.assertEqual(baseline["total_tasks"], 0) + + self.agent.submit_feedback(0.5, note="good progress") + updated = self.agent.get_learning_feedback() + self.assertAlmostEqual(updated["cumulative_reward"], 0.5) + self.assertEqual(updated["total_tasks"], 1) + self.assertGreaterEqual(updated["success_rate"], 0.0) + + def test_cognition_layers_and_meta_review(self): + triggered: List[Dict[str, Any]] = [] + self.agent.cognition.reflex.register_rule( + ReflexRule(trigger="alert", action=lambda event: triggered.append(event)) + ) + response = self.agent.process_event({"type": "alert", "reward": 0.2}) + self.assertTrue(response["handled"]) + self.assertTrue(triggered) + + planning = self.agent.process_event({"type": "observation", "goal": "research"}) + self.assertIn("plan", planning) + + review = self.agent.meta_review() + self.assertIn("status", review) + + def test_react_reasoning_and_self_play(self): + self.agent.knowledge_manager.add_fact("Water analysis complete", category="research") + reason = self.agent.react_reason("water") + self.assertIn("trace", reason) + self.assertIn("memory", reason) + + simulation = self.agent.simulate_self_play("maze-run") + self.assertIn(simulation["outcome"], {"win", "loss"}) + + def test_distributed_mesh_and_routing(self): + messages: List[Dict[str, Any]] = [] + self.agent.join_mesh("Planner", lambda payload: messages.append(payload)) + self.agent.share_mesh_state("goal", "expand-knowledge") + self.agent.mesh.broadcast({"event": "test"}) + self.assertTrue(messages) + self.assertEqual(self.agent.mesh.get_shared_state("goal"), "expand-knowledge") + + self.agent.register_local_inference(lambda task, payload: {"response": "ok"}) + self.assertEqual(self.agent.choose_inference_path("analysis"), "local") + + def test_hybrid_inference_execution(self): + calls = {"local": 0, "cloud": 0} + + def local_handler(task: str, payload: Dict[str, Any]) -> Dict[str, Any]: + calls["local"] += 1 + if task == "fail-local": + raise RuntimeError("local offline") + return {"response": f"local::{payload['prompt']}", "reward": 0.25} + + def cloud_handler(task: str, payload: Dict[str, Any]) -> Dict[str, Any]: + calls["cloud"] += 1 + return {"response": f"cloud::{payload['prompt']}", "reward": 0.1} + + self.agent.register_local_inference(local_handler) + self.agent.register_cloud_inference(cloud_handler) + + first = self.agent.generate_inference("analysis", "diagnose") + self.assertEqual(first["path"], "local") + self.assertIn("local::diagnose", first["result"]["response"]) + self.assertEqual(calls["local"], 1) + + fallback = self.agent.generate_inference("fail-local", "diagnose") + self.assertEqual(fallback["path"], "cloud") + self.assertEqual(calls["cloud"], 1) + + self.agent.register_local_inference(None) + self.agent.register_cloud_inference(None) + cached = self.agent.generate_inference("analysis", "diagnose", prefer_local=False) + self.assertEqual(cached["path"], "cached") + self.assertIsNotNone(cached["result"]) + + stats = self.agent.inference_router.statistics + self.assertGreaterEqual(stats["local"]["success"], 1) + self.assertGreaterEqual(stats["cloud"]["success"], 1) + self.assertGreaterEqual(stats["cached"]["hits"], 1) + + def test_consensus_planning(self): + def planner(payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if payload.get("kind") == "consensus_request": + options = payload.get("options", []) + if options: + choice = options[-1] + return {"vote": choice.get("id"), "weight": 2.0, "note": "Prefer thorough plan"} + return None + + def executor(payload: Dict[str, Any]) -> Optional[Dict[str, Any]]: + if payload.get("kind") == "consensus_request": + options = payload.get("options", []) + if options: + return {"vote": options[0].get("id")} + return None + + self.agent.join_mesh("Planner", planner, weight=2.0) + self.agent.join_mesh("Executor", executor) + + options = [ + {"id": "option-a", "expected_reward": 0.2, "steps": ["quick-draft"]}, + {"id": "option-b", "expected_reward": 0.8, "steps": ["deploy-safely"]}, + ] + + decision = self.agent.consensus_plan("deploy-update", options) + self.assertEqual(decision["plan"], ["deploy-safely"]) + self.assertTrue(decision["consensus"]["passed"]) + self.assertEqual(decision["consensus"]["decision"], "option-b") + self.assertGreaterEqual(decision["consensus"]["decision_weight"], 2.0) + + # Require a higher quorum so the deliberative fallback is used + fallback = self.agent.consensus_plan("deploy-update", options, quorum=10.0) + self.assertFalse(fallback["consensus"]["passed"]) + self.assertEqual(fallback["plan"], ["deploy-safely"]) + + def test_social_and_memory_enrichment(self): + adapted = self.agent.adapt_response_tone("This is great progress") + self.assertEqual(adapted["analysis"]["sentiment"], "positive") + + baseline = self.agent.knowledge_manager.contextual_recall("mission") + self.agent.register_observation("Mission accomplished", success=True, emotion="proud") + recall = self.agent.knowledge_manager.contextual_recall("Mission") + self.assertGreaterEqual(len(recall), len(baseline)) + + def test_reward_updates_source_trust(self): + source = "http://trust.example" + baseline = self.agent.knowledge_manager.get_source_trust(source)["score"] + self.agent._record_reward( + "research", + reward=1.0, + success=True, + processing_time=0.5, + details={"sources": [source]}, + ) + increased = self.agent.knowledge_manager.get_source_trust(source)["score"] + self.assertGreaterEqual(increased, baseline) + + self.agent._record_reward( + "research", + reward=-1.0, + success=False, + processing_time=0.5, + details={"sources": [source]}, + ) + reduced = self.agent.knowledge_manager.get_source_trust(source)["score"] + self.assertLess(reduced, increased) + + def test_prompt_evolution_and_distillation(self): + self.agent.register_prompt("research", "Base prompt with {query} context") + + first_state = self.agent.evolve_prompts("research", success=False, notes="Missed citations") + self.assertEqual(first_state["version"], 1) + + evolved_state = self.agent.evolve_prompts( + "research", + success=False, + notes="Needs deeper analysis", + ) + self.assertGreaterEqual(evolved_state["version"], 2) + + template = self.agent.get_prompt_template("research") + self.assertIsNotNone(template) + self.assertIn("[Adjustment v", template) + + cached = self.agent.inference_router.cached_prompts.get("research") + self.assertIsNotNone(cached) + self.assertEqual(cached["version"], evolved_state["version"]) + + self.agent.meta_layer.record_outcome({"reward": -1.0}) + self.agent.meta_layer.record_outcome({"reward": -0.5}) + auto_keys = self.agent.auto_evolve_prompts() + self.assertIn("research", auto_keys) + + self.agent.log_interaction("research", template or "", "response", reward=0.25) + output_path = Path(self.temp_dir) / "distilled.jsonl" + summary = self.agent.distill_model(output_path) + self.assertEqual(summary["samples"], 1) + self.assertTrue(output_path.exists()) + + lines = [line for line in output_path.read_text().splitlines() if line] + self.assertEqual(len(lines), 1) + record = json.loads(lines[0]) + self.assertEqual(record["prompt_id"], "research") + self.assertAlmostEqual(record["reward"], 0.25) + + def test_resilience_health_checks(self): + checks = self.agent.perform_health_checks() + self.assertIn("knowledge_base", checks) + self.assertEqual(checks["knowledge_base"]["status"], "ok") + + recoveries = {"count": 0} + + def failing_check(): + raise RuntimeError("boom") + + def recovery_hook(): + recoveries["count"] += 1 + + self.agent.resilience.register_health_check( + "dummy", + failing_check, + recover=recovery_hook, + threshold=2, + ) + + outcome = self.agent.resilience.run_health_checks() + self.assertEqual(outcome["dummy"]["status"], "error") + first = self.agent.resilience.record_failure("dummy") + self.assertFalse(first["triggered"]) + second = self.agent.resilience.record_failure("dummy") + self.assertTrue(second["triggered"]) + self.assertGreaterEqual(recoveries["count"], 1) + self.agent.resilience.record_success("dummy") + self.assertEqual(self.agent.resilience.get_failure_count("dummy"), 0) + def main(): unittest.main() diff --git a/AgentSystem/tests/test_sensory_abstraction.py b/AgentSystem/tests/test_sensory_abstraction.py new file mode 100644 index 0000000..76bb3ac --- /dev/null +++ b/AgentSystem/tests/test_sensory_abstraction.py @@ -0,0 +1,170 @@ +"""Tests for hardware abstraction in the sensory input module.""" + +import importlib.util +import time +from pathlib import Path +from typing import Any, Dict, List, Optional + +MODULE_PATH = Path(__file__).resolve().parents[1] / "modules" / "sensory_input.py" +spec = importlib.util.spec_from_file_location("sensory_input_module", MODULE_PATH) +sensory_input = importlib.util.module_from_spec(spec) +assert spec.loader is not None +spec.loader.exec_module(sensory_input) + +AudioCaptureBackend = sensory_input.AudioCaptureBackend +AudioProcessor = sensory_input.AudioProcessor +SensoryInputModule = sensory_input.SensoryInputModule +VideoCaptureBackend = sensory_input.VideoCaptureBackend +VideoProcessor = sensory_input.VideoProcessor + + +class DummyAudioBackend(AudioCaptureBackend): + """Synthetic audio backend that emits a few deterministic chunks.""" + + def __init__(self, iterations: int = 60) -> None: + self.iterations = iterations + self._running = False + self.sample_rate = 16000 + + @property + def is_available(self) -> bool: + return True + + def start(self, sample_rate: int, chunk_size: int, device_index: Optional[int] = None) -> bool: + self.sample_rate = sample_rate + self._running = True + self._remaining = self.iterations + self._chunk = (b"\x00\x01" * (chunk_size // 2)) or b"\x00\x01" + return True + + def read_chunk(self, chunk_size: int) -> bytes: + if not self._running: + return b"" + if self._remaining <= 0: + time.sleep(0.01) + return b"" + self._remaining -= 1 + return self._chunk + + def stop(self) -> None: + self._running = False + + def list_devices(self) -> List[Dict[str, Any]]: + return [ + { + "index": 0, + "name": "dummy-audio", + "channels": 1, + "sample_rate": self.sample_rate, + } + ] + + +class DummyVideoBackend(VideoCaptureBackend): + """Synthetic video backend that emits dictionary frames.""" + + def __init__(self, frames: Optional[List[Dict[str, Any]]] = None) -> None: + self.frames = frames or [{"frame": 1}, {"frame": 2}] + self._running = False + self._cursor = 0 + + @property + def is_available(self) -> bool: + return True + + def start(self, camera_index: int, width: int, height: int, fps: int) -> bool: + self._running = True + self._cursor = 0 + return True + + def read_frame(self) -> Optional[Any]: + if not self._running or self._cursor >= len(self.frames): + time.sleep(0.01) + return None + frame = self.frames[self._cursor] + self._cursor += 1 + return frame + + def stop(self) -> None: + self._running = False + + def list_cameras(self) -> List[Dict[str, Any]]: + return [{"index": 0, "name": "dummy-video"}] + + +def _drain_events(fetcher, timeout: float = 0.5) -> List[Dict[str, Any]]: + events: List[Dict[str, Any]] = [] + start = time.time() + while time.time() - start < timeout: + event = fetcher() + if event: + events.append(event) + else: + time.sleep(0.01) + return events + + +def test_audio_processor_with_dummy_backend() -> None: + backend = DummyAudioBackend() + processor = AudioProcessor(capture_backend=backend) + assert processor.is_available + + assert processor.start_recording() + time.sleep(0.2) + processor.stop_recording() + time.sleep(0.05) + + events = _drain_events(lambda: processor.get_next_audio_event(timeout=0.01)) + assert events, "Expected at least one audio event from dummy backend" + assert all("timestamp" in event for event in events) + + +def test_video_processor_with_dummy_backend() -> None: + backend = DummyVideoBackend() + processor = VideoProcessor(capture_backend=backend) + assert processor.capture_backend is backend + + assert processor.start_capture() + time.sleep(0.1) + processor.stop_capture() + + events = _drain_events(lambda: processor.get_next_video_event(timeout=0.01)) + assert events, "Expected at least one video event from dummy backend" + assert all(event.get("type") == "video_frame" for event in events) + + +def test_sensory_module_accepts_synthetic_backends() -> None: + module = SensoryInputModule() + module.configure_audio_backend(DummyAudioBackend()) + module.configure_video_backend(DummyVideoBackend()) + + audio_info = module.check_audio_availability() + video_info = module.check_video_availability() + + assert audio_info["available"] + assert video_info["available"] + + module.register_event_callback(lambda event: None) + module.start_event_processing() + module.add_test_event({"type": "synthetic"}) + time.sleep(0.1) + module.stop_event_processing() + + latest = module.get_latest_sensory_events() + assert latest["count"] >= 1 + module.shutdown() + + +def test_multimodal_context_generation() -> None: + module = SensoryInputModule() + module.configure_audio_backend(DummyAudioBackend(iterations=5)) + module.configure_video_backend(DummyVideoBackend(frames=[{"frame": 1}])) + + module._latest_audio_event = {"type": "speech", "raw_data": b"hello"} + module._latest_video_event = {"type": "video_frame", "objects": ["pool"], "frame_shape": (64, 64, 3)} + module._try_fuse_events() + + context = module.get_multimodal_context() + assert context["success"] + assert context["fused"]["embedding"], "Expected fused embedding values" + module.shutdown() diff --git a/AgentSystem/utils/env_loader.py b/AgentSystem/utils/env_loader.py index 20192b0..edb66d1 100644 --- a/AgentSystem/utils/env_loader.py +++ b/AgentSystem/utils/env_loader.py @@ -5,11 +5,18 @@ Provides fallback mechanisms and validation """ -import os +import ast import logging +import os +import re +import warnings from pathlib import Path -from typing import Dict, Any, Optional -from dotenv import load_dotenv +from typing import Any, Dict, Optional + +try: + from dotenv import load_dotenv # type: ignore +except ModuleNotFoundError: # pragma: no cover - fallback path for optional dependency + load_dotenv = None # type: ignore[assignment] logger = logging.getLogger(__name__) @@ -33,9 +40,158 @@ def _load_env_file(self) -> None: if not env_path.exists(): logger.warning(f"Environment file not found at {self.env_path}") return - + + if load_dotenv is None: + logger.info( + "python-dotenv is not installed; manually parsing %s", self.env_path + ) + self._load_env_file_manually(env_path) + return + load_dotenv(dotenv_path=self.env_path) logger.info(f"Loaded environment from {self.env_path}") + + def _load_env_file_manually(self, env_path: Path) -> None: + """Fallback parser when python-dotenv is not installed.""" + + def _strip_inline_comment(raw_value: str) -> str: + in_single = False + in_double = False + escape = False + result_chars = [] + + for char in raw_value: + if escape: + if char == "#" and not in_single and not in_double: + result_chars.append("#") + else: + result_chars.append("\\" + char) + escape = False + continue + + if char == "\\": + escape = True + continue + + if char == "'" and not in_double: + in_single = not in_single + result_chars.append(char) + continue + + if char == '"' and not in_single: + in_double = not in_double + result_chars.append(char) + continue + + if char == "#" and not in_single and not in_double: + break + + result_chars.append(char) + + if escape: + result_chars.append("\\") + + return "".join(result_chars).rstrip() + + interpolation_pattern = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}") + escaped_dollar_pattern = re.compile(r"\\(\$)") + + def _preserve_escaped_dollars(text: str) -> tuple[str, Dict[str, str]]: + placeholders: Dict[str, str] = {} + + def replace(match: re.Match[str]) -> str: + placeholder = f"__ESCAPED_DOLLAR_{len(placeholders)}__" + placeholders[placeholder] = match.group(1) + return placeholder + + return escaped_dollar_pattern.sub(replace, text), placeholders + + def _restore_escaped_dollars(text: str, placeholders: Dict[str, str]) -> str: + for placeholder, replacement in placeholders.items(): + text = text.replace(placeholder, replacement) + return text + + def _interpolate_value(raw_value: str, allow_interpolation: bool) -> str: + working, placeholders = _preserve_escaped_dollars(raw_value) + + if allow_interpolation and "${" in working: + previous = working + for _ in range(10): + replaced = interpolation_pattern.sub( + lambda match: os.environ.get(match.group(1), ""), previous + ) + if replaced == previous: + break + previous = replaced + working = previous + + return _restore_escaped_dollars(working, placeholders) + + preexisting_keys = set(os.environ.keys()) + + try: + with env_path.open("r", encoding="utf-8") as env_file: + for raw_line in env_file: + line = raw_line.strip() + if not line or line.startswith("#"): + continue + + delimiter = None + for candidate in ("=", ":"): + if candidate in line: + delimiter = candidate + break + + if delimiter is None: + logger.debug( + "Skipping malformed environment line in %s: %s", + env_path, + raw_line.rstrip("\n"), + ) + continue + + key, value = line.split(delimiter, 1) + key = key.strip() + stripped_value = _strip_inline_comment(value.strip()) + + if key.startswith("export "): + key = key[len("export ") :].strip() + + if not key: + logger.debug( + "Skipping environment line with empty key in %s: %s", + env_path, + raw_line.rstrip("\n"), + ) + continue + + is_quoted = ( + len(stripped_value) >= 2 + and stripped_value[0] == stripped_value[-1] + and stripped_value[0] in {'"', "'"} + ) + is_single_quoted = is_quoted and stripped_value[0] == "'" + + processed_value = stripped_value + + if is_quoted: + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + processed_value = ast.literal_eval(processed_value) + except (SyntaxError, ValueError): + processed_value = processed_value[1:-1] + + processed_value = _interpolate_value( + str(processed_value), allow_interpolation=not is_single_quoted + ) + + if key in preexisting_keys: + continue + + os.environ[key] = processed_value + except OSError as exc: + logger.error("Failed to read environment file %s: %s", env_path, exc) def get(self, key: str, default: Any = None, required: bool = False) -> Any: """