diff --git a/src/openjarvis/system.py b/src/openjarvis/system.py deleted file mode 100644 index bfb6c645..00000000 --- a/src/openjarvis/system.py +++ /dev/null @@ -1,1289 +0,0 @@ -"""Composition layer -- config-driven construction of a fully wired JarvisSystem.""" - -from __future__ import annotations - -import logging -from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional - -from openjarvis.core.config import JarvisConfig, load_config -from openjarvis.core.events import EventBus, get_event_bus -from openjarvis.core.types import Message, Role -from openjarvis.engine._stubs import InferenceEngine -from openjarvis.tools._stubs import BaseTool, ToolExecutor - -logger = logging.getLogger(__name__) - - -@dataclass -class JarvisSystem: - """Fully wired system -- the single source of truth for primitive composition.""" - - config: JarvisConfig - bus: EventBus - engine: InferenceEngine - engine_key: str - model: str - agent: Optional[Any] = None # BaseAgent - agent_name: str = "" - tools: List[BaseTool] = field(default_factory=list) - tool_executor: Optional[ToolExecutor] = None - memory_backend: Optional[Any] = None # MemoryBackend - channel_backend: Optional[Any] = None # BaseChannel - router: Optional[Any] = None # RouterPolicy - mcp_server: Optional[Any] = None # MCPServer - telemetry_store: Optional[Any] = None - trace_store: Optional[Any] = None - trace_collector: Optional[Any] = None - gpu_monitor: Optional[Any] = None - scheduler_store: Optional[Any] = None # SchedulerStore - scheduler: Optional[Any] = None # TaskScheduler - container_runner: Optional[Any] = None # ContainerRunner - workflow_engine: Optional[Any] = None # WorkflowEngine - session_store: Optional[Any] = None # SessionStore - capability_policy: Optional[Any] = None # CapabilityPolicy - audit_logger: Optional[Any] = None # AuditLogger - boundary_guard: Optional[Any] = None # BoundaryGuard - operator_manager: Optional[Any] = None # OperatorManager - agent_manager: Optional[Any] = None # AgentManager - agent_scheduler: Optional[Any] = None # AgentScheduler - agent_executor: Optional[Any] = None # AgentExecutor - speech_backend: Optional[Any] = None # SpeechBackend - skill_manager: Optional[Any] = None # SkillManager - _learning_orchestrator: Optional[Any] = None # LearningOrchestrator - _mcp_clients: List = field(default_factory=list) - - def ask( - self, - query: str, - *, - context: bool = True, - temperature: Optional[float] = None, - max_tokens: Optional[int] = None, - agent: Optional[str] = None, - tools: Optional[List[str]] = None, - system_prompt: Optional[str] = None, - operator_id: Optional[str] = None, - prior_messages: Optional[List[Message]] = None, - ) -> Dict[str, Any]: - """Execute a query through the system and return a result dict.""" - if temperature is None: - temperature = self.config.intelligence.temperature - if max_tokens is None: - max_tokens = self.config.intelligence.max_tokens - - messages = [Message(role=Role.USER, content=query)] - - # Context injection from memory - if context and self.memory_backend and self.config.agent.context_from_memory: - try: - from openjarvis.tools.storage.context import ( - ContextConfig, - inject_context, - ) - - ctx_cfg = ContextConfig( - top_k=self.config.memory.context_top_k, - min_score=self.config.memory.context_min_score, - max_context_tokens=self.config.memory.context_max_tokens, - ) - messages = inject_context( - query, - messages, - self.memory_backend, - config=ctx_cfg, - ) - except Exception as exc: - logger.warning("Failed to inject memory context: %s", exc) - - # Agent mode - use_agent = agent or self.agent_name - # Auto-detect agent intent if no explicit agent was requested - if not agent and use_agent != "none": - detected = self._detect_agent_intent(query) - if detected: - use_agent = detected - if use_agent and use_agent != "none": - return self._run_agent( - query, - messages, - use_agent, - tools, - temperature, - max_tokens, - system_prompt=system_prompt, - operator_id=operator_id, - prior_messages=prior_messages, - ) - - # Direct engine mode - result = self.engine.generate( - messages, - model=self.model, - temperature=temperature, - max_tokens=max_tokens, - ) - return { - "content": result.get("content", ""), - "usage": result.get("usage", {}), - "model": self.model, - "engine": self.engine_key, - } - - def _detect_agent_intent(self, query: str) -> Optional[str]: - """Detect if a query should be routed to a specific agent. - - Uses lightweight pattern matching for known intent triggers. - Returns the agent name or None to use the default. - """ - import re - - from openjarvis.core.registry import AgentRegistry - - # Morning digest triggers - if re.search( - r"\b(good\s+morning|morning\s+digest|daily\s+briefing|morning\s+briefing)\b", - query, - re.IGNORECASE, - ): - if AgentRegistry.contains("morning_digest"): - return "morning_digest" - - return None - - def _run_agent( - self, - query, - messages, - agent_name, - tool_names, - temperature, - max_tokens, - *, - system_prompt=None, - operator_id=None, - prior_messages=None, - ) -> Dict[str, Any]: - """Run through an agent.""" - from openjarvis.agents._stubs import AgentContext - from openjarvis.core.events import EventType - from openjarvis.core.registry import AgentRegistry - - # Resolve agent - try: - agent_cls = AgentRegistry.get(agent_name) - except KeyError: - return {"content": f"Unknown agent: {agent_name}", "error": True} - - # Build tools for agent - agent_tools = self.tools - if tool_names: - agent_tools = self._build_tools(tool_names) - - # Build context - ctx = AgentContext() - - # Seed prior conversation turns (channel session history) - if prior_messages: - for msg in prior_messages: - ctx.conversation.add(msg) - - # Inject memory context messages into the agent conversation - if messages and len(messages) > 1: - # Context messages were prepended by inject_context - for msg in messages[:-1]: - ctx.conversation.add(msg) - - # Instantiate agent with the same pattern as CLI - agent_kwargs: Dict[str, Any] = { - "bus": self.bus, - "temperature": temperature, - "max_tokens": max_tokens, - } - if getattr(agent_cls, "accepts_tools", False): - agent_kwargs["tools"] = agent_tools - agent_kwargs["max_turns"] = self.config.agent.max_turns - # Plan 2B I3: forward optimized few-shot examples to agents - # that accept them. Older agents that don't accept the kwarg - # are handled by the existing TypeError fallback below. - examples = getattr(self, "_skill_few_shot_examples", None) - if examples: - agent_kwargs["skill_few_shot_examples"] = examples - if system_prompt is not None: - agent_kwargs["system_prompt"] = system_prompt - if self.capability_policy is not None: - agent_kwargs["capability_policy"] = self.capability_policy - if operator_id is not None: - agent_kwargs["operator_id"] = operator_id - agent_kwargs["session_store"] = self.session_store - agent_kwargs["memory_backend"] = self.memory_backend - - # Inject DigestConfig when instantiating the morning_digest agent - if agent_name == "morning_digest" and hasattr(self.config, "digest"): - dc = self.config.digest - section_sources = {} - for s in dc.sections: - sc = getattr(dc, s, None) - if sc and hasattr(sc, "sources"): - section_sources[s] = sc.sources - agent_kwargs.update( - { - "persona": dc.persona, - "sections": dc.sections, - "section_sources": section_sources, - "timezone": dc.timezone, - "voice_id": dc.voice_id, - "voice_speed": dc.voice_speed, - "tts_backend": dc.tts_backend, - "honorific": dc.honorific, - } - ) - # Ensure digest agent always has its required tools - from openjarvis.tools.digest_collect import DigestCollectTool - from openjarvis.tools.text_to_speech import TextToSpeechTool - - digest_tools = [DigestCollectTool(), TextToSpeechTool()] - existing = agent_kwargs.get("tools", []) - agent_kwargs["tools"] = digest_tools + list(existing) - - try: - ag = agent_cls(self.engine, self.model, **agent_kwargs) - except TypeError: - try: - ag = agent_cls(self.engine, self.model) - except TypeError: - ag = agent_cls() - - # Collect telemetry from all engine calls during agent run - telemetry_events: List[Dict[str, Any]] = [] - - def _on_inference_end(event: Any) -> None: - telemetry_events.append(event.data if hasattr(event, "data") else event) - - self.bus.subscribe(EventType.INFERENCE_END, _on_inference_end) - - # Run — wrap with TraceCollector when tracing is enabled. - # Check trace_store (set at build time) instead of config.traces.enabled - # because the shared config singleton can be mutated by other SystemBuilder - # instances (e.g. the judge backend). - try: - if self.trace_store is not None: - from openjarvis.traces.collector import TraceCollector - - collector = TraceCollector( - ag, - store=self.trace_store, - bus=self.bus, - ) - result = collector.run(query, context=ctx) - self.trace_collector = collector - else: - result = ag.run(query, context=ctx) - finally: - self.bus.unsubscribe(EventType.INFERENCE_END, _on_inference_end) - - # Aggregate telemetry across all engine calls - _telemetry: Dict[str, Any] = {} - if telemetry_events: - total_energy = sum(e.get("energy_joules", 0.0) for e in telemetry_events) - total_latency = sum(e.get("latency", 0.0) for e in telemetry_events) - power_vals = [ - e.get("power_watts", 0.0) - for e in telemetry_events - if e.get("power_watts", 0.0) > 0 - ] - util_vals = [ - e.get("gpu_utilization_pct", 0.0) - for e in telemetry_events - if e.get("gpu_utilization_pct", 0.0) > 0 - ] - throughput_vals = [ - e.get("throughput_tok_per_sec", 0.0) - for e in telemetry_events - if e.get("throughput_tok_per_sec", 0.0) > 0 - ] - _telemetry = { - "ttft": telemetry_events[0].get("ttft", 0.0), - "energy_joules": total_energy, - "power_watts": ( - sum(power_vals) / len(power_vals) if power_vals else 0.0 - ), - "gpu_utilization_pct": ( - sum(util_vals) / len(util_vals) if util_vals else 0.0 - ), - "throughput_tok_per_sec": ( - sum(throughput_vals) / len(throughput_vals) - if throughput_vals - else 0.0 - ), - "gpu_memory_used_gb": max( - (e.get("gpu_memory_used_gb", 0.0) for e in telemetry_events), - default=0.0, - ), - "gpu_temperature_c": max( - (e.get("gpu_temperature_c", 0.0) for e in telemetry_events), - default=0.0, - ), - "inference_calls": len(telemetry_events), - "total_inference_latency": total_latency, - } - - return { - "content": result.content, - "usage": getattr(result, "usage", {}), - "tool_results": [ - { - "tool_name": tr.tool_name, - "content": tr.content, - "success": tr.success, - "arguments": tr.metadata.get("arguments", {}), - } - for tr in getattr(result, "tool_results", []) - ], - "turns": getattr(result, "turns", 1), - "metadata": getattr(result, "metadata", {}), - "model": self.model, - "engine": self.engine_key, - "_telemetry": _telemetry, - } - - def _build_tools(self, tool_names: List[str]) -> List[BaseTool]: - """Build tool instances from tool names.""" - from openjarvis.core.registry import ToolRegistry - - tools: List[BaseTool] = [] - for name in tool_names: - try: - if name == "retrieval" and self.memory_backend: - from openjarvis.tools.retrieval import RetrievalTool - - tools.append(RetrievalTool(self.memory_backend)) - elif name == "llm": - from openjarvis.tools.llm_tool import LLMTool - - tools.append(LLMTool(self.engine, model=self.model)) - elif ToolRegistry.contains(name): - tools.append(ToolRegistry.create(name)) - except Exception as exc: - logger.warning("Failed to build tool %r: %s", name, exc) - return tools - - def wire_channel(self, channel_bridge: Any) -> None: - """Register a message handler on *channel_bridge* that routes every - incoming message through this system (agent or engine) and replies. - - Sessions are isolated per ``":"`` key so - each chat retains its own history. - - Parameters - ---------- - channel_bridge: - A connected :class:`~openjarvis.channels._stubs.BaseChannel` - instance whose ``on_message`` method accepts a callable. - """ - from openjarvis.core.types import Message, Role - from openjarvis.sessions.session import SessionStore - - if self.session_store is None: - from pathlib import Path - - self.session_store = SessionStore( - db_path=Path(self.config.sessions.db_path).expanduser(), - max_age_hours=self.config.sessions.max_age_hours, - consolidation_threshold=self.config.sessions.consolidation_threshold, - ) - - _system = self # capture for closure - - def _on_channel_message(cm) -> None: - session_key = f"{cm.channel}:{cm.conversation_id}" - session = _system.session_store.get_or_create( - session_key, - channel=cm.channel, - channel_user_id=cm.sender, - ) - - prior_msgs: List[Message] = [] - for sm in session.messages: - try: - role = Role(sm.role) - except ValueError: - role = Role.USER - prior_msgs.append(Message(role=role, content=sm.content)) - - reply = "" - try: - if _system.agent_name and _system.agent_name != "none": - result = _system.ask( - cm.content, - context=False, - agent=_system.agent_name, - prior_messages=prior_msgs, - ) - reply = result.get("content", "") - else: - result = _system.ask( - cm.content, - context=False, - prior_messages=prior_msgs, - ) - reply = result.get("content", "") - except Exception: - logger.exception("Channel message handler error") - reply = "Sorry, I encountered an error processing your message." - - try: - _system.session_store.save_message( - session.session_id, - "user", - cm.content, - channel=cm.channel, - ) - _system.session_store.save_message( - session.session_id, - "assistant", - reply, - channel=cm.channel, - ) - except Exception: - logger.debug("Session save error", exc_info=True) - - if reply: - try: - channel_bridge.send( - cm.channel, - reply, - conversation_id=cm.conversation_id, - ) - except Exception: - logger.exception("Channel send error") - - channel_bridge.on_message(_on_channel_message) - - def _close_mcp_clients(self) -> None: - """Close all persistent MCP client connections.""" - for client in self._mcp_clients: - try: - client.close() - except Exception: - logger.debug("Error closing MCP client", exc_info=True) - - def close(self) -> None: - """Release resources.""" - if self.scheduler and hasattr(self.scheduler, "stop"): - self.scheduler.stop() - for resource in ( - self.scheduler_store, - self.engine, - self.gpu_monitor, - self.telemetry_store, - self.trace_store, - self.memory_backend, - self.session_store, - self.channel_backend, - self.workflow_engine, - self.container_runner, - ): - if resource and hasattr(resource, "close"): - resource.close() - if self.agent_manager is not None: - self.agent_manager.close() - if self.agent_scheduler is not None: - self.agent_scheduler.stop() - self._close_mcp_clients() - - def __enter__(self) -> JarvisSystem: - return self - - def __exit__(self, *exc: Any) -> None: - self.close() - - -class SystemBuilder: - """Config-driven fluent builder for JarvisSystem.""" - - def __init__( - self, - config: Optional[JarvisConfig] = None, - *, - config_path: Optional[Any] = None, - ) -> None: - if config is not None: - self._config = config - elif config_path is not None: - from pathlib import Path - - self._config = load_config(Path(config_path)) - else: - self._config = load_config() - - self._engine_key: Optional[str] = None - self._model: Optional[str] = None - self._agent_name: Optional[str] = None - self._tool_names: Optional[List[str]] = None - self._telemetry: Optional[bool] = None - self._traces: Optional[bool] = None - self._bus: Optional[EventBus] = None - self._sandbox: Optional[bool] = None - self._scheduler: Optional[bool] = None - self._workflow: Optional[bool] = None - self._sessions: Optional[bool] = None - self._speech: Optional[bool] = None - self._mcp_clients: List = [] - - def engine(self, key: str) -> SystemBuilder: - self._engine_key = key - return self - - def model(self, name: str) -> SystemBuilder: - self._model = name - return self - - def agent(self, name: str) -> SystemBuilder: - self._agent_name = name - return self - - def tools(self, names: List[str]) -> SystemBuilder: - self._tool_names = names - return self - - def telemetry(self, enabled: bool) -> SystemBuilder: - self._telemetry = enabled - return self - - def traces(self, enabled: bool) -> SystemBuilder: - self._traces = enabled - return self - - def sandbox(self, enabled: bool) -> SystemBuilder: - self._sandbox = enabled - return self - - def scheduler(self, enabled: bool) -> SystemBuilder: - self._scheduler = enabled - return self - - def workflow(self, enabled: bool) -> SystemBuilder: - self._workflow = enabled - return self - - def sessions(self, enabled: bool) -> SystemBuilder: - self._sessions = enabled - return self - - def speech(self, enabled: bool) -> SystemBuilder: - self._speech = enabled - return self - - def event_bus(self, bus: EventBus) -> SystemBuilder: - self._bus = bus - return self - - def build(self) -> JarvisSystem: - """Construct a fully wired JarvisSystem.""" - config = self._config - bus = self._bus or get_event_bus() - - # Resolve engine - engine, engine_key = self._resolve_engine(config) - - # Resolve model - model = self._resolve_model(config, engine) - - # Compute telemetry_enabled and traces_enabled once - telemetry_enabled = ( - self._telemetry if self._telemetry is not None else config.telemetry.enabled - ) - traces_enabled = ( - self._traces if self._traces is not None else config.traces.enabled - ) - # Apply traces flag to config so downstream code respects it - config.traces.enabled = traces_enabled - gpu_monitor = None - energy_monitor = None - if telemetry_enabled and config.telemetry.gpu_metrics: - # Try new multi-vendor EnergyMonitor first - try: - from openjarvis.telemetry.energy_monitor import ( - create_energy_monitor, - ) - - energy_monitor = create_energy_monitor( - poll_interval_ms=config.telemetry.gpu_poll_interval_ms, - prefer_vendor=config.telemetry.energy_vendor or None, - ) - except ImportError: - pass - - # Fall back to legacy GpuMonitor - if energy_monitor is None: - try: - from openjarvis.telemetry.gpu_monitor import GpuMonitor - - if GpuMonitor.available(): - gpu_monitor = GpuMonitor( - poll_interval_ms=config.telemetry.gpu_poll_interval_ms, - ) - except ImportError: - pass - - # Apply security guardrails FIRST (innermost wrapper) - from openjarvis.security import setup_security - - sec = setup_security(config, engine, bus) - engine = sec.engine - - # Then wrap with InstrumentedEngine (outermost wrapper) - if telemetry_enabled: - from openjarvis.telemetry.instrumented_engine import ( - InstrumentedEngine, - ) - - engine = InstrumentedEngine( - engine, - bus, - gpu_monitor=gpu_monitor, - energy_monitor=energy_monitor, - ) - - # Set up telemetry store - telemetry_store = None - if telemetry_enabled: - telemetry_store = self._setup_telemetry(config, bus) - - # Resolve memory backend - memory_backend = self._resolve_memory(config) - - # Resolve channel backend - channel_backend = self._resolve_channel(config, bus) - - # Resolve tools - tool_list = self._resolve_tools( - config, - engine, - model, - memory_backend, - channel_backend, - ) - - # Build tool executor - tool_executor = ToolExecutor(tool_list, bus) if tool_list else None - - # Resolve skills - skill_manager = None - skill_few_shot_examples: List[str] = [] - if config.skills.enabled: - try: - from pathlib import Path - - from openjarvis.skills.manager import SkillManager - - skill_manager = SkillManager( - bus, capability_policy=sec.capability_policy - ) - skill_paths = [Path(config.skills.skills_dir).expanduser()] - workspace_skills = Path("./skills") - if workspace_skills.exists(): - skill_paths.insert(0, workspace_skills) - skill_manager.discover(paths=skill_paths) - if tool_executor: - skill_manager.set_tool_executor(tool_executor) - skill_tools = skill_manager.get_skill_tools( - tool_executor=tool_executor, - ) - tool_list.extend(skill_tools) - if tool_list: - tool_executor = ToolExecutor(tool_list, bus) - # Plan 2B I3: capture optimized few-shot examples so - # _run_agent can forward them to tool-using agents. - skill_few_shot_examples = skill_manager.get_few_shot_examples() - except Exception as exc: - logger.warning("Failed to initialize skills: %s", exc) - - # Resolve agent name - agent_name = self._agent_name or config.agent.default_agent - - # Set up container sandbox runner - container_runner = self._setup_sandbox(config) - - # Set up scheduler - scheduler_store, task_scheduler = self._setup_scheduler(config, bus) - - # Set up workflow engine - workflow_engine = self._setup_workflow(config, bus) - - # Set up session store - session_store = self._setup_sessions(config) - - # Set up trace store - trace_store = None - if traces_enabled: - try: - from openjarvis.traces.store import TraceStore - - trace_store = TraceStore(config.traces.db_path) - except Exception: - logger.warning("Failed to initialize TraceStore", exc_info=True) - - # Set up capability policy - capability_policy = sec.capability_policy - - # Set up learning orchestrator (when training is enabled) - learning_orchestrator = self._setup_learning_orchestrator(config) - - # Agent Manager - agent_manager = None - if config.agent_manager.enabled: - try: - from pathlib import Path - - from openjarvis.agents.manager import AgentManager - - am_db = config.agent_manager.db_path or str( - Path("~/.openjarvis/agents.db").expanduser() - ) - agent_manager = AgentManager(db_path=am_db) - except Exception as exc: - logger.warning("Failed to initialize agent manager: %s", exc) - - # Executor + Scheduler (depend on agent_manager) - agent_executor = None - agent_scheduler = None - if agent_manager is not None: - try: - from openjarvis.agents.executor import AgentExecutor - from openjarvis.agents.scheduler import AgentScheduler - - # Wire TraceStore into executor when tracing is enabled - _trace_store = None - if config.traces.enabled: - try: - from openjarvis.traces.store import TraceStore - - _trace_store = TraceStore(config.traces.db_path) - except Exception: - logger.warning( - "Failed to initialize TraceStore", - exc_info=True, - ) - - agent_executor = AgentExecutor( - manager=agent_manager, - event_bus=bus, - trace_store=_trace_store, - ) - agent_scheduler = AgentScheduler( - manager=agent_manager, - executor=agent_executor, - ) - except Exception: - logger.warning("Failed to initialize agent scheduler", exc_info=True) - - # Set up speech backend - speech_backend = None - speech_enabled = self._speech if self._speech is not None else True - if speech_enabled: - try: - from openjarvis.speech._discovery import get_speech_backend - - speech_backend = get_speech_backend(config) - except Exception as exc: - logger.warning("Failed to initialize speech backend: %s", exc) - - system = JarvisSystem( - config=config, - bus=bus, - engine=engine, - engine_key=engine_key, - model=model, - agent_name=agent_name, - tools=tool_list, - tool_executor=tool_executor, - memory_backend=memory_backend, - channel_backend=channel_backend, - telemetry_store=telemetry_store, - trace_store=trace_store, - gpu_monitor=gpu_monitor, - scheduler_store=scheduler_store, - scheduler=task_scheduler, - container_runner=container_runner, - workflow_engine=workflow_engine, - session_store=session_store, - capability_policy=capability_policy, - audit_logger=sec.audit_logger, - agent_manager=agent_manager, - agent_scheduler=agent_scheduler, - agent_executor=agent_executor, - speech_backend=speech_backend, - skill_manager=skill_manager, - ) - system._learning_orchestrator = learning_orchestrator - # Plan 2B I3: stash few-shot examples on the system so _run_agent - # can include them in agent_kwargs for tool-using agents. - system._skill_few_shot_examples = skill_few_shot_examples - # Transfer MCP clients so JarvisSystem.close() can shut them down - system._mcp_clients = list(getattr(self, "_mcp_clients", [])) - # Wire system reference — must happen before scheduler.start() - if system.agent_executor is not None: - system.agent_executor.set_system(system) - return system - - def _resolve_engine(self, config: JarvisConfig): - """Resolve the inference engine.""" - from openjarvis.engine._discovery import get_engine - - pref = config.intelligence.preferred_engine - key = self._engine_key or pref or config.engine.default - resolved = get_engine(config, key) - if resolved is None: - raise RuntimeError( - "No inference engine available. " - "Make sure an engine is running (e.g. ollama serve)." - ) - return resolved[1], resolved[0] - - def _resolve_model(self, config: JarvisConfig, engine: InferenceEngine) -> str: - """Resolve which model to use.""" - if self._model: - return self._model - if config.intelligence.default_model: - return config.intelligence.default_model - - # Try to discover from engine - try: - models = engine.list_models() - if models: - return models[0] - except Exception as exc: - logger.warning("Failed to list models from engine: %s", exc) - - return config.intelligence.fallback_model or "" - - def _setup_telemetry(self, config, bus): - """Set up telemetry store.""" - try: - from openjarvis.telemetry.store import TelemetryStore - - store = TelemetryStore(db_path=config.telemetry.db_path) - store.subscribe_to_bus(bus) - return store - except Exception as exc: - logger.warning("Failed to set up telemetry store: %s", exc) - return None - - def _resolve_memory(self, config): - """Resolve memory backend.""" - try: - import openjarvis.tools.storage # noqa: F401 -- trigger registration - from openjarvis.core.registry import MemoryRegistry - - key = config.memory.default_backend - if MemoryRegistry.contains(key): - return MemoryRegistry.create(key, db_path=config.memory.db_path) - except Exception as exc: - logger.warning("Failed to resolve memory backend: %s", exc) - return None - - def _resolve_channel(self, config, bus): - """Resolve channel backend from config.""" - if not config.channel.enabled: - return None - try: - import openjarvis.channels # noqa: F401 -- trigger registration - from openjarvis.core.registry import ChannelRegistry - - key = config.channel.default_channel - if not key: - return None - if not ChannelRegistry.contains(key): - return None - - kwargs: Dict[str, Any] = {"bus": bus} - if key == "telegram": - tc = config.channel.telegram - if tc.bot_token: - kwargs["bot_token"] = tc.bot_token - if tc.parse_mode: - kwargs["parse_mode"] = tc.parse_mode - elif key == "discord": - dc = config.channel.discord - if dc.bot_token: - kwargs["bot_token"] = dc.bot_token - elif key == "slack": - sc = config.channel.slack - if sc.bot_token: - kwargs["bot_token"] = sc.bot_token - if sc.app_token: - kwargs["app_token"] = sc.app_token - elif key == "webhook": - wc = config.channel.webhook - if wc.url: - kwargs["url"] = wc.url - if wc.secret: - kwargs["secret"] = wc.secret - if wc.method: - kwargs["method"] = wc.method - elif key == "email": - ec = config.channel.email - if ec.smtp_host: - kwargs["smtp_host"] = ec.smtp_host - kwargs["smtp_port"] = ec.smtp_port - if ec.imap_host: - kwargs["imap_host"] = ec.imap_host - kwargs["imap_port"] = ec.imap_port - if ec.username: - kwargs["username"] = ec.username - if ec.password: - kwargs["password"] = ec.password - kwargs["use_tls"] = ec.use_tls - elif key == "whatsapp": - wac = config.channel.whatsapp - if wac.access_token: - kwargs["access_token"] = wac.access_token - if wac.phone_number_id: - kwargs["phone_number_id"] = wac.phone_number_id - elif key == "signal": - sgc = config.channel.signal - if sgc.api_url: - kwargs["api_url"] = sgc.api_url - if sgc.phone_number: - kwargs["phone_number"] = sgc.phone_number - elif key == "google_chat": - gcc = config.channel.google_chat - if gcc.webhook_url: - kwargs["webhook_url"] = gcc.webhook_url - elif key == "irc": - ic = config.channel.irc - if ic.server: - kwargs["server"] = ic.server - kwargs["port"] = ic.port - if ic.nick: - kwargs["nick"] = ic.nick - if ic.password: - kwargs["password"] = ic.password - kwargs["use_tls"] = ic.use_tls - elif key == "webchat": - pass # no config needed - elif key == "teams": - tmc = config.channel.teams - if tmc.app_id: - kwargs["app_id"] = tmc.app_id - if tmc.app_password: - kwargs["app_password"] = tmc.app_password - if tmc.service_url: - kwargs["service_url"] = tmc.service_url - elif key == "matrix": - mc = config.channel.matrix - if mc.homeserver: - kwargs["homeserver"] = mc.homeserver - if mc.access_token: - kwargs["access_token"] = mc.access_token - elif key == "mattermost": - mmc = config.channel.mattermost - if mmc.url: - kwargs["url"] = mmc.url - if mmc.token: - kwargs["token"] = mmc.token - elif key == "feishu": - fc = config.channel.feishu - if fc.app_id: - kwargs["app_id"] = fc.app_id - if fc.app_secret: - kwargs["app_secret"] = fc.app_secret - elif key == "bluebubbles": - bbc = config.channel.bluebubbles - if bbc.url: - kwargs["url"] = bbc.url - if bbc.password: - kwargs["password"] = bbc.password - elif key == "whatsapp_baileys": - wbc = config.channel.whatsapp_baileys - if wbc.auth_dir: - kwargs["auth_dir"] = wbc.auth_dir - if wbc.assistant_name: - kwargs["assistant_name"] = wbc.assistant_name - kwargs["assistant_has_own_number"] = wbc.assistant_has_own_number - elif key == "sendblue": - sbc = getattr(config.channel, "sendblue", None) - if sbc: - if getattr(sbc, "api_key_id", ""): - kwargs["api_key_id"] = sbc.api_key_id - if getattr(sbc, "api_secret_key", ""): - kwargs["api_secret_key"] = sbc.api_secret_key - if getattr(sbc, "from_number", ""): - kwargs["from_number"] = sbc.from_number - - return ChannelRegistry.create(key, **kwargs) - except Exception as exc: - logger.warning("Failed to resolve channel backend %r: %s", key, exc) - return None - - def _resolve_tools( - self, config, engine, model, memory_backend, channel_backend=None - ): - """Resolve tool instances via MCPServer (primary) + external MCP servers.""" - from openjarvis.mcp.server import MCPServer - - # 1. Build internal MCPServer with all auto-discovered tools - internal_server = MCPServer() - - # 2. Inject runtime dependencies into tools that need them - for tool in internal_server.get_tools(): - self._inject_tool_deps(tool, engine, model, memory_backend, channel_backend) - - # 3. Determine which tool names to include - tool_names = self._tool_names - if tool_names is None: - raw = config.tools.enabled or config.agent.tools - if raw: - if isinstance(raw, list): - tool_names = [ - n.strip() for n in raw if isinstance(n, str) and n.strip() - ] - else: - tool_names = [n.strip() for n in raw.split(",") if n.strip()] - else: - tool_names = [] - - # 4. Filter to requested tool names (if specified) - if tool_names: - all_tools = {t.spec.name: t for t in internal_server.get_tools()} - tools = [all_tools[n] for n in tool_names if n in all_tools] - else: - tools = [] - - # 5. Discover external MCP server tools - if config.tools.mcp.servers: - try: - import json - - server_list = json.loads(config.tools.mcp.servers) - if isinstance(server_list, list): - for server_cfg in server_list: - try: - external_tools = self._discover_external_mcp(server_cfg) - if tool_names: - external_tools = [ - t - for t in external_tools - if t.spec.name in tool_names - ] - tools.extend(external_tools) - except Exception as exc: - logger.warning( - "Failed to discover external MCP tools: %s", - exc, - ) - except (json.JSONDecodeError, TypeError) as exc: - logger.warning("Failed to parse MCP server config: %s", exc) - - return tools - - @staticmethod - def _inject_tool_deps(tool, engine, model, memory_backend, channel_backend): - """Inject runtime dependencies into tools that need them.""" - name = tool.spec.name - if name == "llm": - if hasattr(tool, "_engine"): - tool._engine = engine - if hasattr(tool, "_model"): - tool._model = model - elif name == "retrieval": - if hasattr(tool, "_backend"): - tool._backend = memory_backend - elif name.startswith("memory_"): - if hasattr(tool, "_backend"): - tool._backend = memory_backend - elif name.startswith("channel_"): - if hasattr(tool, "_channel"): - tool._channel = channel_backend - elif name in ( - "schedule_task", - "list_scheduled_tasks", - "pause_scheduled_task", - "resume_scheduled_task", - "cancel_scheduled_task", - ): - pass # scheduler injection handled post-build - - def _setup_sandbox(self, config): - """Set up container sandbox runner if enabled.""" - sandbox_enabled = ( - self._sandbox if self._sandbox is not None else config.sandbox.enabled - ) - if not sandbox_enabled: - return None - try: - from openjarvis.sandbox.runner import ContainerRunner - - return ContainerRunner( - image=config.sandbox.image, - timeout=config.sandbox.timeout, - mount_allowlist_path=config.sandbox.mount_allowlist_path, - max_concurrent=config.sandbox.max_concurrent, - runtime=config.sandbox.runtime, - ) - except Exception as exc: - logger.warning("Failed to set up container sandbox: %s", exc) - return None - - def _setup_scheduler(self, config, bus): - """Set up task scheduler if enabled.""" - scheduler_enabled = ( - self._scheduler if self._scheduler is not None else config.scheduler.enabled - ) - if not scheduler_enabled: - return None, None - try: - from openjarvis.scheduler.store import SchedulerStore - - db_path = config.scheduler.db_path or str( - config.hardware.platform # unused, just for fallback - ) - if not config.scheduler.db_path: - from openjarvis.core.config import DEFAULT_CONFIG_DIR - - db_path = str(DEFAULT_CONFIG_DIR / "scheduler.db") - - store = SchedulerStore(db_path=db_path) - - from openjarvis.scheduler.scheduler import TaskScheduler - - sched = TaskScheduler( - store, - poll_interval=config.scheduler.poll_interval, - bus=bus, - ) - return store, sched - except Exception as exc: - logger.warning("Failed to set up task scheduler: %s", exc) - return None, None - - def _setup_workflow(self, config, bus): - """Set up workflow engine if enabled.""" - workflow_enabled = ( - self._workflow if self._workflow is not None else config.workflow.enabled - ) - if not workflow_enabled: - return None - try: - from openjarvis.workflow.engine import WorkflowEngine - - return WorkflowEngine( - bus=bus, - max_parallel=config.workflow.max_parallel, - default_node_timeout=config.workflow.default_node_timeout, - ) - except Exception as exc: - logger.warning("Failed to set up workflow engine: %s", exc) - return None - - def _setup_sessions(self, config): - """Set up session store if enabled.""" - sessions_enabled = ( - self._sessions if self._sessions is not None else config.sessions.enabled - ) - if not sessions_enabled: - return None - try: - from openjarvis.sessions.session import SessionStore - - return SessionStore( - db_path=config.sessions.db_path, - max_age_hours=config.sessions.max_age_hours, - consolidation_threshold=config.sessions.consolidation_threshold, - ) - except Exception as exc: - logger.warning("Failed to set up session store: %s", exc) - return None - - @staticmethod - def _setup_learning_orchestrator(config: JarvisConfig): - """Set up LearningOrchestrator when training is enabled.""" - if not config.learning.training_enabled: - return None - try: - from openjarvis.core.config import DEFAULT_CONFIG_DIR - from openjarvis.learning.learning_orchestrator import ( - LearningOrchestrator, - ) - from openjarvis.learning.training.lora import LoRATrainingConfig - from openjarvis.traces.store import TraceStore - - trace_store = TraceStore(db_path=config.traces.db_path) - config_dir = DEFAULT_CONFIG_DIR / "agent_configs" - - sft_cfg = config.learning.intelligence.sft - lora_config = LoRATrainingConfig( - lora_rank=sft_cfg.lora_rank, - lora_alpha=sft_cfg.lora_alpha, - ) - - return LearningOrchestrator( - trace_store=trace_store, - config_dir=config_dir, - min_improvement=config.learning.min_improvement, - min_sft_pairs=sft_cfg.min_pairs, - lora_config=lora_config, - ) - except Exception as exc: - logger.warning("Failed to set up learning orchestrator: %s", exc) - return None - - def _discover_external_mcp(self, server_cfg) -> List[BaseTool]: - """Discover tools from an external MCP server configuration. - - Supports both stdio (command + args) and Streamable HTTP (url) - transports. Persists MCP clients on ``self._mcp_clients`` so - that transports stay alive for runtime tool calls. - """ - import json - - from openjarvis.mcp.client import MCPClient - from openjarvis.mcp.transport import StdioTransport, StreamableHTTPTransport - from openjarvis.tools.mcp_adapter import MCPToolProvider - - cfg = json.loads(server_cfg) if isinstance(server_cfg, str) else server_cfg - name = cfg.get("name", "") - url = cfg.get("url") - command = cfg.get("command", "") - args = cfg.get("args", []) - - # Build transport based on config keys - if url: - transport = StreamableHTTPTransport(url=url) - elif command: - transport = StdioTransport(command=[command] + args) - else: - logger.warning( - "MCP server '%s' has neither 'url' nor 'command' — skipping", - name, - ) - return [] - - client = MCPClient(transport) - client.initialize() - - # Persist client so the transport stays alive for tool calls - self._mcp_clients.append(client) - - provider = MCPToolProvider(client) - discovered = provider.discover() - - # Per-server tool filtering - include_tools = set(cfg.get("include_tools", [])) - exclude_tools = set(cfg.get("exclude_tools", [])) - if include_tools: - discovered = [t for t in discovered if t.spec.name in include_tools] - if exclude_tools: - discovered = [t for t in discovered if t.spec.name not in exclude_tools] - - logger.info( - "Discovered %d tools from MCP server '%s'", - len(discovered), - name, - ) - return discovered - - -__all__ = ["JarvisSystem", "SystemBuilder"] diff --git a/src/openjarvis/system/__init__.py b/src/openjarvis/system/__init__.py new file mode 100644 index 00000000..d79913e3 --- /dev/null +++ b/src/openjarvis/system/__init__.py @@ -0,0 +1,23 @@ +"""Top-level system composition: JarvisSystem, SystemBuilder, and helpers.""" + +from openjarvis.system.builder import SystemBuilder +from openjarvis.system.bundles import ( + AgentRuntime, + Observability, + Scheduling, + SecurityContext, +) +from openjarvis.system.core import JarvisSystem +from openjarvis.system.orchestrator import QueryOrchestrator +from openjarvis.system.protocols import OrchestratorDeps + +__all__ = [ + "AgentRuntime", + "JarvisSystem", + "Observability", + "OrchestratorDeps", + "QueryOrchestrator", + "Scheduling", + "SecurityContext", + "SystemBuilder", +] diff --git a/src/openjarvis/system/_channel_kwargs.py b/src/openjarvis/system/_channel_kwargs.py new file mode 100644 index 00000000..bcd92f24 --- /dev/null +++ b/src/openjarvis/system/_channel_kwargs.py @@ -0,0 +1,199 @@ +"""Per-channel config→kwargs mappers for ChannelRegistry.create().""" + +from __future__ import annotations + +from typing import Any, Callable, Dict, Optional + + +def _telegram(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.bot_token: + kw["bot_token"] = c.bot_token + if c.parse_mode: + kw["parse_mode"] = c.parse_mode + return kw + + +def _discord(c: Any) -> Dict[str, Any]: + return {"bot_token": c.bot_token} if c.bot_token else {} + + +def _slack(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.bot_token: + kw["bot_token"] = c.bot_token + if c.app_token: + kw["app_token"] = c.app_token + return kw + + +def _webhook(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.url: + kw["url"] = c.url + if c.secret: + kw["secret"] = c.secret + if c.method: + kw["method"] = c.method + return kw + + +def _email(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = { + "smtp_port": c.smtp_port, + "imap_port": c.imap_port, + "use_tls": c.use_tls, + } + if c.smtp_host: + kw["smtp_host"] = c.smtp_host + if c.imap_host: + kw["imap_host"] = c.imap_host + if c.username: + kw["username"] = c.username + if c.password: + kw["password"] = c.password + return kw + + +def _whatsapp(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.access_token: + kw["access_token"] = c.access_token + if c.phone_number_id: + kw["phone_number_id"] = c.phone_number_id + return kw + + +def _signal(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.api_url: + kw["api_url"] = c.api_url + if c.phone_number: + kw["phone_number"] = c.phone_number + return kw + + +def _google_chat(c: Any) -> Dict[str, Any]: + return {"webhook_url": c.webhook_url} if c.webhook_url else {} + + +def _irc(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {"port": c.port, "use_tls": c.use_tls} + if c.server: + kw["server"] = c.server + if c.nick: + kw["nick"] = c.nick + if c.password: + kw["password"] = c.password + return kw + + +def _webchat(c: Any) -> Dict[str, Any]: + return {} + + +def _teams(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.app_id: + kw["app_id"] = c.app_id + if c.app_password: + kw["app_password"] = c.app_password + if c.service_url: + kw["service_url"] = c.service_url + return kw + + +def _matrix(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.homeserver: + kw["homeserver"] = c.homeserver + if c.access_token: + kw["access_token"] = c.access_token + return kw + + +def _mattermost(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.url: + kw["url"] = c.url + if c.token: + kw["token"] = c.token + return kw + + +def _feishu(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.app_id: + kw["app_id"] = c.app_id + if c.app_secret: + kw["app_secret"] = c.app_secret + return kw + + +def _bluebubbles(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {} + if c.url: + kw["url"] = c.url + if c.password: + kw["password"] = c.password + return kw + + +def _whatsapp_baileys(c: Any) -> Dict[str, Any]: + kw: Dict[str, Any] = {"assistant_has_own_number": c.assistant_has_own_number} + if c.auth_dir: + kw["auth_dir"] = c.auth_dir + if c.assistant_name: + kw["assistant_name"] = c.assistant_name + return kw + + +def _sendblue(c: Any) -> Dict[str, Any]: + if c is None: + return {} + kw: Dict[str, Any] = {} + if getattr(c, "api_key_id", ""): + kw["api_key_id"] = c.api_key_id + if getattr(c, "api_secret_key", ""): + kw["api_secret_key"] = c.api_secret_key + if getattr(c, "from_number", ""): + kw["from_number"] = c.from_number + return kw + + +# Maps channel key → (config.channel., mapper). +# Omit a channel here to have it fall through with only {"bus": bus}. +_CHANNEL_MAPPERS: Dict[str, tuple] = { + "telegram": ("telegram", _telegram), + "discord": ("discord", _discord), + "slack": ("slack", _slack), + "webhook": ("webhook", _webhook), + "email": ("email", _email), + "whatsapp": ("whatsapp", _whatsapp), + "signal": ("signal", _signal), + "google_chat": ("google_chat", _google_chat), + "irc": ("irc", _irc), + "webchat": ("webchat", _webchat), + "teams": ("teams", _teams), + "matrix": ("matrix", _matrix), + "mattermost": ("mattermost", _mattermost), + "feishu": ("feishu", _feishu), + "bluebubbles": ("bluebubbles", _bluebubbles), + "whatsapp_baileys": ("whatsapp_baileys", _whatsapp_baileys), + "sendblue": ("sendblue", _sendblue), +} + + +def build_channel_kwargs(channel_config: Any, key: str) -> Dict[str, Any]: + """Return channel-specific kwargs for ChannelRegistry.create(key, ...). + + Does not include ``bus`` — the caller adds that. + """ + entry: Optional[tuple] = _CHANNEL_MAPPERS.get(key) + if entry is None: + return {} + attr, mapper = entry + sub_cfg: Callable[[Any], Dict[str, Any]] = getattr(channel_config, attr, None) + if sub_cfg is None: + return {} + return mapper(sub_cfg) diff --git a/src/openjarvis/system/builder.py b/src/openjarvis/system/builder.py new file mode 100644 index 00000000..1197ade7 --- /dev/null +++ b/src/openjarvis/system/builder.py @@ -0,0 +1,621 @@ +"""Config-driven fluent builder that wires up a JarvisSystem.""" + +from __future__ import annotations + +import logging +from typing import Any, List, Optional + +from openjarvis.core.config import JarvisConfig, load_config +from openjarvis.core.events import EventBus, get_event_bus +from openjarvis.engine._stubs import InferenceEngine +from openjarvis.system.core import JarvisSystem +from openjarvis.tools._stubs import BaseTool, ToolExecutor + +logger = logging.getLogger(__name__) + + +class SystemBuilder: + """Config-driven fluent builder for JarvisSystem.""" + + def __init__( + self, + config: Optional[JarvisConfig] = None, + *, + config_path: Optional[Any] = None, + ) -> None: + if config is not None: + self._config = config + elif config_path is not None: + from pathlib import Path + + self._config = load_config(Path(config_path)) + else: + self._config = load_config() + + self._engine_key: Optional[str] = None + self._model: Optional[str] = None + self._agent_name: Optional[str] = None + self._tool_names: Optional[List[str]] = None + self._telemetry: Optional[bool] = None + self._traces: Optional[bool] = None + self._bus: Optional[EventBus] = None + self._sandbox: Optional[bool] = None + self._scheduler: Optional[bool] = None + self._workflow: Optional[bool] = None + self._sessions: Optional[bool] = None + self._speech: Optional[bool] = None + self._mcp_clients: List = [] + + def engine(self, key: str) -> SystemBuilder: + self._engine_key = key + return self + + def model(self, name: str) -> SystemBuilder: + self._model = name + return self + + def agent(self, name: str) -> SystemBuilder: + self._agent_name = name + return self + + def tools(self, names: List[str]) -> SystemBuilder: + self._tool_names = names + return self + + def telemetry(self, enabled: bool) -> SystemBuilder: + self._telemetry = enabled + return self + + def traces(self, enabled: bool) -> SystemBuilder: + self._traces = enabled + return self + + def sandbox(self, enabled: bool) -> SystemBuilder: + self._sandbox = enabled + return self + + def scheduler(self, enabled: bool) -> SystemBuilder: + self._scheduler = enabled + return self + + def workflow(self, enabled: bool) -> SystemBuilder: + self._workflow = enabled + return self + + def sessions(self, enabled: bool) -> SystemBuilder: + self._sessions = enabled + return self + + def speech(self, enabled: bool) -> SystemBuilder: + self._speech = enabled + return self + + def event_bus(self, bus: EventBus) -> SystemBuilder: + self._bus = bus + return self + + def build(self) -> JarvisSystem: + """Construct a fully wired JarvisSystem.""" + config = self._config + bus = self._bus or get_event_bus() + + engine, engine_key = self._resolve_engine(config) + model = self._resolve_model(config, engine) + + telemetry_enabled = ( + self._telemetry if self._telemetry is not None else config.telemetry.enabled + ) + traces_enabled = ( + self._traces if self._traces is not None else config.traces.enabled + ) + config.traces.enabled = traces_enabled + gpu_monitor = None + energy_monitor = None + if telemetry_enabled and config.telemetry.gpu_metrics: + try: + from openjarvis.telemetry.energy_monitor import ( + create_energy_monitor, + ) + + energy_monitor = create_energy_monitor( + poll_interval_ms=config.telemetry.gpu_poll_interval_ms, + prefer_vendor=config.telemetry.energy_vendor or None, + ) + except ImportError: + pass + + if energy_monitor is None: + try: + from openjarvis.telemetry.gpu_monitor import GpuMonitor + + if GpuMonitor.available(): + gpu_monitor = GpuMonitor( + poll_interval_ms=config.telemetry.gpu_poll_interval_ms, + ) + except ImportError: + pass + + from openjarvis.security import setup_security + + sec = setup_security(config, engine, bus) + engine = sec.engine + + if telemetry_enabled: + from openjarvis.telemetry.instrumented_engine import ( + InstrumentedEngine, + ) + + engine = InstrumentedEngine( + engine, + bus, + gpu_monitor=gpu_monitor, + energy_monitor=energy_monitor, + ) + + telemetry_store = None + if telemetry_enabled: + telemetry_store = self._setup_telemetry(config, bus) + + memory_backend = self._resolve_memory(config) + channel_backend = self._resolve_channel(config, bus) + tool_list = self._resolve_tools( + config, + engine, + model, + memory_backend, + channel_backend, + ) + tool_executor = ToolExecutor(tool_list, bus) if tool_list else None + + skill_manager = None + skill_few_shot_examples: List[str] = [] + if config.skills.enabled: + try: + from pathlib import Path + + from openjarvis.skills.manager import SkillManager + + skill_manager = SkillManager( + bus, capability_policy=sec.capability_policy + ) + skill_paths = [Path(config.skills.skills_dir).expanduser()] + workspace_skills = Path("./skills") + if workspace_skills.exists(): + skill_paths.insert(0, workspace_skills) + skill_manager.discover(paths=skill_paths) + if tool_executor: + skill_manager.set_tool_executor(tool_executor) + skill_tools = skill_manager.get_skill_tools( + tool_executor=tool_executor, + ) + tool_list.extend(skill_tools) + if tool_list: + tool_executor = ToolExecutor(tool_list, bus) + skill_few_shot_examples = skill_manager.get_few_shot_examples() + except Exception as exc: + logger.warning("Failed to initialize skills: %s", exc) + + agent_name = self._agent_name or config.agent.default_agent + container_runner = self._setup_sandbox(config) + scheduler_store, task_scheduler = self._setup_scheduler(config, bus) + workflow_engine = self._setup_workflow(config, bus) + session_store = self._setup_sessions(config) + + trace_store = None + if traces_enabled: + try: + from openjarvis.traces.store import TraceStore + + trace_store = TraceStore(config.traces.db_path) + except Exception: + logger.warning("Failed to initialize TraceStore", exc_info=True) + + capability_policy = sec.capability_policy + learning_orchestrator = self._setup_learning_orchestrator(config) + + agent_manager = None + if config.agent_manager.enabled: + try: + from pathlib import Path + + from openjarvis.agents.manager import AgentManager + + am_db = config.agent_manager.db_path or str( + Path("~/.openjarvis/agents.db").expanduser() + ) + agent_manager = AgentManager(db_path=am_db) + except Exception as exc: + logger.warning("Failed to initialize agent manager: %s", exc) + + agent_executor = None + agent_scheduler = None + if agent_manager is not None: + try: + from openjarvis.agents.executor import AgentExecutor + from openjarvis.agents.scheduler import AgentScheduler + + _trace_store = None + if config.traces.enabled: + try: + from openjarvis.traces.store import TraceStore + + _trace_store = TraceStore(config.traces.db_path) + except Exception: + logger.warning( + "Failed to initialize TraceStore", + exc_info=True, + ) + + agent_executor = AgentExecutor( + manager=agent_manager, + event_bus=bus, + trace_store=_trace_store, + ) + agent_scheduler = AgentScheduler( + manager=agent_manager, + executor=agent_executor, + ) + except Exception: + logger.warning("Failed to initialize agent scheduler", exc_info=True) + + speech_backend = None + speech_enabled = self._speech if self._speech is not None else True + if speech_enabled: + try: + from openjarvis.speech._discovery import get_speech_backend + + speech_backend = get_speech_backend(config) + except Exception as exc: + logger.warning("Failed to initialize speech backend: %s", exc) + + system = JarvisSystem( + config=config, + bus=bus, + engine=engine, + engine_key=engine_key, + model=model, + agent_name=agent_name, + tools=tool_list, + tool_executor=tool_executor, + memory_backend=memory_backend, + channel_backend=channel_backend, + telemetry_store=telemetry_store, + trace_store=trace_store, + gpu_monitor=gpu_monitor, + scheduler_store=scheduler_store, + scheduler=task_scheduler, + container_runner=container_runner, + workflow_engine=workflow_engine, + session_store=session_store, + capability_policy=capability_policy, + audit_logger=sec.audit_logger, + agent_manager=agent_manager, + agent_scheduler=agent_scheduler, + agent_executor=agent_executor, + speech_backend=speech_backend, + skill_manager=skill_manager, + ) + system._learning_orchestrator = learning_orchestrator + system._skill_few_shot_examples = skill_few_shot_examples + system._mcp_clients = list(getattr(self, "_mcp_clients", [])) + if system.agent_executor is not None: + system.agent_executor.set_system(system) + return system + + def _resolve_engine(self, config: JarvisConfig): + from openjarvis.engine._discovery import get_engine + + pref = config.intelligence.preferred_engine + key = self._engine_key or pref or config.engine.default + resolved = get_engine(config, key) + if resolved is None: + raise RuntimeError( + "No inference engine available. " + "Make sure an engine is running (e.g. ollama serve)." + ) + return resolved[1], resolved[0] + + def _resolve_model(self, config: JarvisConfig, engine: InferenceEngine) -> str: + if self._model: + return self._model + if config.intelligence.default_model: + return config.intelligence.default_model + try: + models = engine.list_models() + if models: + return models[0] + except Exception as exc: + logger.warning("Failed to list models from engine: %s", exc) + return config.intelligence.fallback_model or "" + + def _setup_telemetry(self, config, bus): + try: + from openjarvis.telemetry.store import TelemetryStore + + store = TelemetryStore(db_path=config.telemetry.db_path) + store.subscribe_to_bus(bus) + return store + except Exception as exc: + logger.warning("Failed to set up telemetry store: %s", exc) + return None + + def _resolve_memory(self, config): + try: + import openjarvis.tools.storage # noqa: F401 -- trigger registration + from openjarvis.core.registry import MemoryRegistry + + key = config.memory.default_backend + if MemoryRegistry.contains(key): + return MemoryRegistry.create(key, db_path=config.memory.db_path) + except Exception as exc: + logger.warning("Failed to resolve memory backend: %s", exc) + return None + + def _resolve_channel(self, config, bus): + if not config.channel.enabled: + return None + key = config.channel.default_channel + try: + import openjarvis.channels # noqa: F401 -- trigger registration + from openjarvis.core.registry import ChannelRegistry + from openjarvis.system._channel_kwargs import build_channel_kwargs + + if not key or not ChannelRegistry.contains(key): + return None + kwargs = {"bus": bus, **build_channel_kwargs(config.channel, key)} + return ChannelRegistry.create(key, **kwargs) + except Exception as exc: + logger.warning("Failed to resolve channel backend %r: %s", key, exc) + return None + + def _resolve_tools( + self, config, engine, model, memory_backend, channel_backend=None + ): + """Resolve tool instances via MCPServer (primary) + external MCP servers.""" + from openjarvis.mcp.server import MCPServer + + internal_server = MCPServer() + for tool in internal_server.get_tools(): + self._inject_tool_deps(tool, engine, model, memory_backend, channel_backend) + + tool_names = self._tool_names + if tool_names is None: + raw = config.tools.enabled or config.agent.tools + if raw: + if isinstance(raw, list): + tool_names = [ + n.strip() for n in raw if isinstance(n, str) and n.strip() + ] + else: + tool_names = [n.strip() for n in raw.split(",") if n.strip()] + else: + tool_names = [] + + if tool_names: + all_tools = {t.spec.name: t for t in internal_server.get_tools()} + tools = [all_tools[n] for n in tool_names if n in all_tools] + else: + tools = [] + + if config.tools.mcp.servers: + try: + import json + + server_list = json.loads(config.tools.mcp.servers) + if isinstance(server_list, list): + for server_cfg in server_list: + try: + external_tools = self._discover_external_mcp(server_cfg) + if tool_names: + external_tools = [ + t + for t in external_tools + if t.spec.name in tool_names + ] + tools.extend(external_tools) + except Exception as exc: + logger.warning( + "Failed to discover external MCP tools: %s", + exc, + ) + except (json.JSONDecodeError, TypeError) as exc: + logger.warning("Failed to parse MCP server config: %s", exc) + + return tools + + @staticmethod + def _inject_tool_deps(tool, engine, model, memory_backend, channel_backend): + name = tool.spec.name + if name == "llm": + if hasattr(tool, "_engine"): + tool._engine = engine + if hasattr(tool, "_model"): + tool._model = model + elif name == "retrieval": + if hasattr(tool, "_backend"): + tool._backend = memory_backend + elif name.startswith("memory_"): + if hasattr(tool, "_backend"): + tool._backend = memory_backend + elif name.startswith("channel_"): + if hasattr(tool, "_channel"): + tool._channel = channel_backend + elif name in ( + "schedule_task", + "list_scheduled_tasks", + "pause_scheduled_task", + "resume_scheduled_task", + "cancel_scheduled_task", + ): + pass # scheduler injection handled post-build + + def _setup_sandbox(self, config): + sandbox_enabled = ( + self._sandbox if self._sandbox is not None else config.sandbox.enabled + ) + if not sandbox_enabled: + return None + try: + from openjarvis.sandbox.runner import ContainerRunner + + return ContainerRunner( + image=config.sandbox.image, + timeout=config.sandbox.timeout, + mount_allowlist_path=config.sandbox.mount_allowlist_path, + max_concurrent=config.sandbox.max_concurrent, + runtime=config.sandbox.runtime, + ) + except Exception as exc: + logger.warning("Failed to set up container sandbox: %s", exc) + return None + + def _setup_scheduler(self, config, bus): + scheduler_enabled = ( + self._scheduler if self._scheduler is not None else config.scheduler.enabled + ) + if not scheduler_enabled: + return None, None + try: + from openjarvis.scheduler.store import SchedulerStore + + db_path = config.scheduler.db_path or str( + config.hardware.platform # unused, just for fallback + ) + if not config.scheduler.db_path: + from openjarvis.core.config import DEFAULT_CONFIG_DIR + + db_path = str(DEFAULT_CONFIG_DIR / "scheduler.db") + + store = SchedulerStore(db_path=db_path) + + from openjarvis.scheduler.scheduler import TaskScheduler + + sched = TaskScheduler( + store, + poll_interval=config.scheduler.poll_interval, + bus=bus, + ) + return store, sched + except Exception as exc: + logger.warning("Failed to set up task scheduler: %s", exc) + return None, None + + def _setup_workflow(self, config, bus): + workflow_enabled = ( + self._workflow if self._workflow is not None else config.workflow.enabled + ) + if not workflow_enabled: + return None + try: + from openjarvis.workflow.engine import WorkflowEngine + + return WorkflowEngine( + bus=bus, + max_parallel=config.workflow.max_parallel, + default_node_timeout=config.workflow.default_node_timeout, + ) + except Exception as exc: + logger.warning("Failed to set up workflow engine: %s", exc) + return None + + def _setup_sessions(self, config): + sessions_enabled = ( + self._sessions if self._sessions is not None else config.sessions.enabled + ) + if not sessions_enabled: + return None + try: + from openjarvis.sessions.session import SessionStore + + return SessionStore( + db_path=config.sessions.db_path, + max_age_hours=config.sessions.max_age_hours, + consolidation_threshold=config.sessions.consolidation_threshold, + ) + except Exception as exc: + logger.warning("Failed to set up session store: %s", exc) + return None + + @staticmethod + def _setup_learning_orchestrator(config: JarvisConfig): + if not config.learning.training_enabled: + return None + try: + from openjarvis.core.config import DEFAULT_CONFIG_DIR + from openjarvis.learning.learning_orchestrator import ( + LearningOrchestrator, + ) + from openjarvis.learning.training.lora import LoRATrainingConfig + from openjarvis.traces.store import TraceStore + + trace_store = TraceStore(db_path=config.traces.db_path) + config_dir = DEFAULT_CONFIG_DIR / "agent_configs" + + sft_cfg = config.learning.intelligence.sft + lora_config = LoRATrainingConfig( + lora_rank=sft_cfg.lora_rank, + lora_alpha=sft_cfg.lora_alpha, + ) + + return LearningOrchestrator( + trace_store=trace_store, + config_dir=config_dir, + min_improvement=config.learning.min_improvement, + min_sft_pairs=sft_cfg.min_pairs, + lora_config=lora_config, + ) + except Exception as exc: + logger.warning("Failed to set up learning orchestrator: %s", exc) + return None + + def _discover_external_mcp(self, server_cfg) -> List[BaseTool]: + """Discover tools from an external MCP server configuration. + + Supports both stdio (command + args) and Streamable HTTP (url) + transports. Persists MCP clients on ``self._mcp_clients`` so + that transports stay alive for runtime tool calls. + """ + import json + + from openjarvis.mcp.client import MCPClient + from openjarvis.mcp.transport import StdioTransport, StreamableHTTPTransport + from openjarvis.tools.mcp_adapter import MCPToolProvider + + cfg = json.loads(server_cfg) if isinstance(server_cfg, str) else server_cfg + name = cfg.get("name", "") + url = cfg.get("url") + command = cfg.get("command", "") + args = cfg.get("args", []) + + if url: + transport = StreamableHTTPTransport(url=url) + elif command: + transport = StdioTransport(command=[command] + args) + else: + logger.warning( + "MCP server '%s' has neither 'url' nor 'command' — skipping", + name, + ) + return [] + + client = MCPClient(transport) + client.initialize() + + self._mcp_clients.append(client) + + provider = MCPToolProvider(client) + discovered = provider.discover() + + include_tools = set(cfg.get("include_tools", [])) + exclude_tools = set(cfg.get("exclude_tools", [])) + if include_tools: + discovered = [t for t in discovered if t.spec.name in include_tools] + if exclude_tools: + discovered = [t for t in discovered if t.spec.name not in exclude_tools] + + logger.info( + "Discovered %d tools from MCP server '%s'", + len(discovered), + name, + ) + return discovered diff --git a/src/openjarvis/system/bundles.py b/src/openjarvis/system/bundles.py new file mode 100644 index 00000000..f3d3583f --- /dev/null +++ b/src/openjarvis/system/bundles.py @@ -0,0 +1,59 @@ +"""Bundle dataclasses that group cohesive subsystems of JarvisSystem.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from openjarvis.agents._stubs import BaseAgent + from openjarvis.agents.executor import AgentExecutor + from openjarvis.agents.manager import AgentManager + from openjarvis.agents.scheduler import AgentScheduler + from openjarvis.scheduler.scheduler import TaskScheduler + from openjarvis.scheduler.store import SchedulerStore + from openjarvis.security.audit import AuditLogger + from openjarvis.security.boundary import BoundaryGuard + from openjarvis.security.capabilities import CapabilityPolicy + from openjarvis.telemetry.gpu_monitor import GpuMonitor + from openjarvis.telemetry.store import TelemetryStore + from openjarvis.traces.collector import TraceCollector + from openjarvis.traces.store import TraceStore + + +@dataclass +class SecurityContext: + """Security policy, audit, and boundary enforcement.""" + + capability_policy: Optional[CapabilityPolicy] = None + audit_logger: Optional[AuditLogger] = None + boundary_guard: Optional[BoundaryGuard] = None + + +@dataclass +class Observability: + """Telemetry, traces, and hardware monitoring.""" + + telemetry_store: Optional[TelemetryStore] = None + trace_store: Optional[TraceStore] = None + trace_collector: Optional[TraceCollector] = None + gpu_monitor: Optional[GpuMonitor] = None + + +@dataclass +class AgentRuntime: + """Active agent and agent lifecycle managers.""" + + agent: Optional[BaseAgent] = None + agent_name: str = "" + manager: Optional[AgentManager] = None + scheduler: Optional[AgentScheduler] = None + executor: Optional[AgentExecutor] = None + + +@dataclass +class Scheduling: + """Task scheduler and its persistent store.""" + + store: Optional[SchedulerStore] = None + runner: Optional[TaskScheduler] = None diff --git a/src/openjarvis/system/core.py b/src/openjarvis/system/core.py new file mode 100644 index 00000000..bfb94ed8 --- /dev/null +++ b/src/openjarvis/system/core.py @@ -0,0 +1,320 @@ +"""JarvisSystem — the fully wired system dataclass.""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +from openjarvis.core.config import JarvisConfig +from openjarvis.core.events import EventBus +from openjarvis.core.types import Message, Role +from openjarvis.engine._stubs import InferenceEngine +from openjarvis.system.bundles import ( + AgentRuntime, + Observability, + Scheduling, + SecurityContext, +) +from openjarvis.tools._stubs import BaseTool, ToolExecutor + +if TYPE_CHECKING: + from openjarvis.agents._stubs import BaseAgent + from openjarvis.agents.executor import AgentExecutor + from openjarvis.agents.manager import AgentManager + from openjarvis.agents.scheduler import AgentScheduler + from openjarvis.channels._stubs import BaseChannel + from openjarvis.learning._stubs import RouterPolicy + from openjarvis.learning.learning_orchestrator import LearningOrchestrator + from openjarvis.mcp.client import MCPClient + from openjarvis.mcp.server import MCPServer + from openjarvis.operators.manager import OperatorManager + from openjarvis.sandbox.runner import ContainerRunner + from openjarvis.scheduler.scheduler import TaskScheduler + from openjarvis.scheduler.store import SchedulerStore + from openjarvis.security.audit import AuditLogger + from openjarvis.security.boundary import BoundaryGuard + from openjarvis.security.capabilities import CapabilityPolicy + from openjarvis.sessions.session import SessionStore + from openjarvis.skills.manager import SkillManager + from openjarvis.speech._stubs import SpeechBackend + from openjarvis.system.orchestrator import QueryOrchestrator + from openjarvis.telemetry.gpu_monitor import GpuMonitor + from openjarvis.telemetry.store import TelemetryStore + from openjarvis.tools.storage._stubs import MemoryBackend + from openjarvis.traces.collector import TraceCollector + from openjarvis.traces.store import TraceStore + from openjarvis.workflow.engine import WorkflowEngine + +logger = logging.getLogger(__name__) + + +@dataclass +class JarvisSystem: + """Fully wired system -- the single source of truth for primitive composition.""" + + config: JarvisConfig + bus: EventBus + engine: InferenceEngine + engine_key: str + model: str + agent: Optional[BaseAgent] = None + agent_name: str = "" + tools: List[BaseTool] = field(default_factory=list) + tool_executor: Optional[ToolExecutor] = None + memory_backend: Optional[MemoryBackend] = None + channel_backend: Optional[BaseChannel] = None + router: Optional[RouterPolicy] = None + mcp_server: Optional[MCPServer] = None + telemetry_store: Optional[TelemetryStore] = None + trace_store: Optional[TraceStore] = None + trace_collector: Optional[TraceCollector] = None + gpu_monitor: Optional[GpuMonitor] = None + scheduler_store: Optional[SchedulerStore] = None + scheduler: Optional[TaskScheduler] = None + container_runner: Optional[ContainerRunner] = None + workflow_engine: Optional[WorkflowEngine] = None + session_store: Optional[SessionStore] = None + capability_policy: Optional[CapabilityPolicy] = None + audit_logger: Optional[AuditLogger] = None + boundary_guard: Optional[BoundaryGuard] = None + operator_manager: Optional[OperatorManager] = None + agent_manager: Optional[AgentManager] = None + agent_scheduler: Optional[AgentScheduler] = None + agent_executor: Optional[AgentExecutor] = None + speech_backend: Optional[SpeechBackend] = None + skill_manager: Optional[SkillManager] = None + _learning_orchestrator: Optional[LearningOrchestrator] = None + _mcp_clients: List[MCPClient] = field(default_factory=list) + + @property + def security(self) -> SecurityContext: + return SecurityContext( + capability_policy=self.capability_policy, + audit_logger=self.audit_logger, + boundary_guard=self.boundary_guard, + ) + + @property + def observability(self) -> Observability: + return Observability( + telemetry_store=self.telemetry_store, + trace_store=self.trace_store, + trace_collector=self.trace_collector, + gpu_monitor=self.gpu_monitor, + ) + + @property + def agents(self) -> AgentRuntime: + return AgentRuntime( + agent=self.agent, + agent_name=self.agent_name, + manager=self.agent_manager, + scheduler=self.agent_scheduler, + executor=self.agent_executor, + ) + + @property + def scheduling(self) -> Scheduling: + return Scheduling( + store=self.scheduler_store, + runner=self.scheduler, + ) + + def _get_orchestrator(self) -> QueryOrchestrator: + orch = self.__dict__.get("_orchestrator") + if orch is None: + from openjarvis.system.orchestrator import QueryOrchestrator + + orch = QueryOrchestrator(self) + self.__dict__["_orchestrator"] = orch + return orch + + def ask( + self, + query: str, + *, + context: bool = True, + temperature: Optional[float] = None, + max_tokens: Optional[int] = None, + agent: Optional[str] = None, + tools: Optional[List[str]] = None, + system_prompt: Optional[str] = None, + operator_id: Optional[str] = None, + prior_messages: Optional[List[Message]] = None, + ) -> Dict[str, Any]: + return self._get_orchestrator().ask( + query, + context=context, + temperature=temperature, + max_tokens=max_tokens, + agent=agent, + tools=tools, + system_prompt=system_prompt, + operator_id=operator_id, + prior_messages=prior_messages, + ) + + def _detect_agent_intent(self, query: str) -> Optional[str]: + return self._get_orchestrator()._detect_agent_intent(query) + + def _build_tools(self, tool_names: List[str]) -> List[BaseTool]: + return self._get_orchestrator()._build_tools(tool_names) + + def _run_agent( + self, + query, + messages, + agent_name, + tool_names, + temperature, + max_tokens, + *, + system_prompt=None, + operator_id=None, + prior_messages=None, + ) -> Dict[str, Any]: + return self._get_orchestrator()._run_agent( + query, + messages, + agent_name, + tool_names, + temperature, + max_tokens, + system_prompt=system_prompt, + operator_id=operator_id, + prior_messages=prior_messages, + ) + + def wire_channel(self, channel_bridge: Any) -> None: + """Register a message handler on *channel_bridge* that routes every + incoming message through this system (agent or engine) and replies. + + Sessions are isolated per ``":"`` key so + each chat retains its own history. + + Parameters + ---------- + channel_bridge: + A connected :class:`~openjarvis.channels._stubs.BaseChannel` + instance whose ``on_message`` method accepts a callable. + """ + from openjarvis.core.types import Message + from openjarvis.sessions.session import SessionStore + + if self.session_store is None: + from pathlib import Path + + self.session_store = SessionStore( + db_path=Path(self.config.sessions.db_path).expanduser(), + max_age_hours=self.config.sessions.max_age_hours, + consolidation_threshold=self.config.sessions.consolidation_threshold, + ) + + _system = self # capture for closure + + def _on_channel_message(cm) -> None: + session_key = f"{cm.channel}:{cm.conversation_id}" + session = _system.session_store.get_or_create( + session_key, + channel=cm.channel, + channel_user_id=cm.sender, + ) + + prior_msgs: List[Message] = [] + for sm in session.messages: + try: + role = Role(sm.role) + except ValueError: + role = Role.USER + prior_msgs.append(Message(role=role, content=sm.content)) + + reply = "" + try: + if _system.agent_name and _system.agent_name != "none": + result = _system.ask( + cm.content, + context=False, + agent=_system.agent_name, + prior_messages=prior_msgs, + ) + reply = result.get("content", "") + else: + result = _system.ask( + cm.content, + context=False, + prior_messages=prior_msgs, + ) + reply = result.get("content", "") + except Exception: + logger.exception("Channel message handler error") + reply = "Sorry, I encountered an error processing your message." + + try: + _system.session_store.save_message( + session.session_id, + "user", + cm.content, + channel=cm.channel, + ) + _system.session_store.save_message( + session.session_id, + "assistant", + reply, + channel=cm.channel, + ) + except Exception: + logger.debug("Session save error", exc_info=True) + + if reply: + try: + channel_bridge.send( + cm.channel, + reply, + conversation_id=cm.conversation_id, + ) + except Exception: + logger.exception("Channel send error") + + channel_bridge.on_message(_on_channel_message) + + def _close_mcp_clients(self) -> None: + """Close all persistent MCP client connections.""" + for client in self._mcp_clients: + try: + client.close() + except Exception: + logger.debug("Error closing MCP client", exc_info=True) + + def close(self) -> None: + """Release resources.""" + if self.scheduler and hasattr(self.scheduler, "stop"): + self.scheduler.stop() + for resource in ( + self.scheduler_store, + self.engine, + self.gpu_monitor, + self.telemetry_store, + self.trace_store, + self.memory_backend, + self.session_store, + self.channel_backend, + self.workflow_engine, + self.container_runner, + ): + if resource and hasattr(resource, "close"): + resource.close() + if self.agent_manager is not None: + self.agent_manager.close() + if self.agent_scheduler is not None: + self.agent_scheduler.stop() + self._close_mcp_clients() + + def __enter__(self) -> JarvisSystem: + return self + + def __exit__(self, *exc: Any) -> None: + self.close() + + +__all__ = ["JarvisSystem"] diff --git a/src/openjarvis/system/orchestrator.py b/src/openjarvis/system/orchestrator.py new file mode 100644 index 00000000..8570871c --- /dev/null +++ b/src/openjarvis/system/orchestrator.py @@ -0,0 +1,314 @@ +"""Executes user queries through the engine or through an agent.""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +from openjarvis.core.types import Message, Role +from openjarvis.tools._stubs import BaseTool + +if TYPE_CHECKING: + from openjarvis.system.protocols import OrchestratorDeps + +logger = logging.getLogger(__name__) + + +class QueryOrchestrator: + def __init__(self, system: OrchestratorDeps) -> None: + self._system = system + + def ask( + self, + query: str, + *, + context: bool = True, + temperature: Optional[float] = None, + max_tokens: Optional[int] = None, + agent: Optional[str] = None, + tools: Optional[List[str]] = None, + system_prompt: Optional[str] = None, + operator_id: Optional[str] = None, + prior_messages: Optional[List[Message]] = None, + ) -> Dict[str, Any]: + """Execute a query through the system and return a result dict.""" + s = self._system + if temperature is None: + temperature = s.config.intelligence.temperature + if max_tokens is None: + max_tokens = s.config.intelligence.max_tokens + + messages = [Message(role=Role.USER, content=query)] + + if context and s.memory_backend and s.config.agent.context_from_memory: + try: + from openjarvis.tools.storage.context import ( + ContextConfig, + inject_context, + ) + + ctx_cfg = ContextConfig( + top_k=s.config.memory.context_top_k, + min_score=s.config.memory.context_min_score, + max_context_tokens=s.config.memory.context_max_tokens, + ) + messages = inject_context( + query, + messages, + s.memory_backend, + config=ctx_cfg, + ) + except Exception as exc: + logger.warning("Failed to inject memory context: %s", exc) + + use_agent = agent or s.agent_name + if not agent and use_agent != "none": + detected = self._detect_agent_intent(query) + if detected: + use_agent = detected + if use_agent and use_agent != "none": + return self._run_agent( + query, + messages, + use_agent, + tools, + temperature, + max_tokens, + system_prompt=system_prompt, + operator_id=operator_id, + prior_messages=prior_messages, + ) + + result = s.engine.generate( + messages, + model=s.model, + temperature=temperature, + max_tokens=max_tokens, + ) + return { + "content": result.get("content", ""), + "usage": result.get("usage", {}), + "model": s.model, + "engine": s.engine_key, + } + + def _detect_agent_intent(self, query: str) -> Optional[str]: + """Detect if a query should be routed to a specific agent.""" + import re + + from openjarvis.core.registry import AgentRegistry + + if re.search( + r"\b(good\s+morning|morning\s+digest|daily\s+briefing|morning\s+briefing)\b", + query, + re.IGNORECASE, + ): + if AgentRegistry.contains("morning_digest"): + return "morning_digest" + + return None + + def _run_agent( + self, + query, + messages, + agent_name, + tool_names, + temperature, + max_tokens, + *, + system_prompt=None, + operator_id=None, + prior_messages=None, + ) -> Dict[str, Any]: + """Run through an agent.""" + from openjarvis.agents._stubs import AgentContext + from openjarvis.core.events import EventType + from openjarvis.core.registry import AgentRegistry + + s = self._system + + try: + agent_cls = AgentRegistry.get(agent_name) + except KeyError: + return {"content": f"Unknown agent: {agent_name}", "error": True} + + agent_tools = s.tools + if tool_names: + agent_tools = self._build_tools(tool_names) + + ctx = AgentContext() + + if prior_messages: + for msg in prior_messages: + ctx.conversation.add(msg) + + if messages and len(messages) > 1: + for msg in messages[:-1]: + ctx.conversation.add(msg) + + agent_kwargs: Dict[str, Any] = { + "bus": s.bus, + "temperature": temperature, + "max_tokens": max_tokens, + } + if getattr(agent_cls, "accepts_tools", False): + agent_kwargs["tools"] = agent_tools + agent_kwargs["max_turns"] = s.config.agent.max_turns + examples = getattr(s, "_skill_few_shot_examples", None) + if examples: + agent_kwargs["skill_few_shot_examples"] = examples + if system_prompt is not None: + agent_kwargs["system_prompt"] = system_prompt + if s.capability_policy is not None: + agent_kwargs["capability_policy"] = s.capability_policy + if operator_id is not None: + agent_kwargs["operator_id"] = operator_id + agent_kwargs["session_store"] = s.session_store + agent_kwargs["memory_backend"] = s.memory_backend + + if agent_name == "morning_digest" and hasattr(s.config, "digest"): + dc = s.config.digest + section_sources = {} + for sec in dc.sections: + sc = getattr(dc, sec, None) + if sc and hasattr(sc, "sources"): + section_sources[sec] = sc.sources + agent_kwargs.update( + { + "persona": dc.persona, + "sections": dc.sections, + "section_sources": section_sources, + "timezone": dc.timezone, + "voice_id": dc.voice_id, + "voice_speed": dc.voice_speed, + "tts_backend": dc.tts_backend, + "honorific": dc.honorific, + } + ) + from openjarvis.tools.digest_collect import DigestCollectTool + from openjarvis.tools.text_to_speech import TextToSpeechTool + + digest_tools = [DigestCollectTool(), TextToSpeechTool()] + existing = agent_kwargs.get("tools", []) + agent_kwargs["tools"] = digest_tools + list(existing) + + try: + ag = agent_cls(s.engine, s.model, **agent_kwargs) + except TypeError: + try: + ag = agent_cls(s.engine, s.model) + except TypeError: + ag = agent_cls() + + telemetry_events: List[Dict[str, Any]] = [] + + def _on_inference_end(event: Any) -> None: + telemetry_events.append(event.data if hasattr(event, "data") else event) + + s.bus.subscribe(EventType.INFERENCE_END, _on_inference_end) + + # Check trace_store (set at build time) instead of config.traces.enabled + # because the shared config singleton can be mutated by other SystemBuilder + # instances (e.g. the judge backend). + try: + if s.trace_store is not None: + from openjarvis.traces.collector import TraceCollector + + collector = TraceCollector( + ag, + store=s.trace_store, + bus=s.bus, + ) + result = collector.run(query, context=ctx) + s.trace_collector = collector + else: + result = ag.run(query, context=ctx) + finally: + s.bus.unsubscribe(EventType.INFERENCE_END, _on_inference_end) + + _telemetry: Dict[str, Any] = {} + if telemetry_events: + total_energy = sum(e.get("energy_joules", 0.0) for e in telemetry_events) + total_latency = sum(e.get("latency", 0.0) for e in telemetry_events) + power_vals = [ + e.get("power_watts", 0.0) + for e in telemetry_events + if e.get("power_watts", 0.0) > 0 + ] + util_vals = [ + e.get("gpu_utilization_pct", 0.0) + for e in telemetry_events + if e.get("gpu_utilization_pct", 0.0) > 0 + ] + throughput_vals = [ + e.get("throughput_tok_per_sec", 0.0) + for e in telemetry_events + if e.get("throughput_tok_per_sec", 0.0) > 0 + ] + _telemetry = { + "ttft": telemetry_events[0].get("ttft", 0.0), + "energy_joules": total_energy, + "power_watts": ( + sum(power_vals) / len(power_vals) if power_vals else 0.0 + ), + "gpu_utilization_pct": ( + sum(util_vals) / len(util_vals) if util_vals else 0.0 + ), + "throughput_tok_per_sec": ( + sum(throughput_vals) / len(throughput_vals) + if throughput_vals + else 0.0 + ), + "gpu_memory_used_gb": max( + (e.get("gpu_memory_used_gb", 0.0) for e in telemetry_events), + default=0.0, + ), + "gpu_temperature_c": max( + (e.get("gpu_temperature_c", 0.0) for e in telemetry_events), + default=0.0, + ), + "inference_calls": len(telemetry_events), + "total_inference_latency": total_latency, + } + + return { + "content": result.content, + "usage": getattr(result, "usage", {}), + "tool_results": [ + { + "tool_name": tr.tool_name, + "content": tr.content, + "success": tr.success, + "arguments": tr.metadata.get("arguments", {}), + } + for tr in getattr(result, "tool_results", []) + ], + "turns": getattr(result, "turns", 1), + "metadata": getattr(result, "metadata", {}), + "model": s.model, + "engine": s.engine_key, + "_telemetry": _telemetry, + } + + def _build_tools(self, tool_names: List[str]) -> List[BaseTool]: + """Build tool instances from tool names.""" + from openjarvis.core.registry import ToolRegistry + + s = self._system + tools: List[BaseTool] = [] + for name in tool_names: + try: + if name == "retrieval" and s.memory_backend: + from openjarvis.tools.retrieval import RetrievalTool + + tools.append(RetrievalTool(s.memory_backend)) + elif name == "llm": + from openjarvis.tools.llm_tool import LLMTool + + tools.append(LLMTool(s.engine, model=s.model)) + elif ToolRegistry.contains(name): + tools.append(ToolRegistry.create(name)) + except Exception as exc: + logger.warning("Failed to build tool %r: %s", name, exc) + return tools diff --git a/src/openjarvis/system/protocols.py b/src/openjarvis/system/protocols.py new file mode 100644 index 00000000..405bc5fa --- /dev/null +++ b/src/openjarvis/system/protocols.py @@ -0,0 +1,40 @@ +"""Structural protocols for substituting fakes in place of JarvisSystem.""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, List, Optional, Protocol + +if TYPE_CHECKING: + from openjarvis.core.config import JarvisConfig + from openjarvis.core.events import EventBus + from openjarvis.engine._stubs import InferenceEngine + from openjarvis.security.capabilities import CapabilityPolicy + from openjarvis.sessions.session import SessionStore + from openjarvis.tools._stubs import BaseTool + from openjarvis.tools.storage._stubs import MemoryBackend + from openjarvis.traces.collector import TraceCollector + from openjarvis.traces.store import TraceStore + + +class OrchestratorDeps(Protocol): + """Minimum surface of JarvisSystem that QueryOrchestrator depends on. + + Tests can satisfy this with a lightweight class — no need to construct + the full JarvisSystem dataclass or materialize every subsystem. + """ + + config: JarvisConfig + bus: EventBus + engine: InferenceEngine + engine_key: str + model: str + agent_name: str + tools: List[BaseTool] + memory_backend: Optional[MemoryBackend] + capability_policy: Optional[CapabilityPolicy] + session_store: Optional[SessionStore] + trace_store: Optional[TraceStore] + trace_collector: Optional[TraceCollector] # written by _run_agent + + # Optional attribute (getattr with default) — declared for type clarity. + _skill_few_shot_examples: Any diff --git a/tests/mcp/test_discovery.py b/tests/mcp/test_discovery.py index ac9c061e..6e516966 100644 --- a/tests/mcp/test_discovery.py +++ b/tests/mcp/test_discovery.py @@ -36,7 +36,7 @@ _PATCH_HTTP = "openjarvis.mcp.transport.StreamableHTTPTransport" _PATCH_STDIO = "openjarvis.mcp.transport.StdioTransport" _PATCH_CLIENT = "openjarvis.mcp.client.MCPClient" _PATCH_PROVIDER = "openjarvis.tools.mcp_adapter.MCPToolProvider" -_PATCH_LOGGER = "openjarvis.system.logger" +_PATCH_LOGGER = "openjarvis.system.builder.logger" class TestDiscoverHTTPServer: diff --git a/tests/operators/test_operators.py b/tests/operators/test_operators.py index 87e51eb4..6e965748 100644 --- a/tests/operators/test_operators.py +++ b/tests/operators/test_operators.py @@ -717,7 +717,9 @@ class TestSystemAskPassthrough: captured.update(kwargs) return {"content": "OK"} - with patch.object(JarvisSystem, "_run_agent", patched_run_agent): + from openjarvis.system import QueryOrchestrator + + with patch.object(QueryOrchestrator, "_run_agent", patched_run_agent): real_system = JarvisSystem( config=system.config, bus=system.bus, @@ -752,7 +754,9 @@ class TestSystemAskPassthrough: captured.update(kwargs) return {"content": "OK"} - with patch.object(JarvisSystem, "_run_agent", patched_run_agent): + from openjarvis.system import QueryOrchestrator + + with patch.object(QueryOrchestrator, "_run_agent", patched_run_agent): real_system = JarvisSystem( config=system.config, bus=system.bus, diff --git a/tests/test_query_orchestrator.py b/tests/test_query_orchestrator.py new file mode 100644 index 00000000..da9b1323 --- /dev/null +++ b/tests/test_query_orchestrator.py @@ -0,0 +1,145 @@ +"""Isolated QueryOrchestrator tests using a minimal fake system.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +import pytest + +from openjarvis.core.config import JarvisConfig +from openjarvis.core.events import EventBus +from openjarvis.system import QueryOrchestrator + + +class _FakeEngine: + def __init__(self, reply: Dict[str, Any]) -> None: + self._reply = reply + self.calls: List[Dict[str, Any]] = [] + + def generate(self, messages, *, model, temperature, max_tokens, **_): + self.calls.append( + { + "messages": list(messages), + "model": model, + "temperature": temperature, + "max_tokens": max_tokens, + } + ) + return self._reply + + def list_models(self): + return [] + + +@dataclass +class _FakeSystem: + """Minimum surface QueryOrchestrator reads — no subsystems wired.""" + + config: JarvisConfig = field(default_factory=JarvisConfig) + bus: EventBus = field(default_factory=EventBus) + engine: Any = None + engine_key: str = "fake" + model: str = "fake-model" + agent_name: str = "" + tools: List[Any] = field(default_factory=list) + memory_backend: Optional[Any] = None + capability_policy: Optional[Any] = None + session_store: Optional[Any] = None + trace_store: Optional[Any] = None + trace_collector: Optional[Any] = None + _skill_few_shot_examples: Optional[List[str]] = None + + +class TestAskDirectEngineMode: + def test_direct_engine_returns_content(self): + engine = _FakeEngine({"content": "hi there", "usage": {"tokens": 4}}) + system = _FakeSystem(engine=engine) + orchestrator = QueryOrchestrator(system) + + result = orchestrator.ask("hello", context=False) + + assert result["content"] == "hi there" + assert result["usage"] == {"tokens": 4} + assert result["model"] == "fake-model" + assert result["engine"] == "fake" + + def test_forwards_temperature_and_max_tokens(self): + engine = _FakeEngine({"content": ""}) + system = _FakeSystem(engine=engine) + orchestrator = QueryOrchestrator(system) + + orchestrator.ask("q", context=False, temperature=0.9, max_tokens=128) + + assert engine.calls[0]["temperature"] == 0.9 + assert engine.calls[0]["max_tokens"] == 128 + + def test_uses_config_defaults_when_omitted(self): + engine = _FakeEngine({"content": ""}) + config = JarvisConfig() + config.intelligence.temperature = 0.42 + config.intelligence.max_tokens = 77 + system = _FakeSystem(config=config, engine=engine) + orchestrator = QueryOrchestrator(system) + + orchestrator.ask("q", context=False) + + assert engine.calls[0]["temperature"] == 0.42 + assert engine.calls[0]["max_tokens"] == 77 + + +class TestAskAgentRouting: + def test_agent_none_stays_on_engine(self): + engine = _FakeEngine({"content": "engine path"}) + system = _FakeSystem(engine=engine, agent_name="none") + orchestrator = QueryOrchestrator(system) + + result = orchestrator.ask("plain question", context=False) + + assert result["content"] == "engine path" + assert len(engine.calls) == 1 + + def test_unknown_agent_returns_error_dict(self): + engine = _FakeEngine({"content": ""}) + system = _FakeSystem(engine=engine, agent_name="does_not_exist") + orchestrator = QueryOrchestrator(system) + + result = orchestrator.ask("q", context=False) + + assert result.get("error") is True + assert "does_not_exist" in result["content"] + + +class TestDetectAgentIntent: + @pytest.mark.parametrize( + "query", + [ + "good morning", + "morning digest please", + "can you run the daily briefing", + "morning briefing time", + ], + ) + def test_morning_digest_triggers(self, query): + from openjarvis.core.registry import AgentRegistry + + # Register a stub so the intent check returns the name. + try: + AgentRegistry.get("morning_digest") + registered = True + except KeyError: + registered = False + + system = _FakeSystem() + orchestrator = QueryOrchestrator(system) + detected = orchestrator._detect_agent_intent(query) + + if registered: + assert detected == "morning_digest" + else: + assert detected is None + + def test_plain_query_returns_none(self): + system = _FakeSystem() + orchestrator = QueryOrchestrator(system) + assert orchestrator._detect_agent_intent("what's the weather") is None