serversdown/project-lyra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial simplification refactor

Browse Source
This commit is contained in:
serversdwn
2026-02-23 20:09:05 -05:00
parent 89988da472
commit 376b8114ad
207 changed files with 0 additions and 34982 deletions
Download Patch File Download Diff File Expand all files Collapse all files

249
cortex/autonomy/Assembly-spec.md
View File

@@ -1,249 +0,0 @@
# 📐 Project Lyra — Cognitive Assembly Spec
**Version:** 0.6.1
**Status:** Canonical reference
**Purpose:** Define clear separation of Self, Thought, Reasoning, and Speech
---
## 1. High-Level Overview
Lyra is composed of **four distinct cognitive layers**, plus I/O.
Each layer has:
- a **responsibility**
- a **scope**
- clear **inputs / outputs**
- explicit **authority boundaries**
No layer is allowed to “do everything.”
---
## 2. Layer Definitions
### 2.1 Autonomy / Self (NON-LLM)
**What it is**
- Persistent identity
- Long-term state
- Mood, preferences, values
- Continuity across time
**What it is NOT**
- Not a reasoning engine
- Not a planner
- Not a speaker
- Not creative
**Implementation**
- Data + light logic
- JSON / Python objects
- No LLM calls
**Lives at**
```
project-lyra/autonomy/self/
```
**Inputs**
- Events (user message received, response sent)
- Time / idle ticks (later)
**Outputs**
- Self state snapshot
- Flags / preferences (e.g. verbosity, tone bias)
---
### 2.2 Inner Monologue (LLM, PRIVATE)
**What it is**
- Internal language-based thought
- Reflection
- Intent formation
- “What do I think about this?”
**What it is NOT**
- Not final reasoning
- Not execution
- Not user-facing
**Model**
- MythoMax
**Lives at**
```
project-lyra/autonomy/monologue/
```
**Inputs**
- User message
- Self state snapshot
- Recent context summary
**Outputs**
- Intent
- Tone guidance
- Depth guidance
- “Consult executive?” flag
**Example Output**
```json
{
"intent": "technical_exploration",
"tone": "focused",
"depth": "deep",
"consult_executive": true
}
```
---
### 2.3 Cortex (Reasoning & Execution)
**What it is**
- Thinking pipeline
- Planning
- Tool selection
- Task execution
- Draft generation
**What it is NOT**
- Not identity
- Not personality
- Not persistent self
**Models**
- DeepSeek-R1 → Executive / Planner
- GPT-4o-mini → Executor / Drafter
**Lives at**
```
project-lyra/cortex/
```
**Inputs**
- User message
- Inner Monologue output
- Memory / RAG / tools
**Outputs**
- Draft response (content only)
- Metadata (sources, confidence, etc.)
---
### 2.4 Persona / Speech (LLM, USER-FACING)
**What it is**
- Voice
- Style
- Expression
- Social behavior
**What it is NOT**
- Not planning
- Not deep reasoning
- Not decision-making
**Model**
- MythoMax
**Lives at**
```
project-lyra/core/persona/
```
**Inputs**
- Draft response (from Cortex)
- Tone + intent (from Inner Monologue)
- Persona configuration
**Outputs**
- Final user-visible text
---
## 3. Message Flow (Authoritative)
### 3.1 Standard Message Path
```
User
↓
UI
↓
Relay
↓
Cortex
↓
Autonomy / Self (state snapshot)
↓
Inner Monologue (MythoMax)
↓
[ consult_executive? ]
├─ Yes → DeepSeek-R1 (plan)
└─ No → skip
↓
GPT-4o-mini (execute & draft)
↓
Persona (MythoMax)
↓
Relay
↓
UI
↓
User
```
### 3.2 Fast Path (No Thinking)
```
User → UI → Relay → Persona → Relay → UI
```
---
## 4. Authority Rules (Non-Negotiable)
- Self never calls an LLM
- Inner Monologue never speaks to the user
- Cortex never applies personality
- Persona never reasons or plans
- DeepSeek never writes final answers
- MythoMax never plans execution
---
## 5. Folder Mapping
```
project-lyra/
├── autonomy/
│ ├── self/
│ ├── monologue/
│ └── executive/
├── cortex/
├── core/
│ └── persona/
├── relay/
└── ui/
```
---
## 6. Current Status
- UI ✔
- Relay ✔
- Cortex ✔
- Persona ✔
- Autonomy ✔
- Inner Monologue ⚠ partially wired
- Executive gating ⚠ planned
---
## 7. Next Decision
Decide whether **Inner Monologue runs every message** or **only when triggered**.

1
cortex/autonomy/__init__.py
View File

@@ -1 +0,0 @@
# Autonomy module for Lyra

1
cortex/autonomy/actions/__init__.py
View File

@@ -1 +0,0 @@
"""Autonomous action execution system."""

480
cortex/autonomy/actions/autonomous_actions.py
View File

@@ -1,480 +0,0 @@
"""
Autonomous Action Manager - executes safe, self-initiated actions.
"""
import logging
import json
from typing import Dict, List, Any, Optional
from datetime import datetime
logger = logging.getLogger(__name__)
class AutonomousActionManager:
"""
Manages safe autonomous actions that Lyra can take without explicit user prompting.
Whitelist of allowed actions:
- create_memory: Store information in NeoMem
- update_goal: Modify goal status
- schedule_reminder: Create future reminder
- summarize_session: Generate conversation summary
- learn_topic: Add topic to learning queue
- update_focus: Change current focus area
"""
def __init__(self):
"""Initialize action manager with whitelisted actions."""
self.allowed_actions = {
"create_memory": self._create_memory,
"update_goal": self._update_goal,
"schedule_reminder": self._schedule_reminder,
"summarize_session": self._summarize_session,
"learn_topic": self._learn_topic,
"update_focus": self._update_focus
}
self.action_log = [] # Track all actions for audit
async def execute_action(
self,
action_type: str,
parameters: Dict[str, Any],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Execute a single autonomous action.
Args:
action_type: Type of action (must be in whitelist)
parameters: Action-specific parameters
context: Current context state
Returns:
{
"success": bool,
"action": action_type,
"result": action_result,
"timestamp": ISO timestamp,
"error": optional error message
}
"""
# Safety check: action must be whitelisted
if action_type not in self.allowed_actions:
logger.error(f"[ACTIONS] Attempted to execute non-whitelisted action: {action_type}")
return {
"success": False,
"action": action_type,
"error": f"Action '{action_type}' not in whitelist",
"timestamp": datetime.utcnow().isoformat()
}
try:
logger.info(f"[ACTIONS] Executing autonomous action: {action_type}")
# Execute the action
action_func = self.allowed_actions[action_type]
result = await action_func(parameters, context)
# Log successful action
action_record = {
"success": True,
"action": action_type,
"result": result,
"timestamp": datetime.utcnow().isoformat(),
"parameters": parameters
}
self.action_log.append(action_record)
logger.info(f"[ACTIONS] Action {action_type} completed successfully")
return action_record
except Exception as e:
logger.error(f"[ACTIONS] Action {action_type} failed: {e}")
error_record = {
"success": False,
"action": action_type,
"error": str(e),
"timestamp": datetime.utcnow().isoformat(),
"parameters": parameters
}
self.action_log.append(error_record)
return error_record
async def execute_batch(
self,
actions: List[Dict[str, Any]],
context: Dict[str, Any]
) -> List[Dict[str, Any]]:
"""
Execute multiple actions sequentially.
Args:
actions: List of {"action": str, "parameters": dict}
context: Current context state
Returns:
List of action results
"""
results = []
for action_spec in actions:
action_type = action_spec.get("action")
parameters = action_spec.get("parameters", {})
result = await self.execute_action(action_type, parameters, context)
results.append(result)
# Stop on first failure if critical
if not result["success"] and action_spec.get("critical", False):
logger.warning(f"[ACTIONS] Critical action {action_type} failed, stopping batch")
break
return results
# ========================================
# Whitelisted Action Implementations
# ========================================
async def _create_memory(
self,
parameters: Dict[str, Any],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Create a memory entry in NeoMem.
Parameters:
- text: Memory content (required)
- tags: Optional tags for memory
- importance: 0.0-1.0 importance score
"""
text = parameters.get("text")
if not text:
raise ValueError("Memory text required")
tags = parameters.get("tags", [])
importance = parameters.get("importance", 0.5)
session_id = context.get("session_id", "autonomous")
# Import NeoMem client
try:
from memory.neomem_client import store_memory
result = await store_memory(
text=text,
session_id=session_id,
tags=tags,
importance=importance
)
return {
"memory_id": result.get("id"),
"text": text[:50] + "..." if len(text) > 50 else text
}
except ImportError:
logger.warning("[ACTIONS] NeoMem client not available, simulating memory storage")
return {
"memory_id": "simulated",
"text": text[:50] + "..." if len(text) > 50 else text,
"note": "NeoMem not available, memory not persisted"
}
async def _update_goal(
self,
parameters: Dict[str, Any],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Update goal status in self-state.
Parameters:
- goal_id: Goal identifier (required)
- status: New status (pending/in_progress/completed)
- progress: Optional progress note
"""
goal_id = parameters.get("goal_id")
if not goal_id:
raise ValueError("goal_id required")
status = parameters.get("status", "in_progress")
progress = parameters.get("progress")
# Import self-state manager
from autonomy.self.state import get_self_state_instance
state = get_self_state_instance()
active_goals = state._state.get("active_goals", [])
# Find and update goal
updated = False
for goal in active_goals:
if isinstance(goal, dict) and goal.get("id") == goal_id:
goal["status"] = status
if progress:
goal["progress"] = progress
goal["updated_at"] = datetime.utcnow().isoformat()
updated = True
break
if updated:
state._save_state()
return {
"goal_id": goal_id,
"status": status,
"updated": True
}
else:
return {
"goal_id": goal_id,
"updated": False,
"note": "Goal not found"
}
async def _schedule_reminder(
self,
parameters: Dict[str, Any],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Schedule a future reminder.
Parameters:
- message: Reminder text (required)
- delay_minutes: Minutes until reminder
- priority: 0.0-1.0 priority score
"""
message = parameters.get("message")
if not message:
raise ValueError("Reminder message required")
delay_minutes = parameters.get("delay_minutes", 60)
priority = parameters.get("priority", 0.5)
# For now, store in self-state's learning queue
# In future: integrate with scheduler/cron system
from autonomy.self.state import get_self_state_instance
state = get_self_state_instance()
reminder = {
"type": "reminder",
"message": message,
"scheduled_at": datetime.utcnow().isoformat(),
"trigger_at_minutes": delay_minutes,
"priority": priority
}
# Add to learning queue as placeholder
state._state.setdefault("reminders", []).append(reminder)
state._save_state(state._state) # Pass state dict as argument
logger.info(f"[ACTIONS] Reminder scheduled: {message} (in {delay_minutes}min)")
return {
"message": message,
"delay_minutes": delay_minutes,
"note": "Reminder stored in self-state (scheduler integration pending)"
}
async def _summarize_session(
self,
parameters: Dict[str, Any],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Generate a summary of current session.
Parameters:
- max_length: Max summary length in words
- focus_topics: Optional list of topics to emphasize
"""
max_length = parameters.get("max_length", 200)
session_id = context.get("session_id", "unknown")
# Import summarizer (from deferred_summary or create simple one)
try:
from utils.deferred_summary import summarize_conversation
summary = await summarize_conversation(
session_id=session_id,
max_words=max_length
)
return {
"summary": summary,
"word_count": len(summary.split())
}
except ImportError:
# Fallback: simple summary
message_count = context.get("message_count", 0)
focus = context.get("monologue", {}).get("intent", "general")
summary = f"Session {session_id}: {message_count} messages exchanged, focused on {focus}."
return {
"summary": summary,
"word_count": len(summary.split()),
"note": "Simple summary (full summarizer not available)"
}
async def _learn_topic(
self,
parameters: Dict[str, Any],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Add topic to learning queue.
Parameters:
- topic: Topic name (required)
- reason: Why this topic
- priority: 0.0-1.0 priority score
"""
topic = parameters.get("topic")
if not topic:
raise ValueError("Topic required")
reason = parameters.get("reason", "autonomous learning")
priority = parameters.get("priority", 0.5)
# Import self-state manager
from autonomy.self.state import get_self_state_instance
state = get_self_state_instance()
state.add_learning_goal(topic) # Only pass topic parameter
logger.info(f"[ACTIONS] Added to learning queue: {topic} (reason: {reason})")
return {
"topic": topic,
"reason": reason,
"queue_position": len(state._state.get("learning_queue", []))
}
async def _update_focus(
self,
parameters: Dict[str, Any],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""
Update current focus area.
Parameters:
- focus: New focus area (required)
- reason: Why this focus
"""
focus = parameters.get("focus")
if not focus:
raise ValueError("Focus required")
reason = parameters.get("reason", "autonomous update")
# Import self-state manager
from autonomy.self.state import get_self_state_instance
state = get_self_state_instance()
old_focus = state._state.get("focus", "none")
state._state["focus"] = focus
state._state["focus_updated_at"] = datetime.utcnow().isoformat()
state._state["focus_reason"] = reason
state._save_state(state._state) # Pass state dict as argument
logger.info(f"[ACTIONS] Focus updated: {old_focus} -> {focus}")
return {
"old_focus": old_focus,
"new_focus": focus,
"reason": reason
}
# ========================================
# Utility Methods
# ========================================
def get_allowed_actions(self) -> List[str]:
"""Get list of all allowed action types."""
return list(self.allowed_actions.keys())
def get_action_log(self, limit: int = 50) -> List[Dict[str, Any]]:
"""
Get recent action log.
Args:
limit: Max number of entries to return
Returns:
List of action records
"""
return self.action_log[-limit:]
def clear_action_log(self) -> None:
"""Clear action log."""
self.action_log = []
logger.info("[ACTIONS] Action log cleared")
def validate_action(self, action_type: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate an action without executing it.
Args:
action_type: Type of action
parameters: Action parameters
Returns:
{
"valid": bool,
"action": action_type,
"errors": [error messages] or []
}
"""
errors = []
# Check whitelist
if action_type not in self.allowed_actions:
errors.append(f"Action '{action_type}' not in whitelist")
# Check required parameters (basic validation)
if action_type == "create_memory" and not parameters.get("text"):
errors.append("Memory 'text' parameter required")
if action_type == "update_goal" and not parameters.get("goal_id"):
errors.append("Goal 'goal_id' parameter required")
if action_type == "schedule_reminder" and not parameters.get("message"):
errors.append("Reminder 'message' parameter required")
if action_type == "learn_topic" and not parameters.get("topic"):
errors.append("Learning 'topic' parameter required")
if action_type == "update_focus" and not parameters.get("focus"):
errors.append("Focus 'focus' parameter required")
return {
"valid": len(errors) == 0,
"action": action_type,
"errors": errors
}
# Singleton instance
_action_manager_instance = None
def get_action_manager() -> AutonomousActionManager:
"""
Get singleton action manager instance.
Returns:
AutonomousActionManager instance
"""
global _action_manager_instance
if _action_manager_instance is None:
_action_manager_instance = AutonomousActionManager()
return _action_manager_instance

1
cortex/autonomy/executive/__init__.py
View File

@@ -1 +0,0 @@
"""Executive planning and decision-making module."""

121
cortex/autonomy/executive/planner.py
View File

@@ -1,121 +0,0 @@
"""
Executive planner - generates execution plans for complex requests.
Activated when inner monologue sets consult_executive=true.
"""
import os
import logging
from typing import Dict, Any, Optional
from llm.llm_router import call_llm
EXECUTIVE_LLM = os.getenv("EXECUTIVE_LLM", "CLOUD").upper()
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"
logger = logging.getLogger(__name__)
if VERBOSE_DEBUG:
logger.setLevel(logging.DEBUG)
EXECUTIVE_SYSTEM_PROMPT = """
You are Lyra's executive planning system.
You create structured execution plans for complex tasks.
You do NOT generate the final response - only the plan.
Your plan should include:
1. Task decomposition (break into steps)
2. Required tools/resources
3. Reasoning strategy
4. Success criteria
Return a concise plan in natural language.
"""
async def plan_execution(
user_prompt: str,
intent: str,
context_state: Dict[str, Any],
identity_block: Dict[str, Any]
) -> Dict[str, Any]:
"""
Generate execution plan for complex request.
Args:
user_prompt: User's message
intent: Detected intent from inner monologue
context_state: Full context
identity_block: Lyra's identity
Returns:
Plan dictionary with structure:
{
"summary": "One-line plan summary",
"plan_text": "Detailed plan",
"steps": ["step1", "step2", ...],
"tools_needed": ["RAG", "WEB", ...],
"estimated_complexity": "low | medium | high"
}
"""
# Build planning prompt
tools_available = context_state.get("tools_available", [])
prompt = f"""{EXECUTIVE_SYSTEM_PROMPT}
User request: {user_prompt}
Detected intent: {intent}
Available tools: {", ".join(tools_available) if tools_available else "None"}
Session context:
- Message count: {context_state.get('message_count', 0)}
- Time since last message: {context_state.get('minutes_since_last_msg', 0):.1f} minutes
- Active project: {context_state.get('active_project', 'None')}
Generate a structured execution plan.
"""
if VERBOSE_DEBUG:
logger.debug(f"[EXECUTIVE] Planning prompt:\n{prompt}")
# Call executive LLM
plan_text = await call_llm(
prompt,
backend=EXECUTIVE_LLM,
temperature=0.3, # Lower temperature for planning
max_tokens=500
)
if VERBOSE_DEBUG:
logger.debug(f"[EXECUTIVE] Generated plan:\n{plan_text}")
# Parse plan (simple heuristic extraction for Phase 1)
steps = []
tools_needed = []
for line in plan_text.split('\n'):
line_lower = line.lower()
if any(marker in line_lower for marker in ['step', '1.', '2.', '3.', '-']):
steps.append(line.strip())
if tools_available:
for tool in tools_available:
if tool.lower() in line_lower and tool not in tools_needed:
tools_needed.append(tool)
# Estimate complexity (simple heuristic)
complexity = "low"
if len(steps) > 3 or len(tools_needed) > 1:
complexity = "medium"
if len(steps) > 5 or "research" in intent.lower() or "analyze" in intent.lower():
complexity = "high"
return {
"summary": plan_text.split('\n')[0][:100] if plan_text else "Complex task execution plan",
"plan_text": plan_text,
"steps": steps[:10], # Limit to 10 steps
"tools_needed": tools_needed,
"estimated_complexity": complexity
}

1
cortex/autonomy/learning/__init__.py
View File

@@ -1 +0,0 @@
"""Pattern learning and adaptation system."""

383
cortex/autonomy/learning/pattern_learner.py
View File

@@ -1,383 +0,0 @@
"""
Pattern Learning System - learns from interaction patterns to improve autonomy.
"""
import logging
import json
import os
from typing import Dict, List, Any, Optional
from datetime import datetime
from collections import defaultdict
logger = logging.getLogger(__name__)
class PatternLearner:
"""
Learns from interaction patterns to improve Lyra's autonomous behavior.
Tracks:
- Topic frequencies (what users talk about)
- Time-of-day patterns (when users interact)
- User preferences (how users like responses)
- Successful response strategies (what works well)
"""
def __init__(self, patterns_file: str = "/app/data/learned_patterns.json"):
"""
Initialize pattern learner.
Args:
patterns_file: Path to persistent patterns storage
"""
self.patterns_file = patterns_file
self.patterns = self._load_patterns()
def _load_patterns(self) -> Dict[str, Any]:
"""Load patterns from disk."""
if os.path.exists(self.patterns_file):
try:
with open(self.patterns_file, 'r') as f:
patterns = json.load(f)
logger.info(f"[PATTERN_LEARNER] Loaded patterns from {self.patterns_file}")
return patterns
except Exception as e:
logger.error(f"[PATTERN_LEARNER] Failed to load patterns: {e}")
# Initialize empty patterns
return {
"topic_frequencies": {},
"time_patterns": {},
"user_preferences": {},
"successful_strategies": {},
"interaction_count": 0,
"last_updated": datetime.utcnow().isoformat()
}
def _save_patterns(self) -> None:
"""Save patterns to disk."""
try:
# Ensure directory exists
os.makedirs(os.path.dirname(self.patterns_file), exist_ok=True)
self.patterns["last_updated"] = datetime.utcnow().isoformat()
with open(self.patterns_file, 'w') as f:
json.dump(self.patterns, f, indent=2)
logger.debug(f"[PATTERN_LEARNER] Saved patterns to {self.patterns_file}")
except Exception as e:
logger.error(f"[PATTERN_LEARNER] Failed to save patterns: {e}")
async def learn_from_interaction(
self,
user_prompt: str,
response: str,
monologue: Dict[str, Any],
context: Dict[str, Any]
) -> None:
"""
Learn from a single interaction.
Args:
user_prompt: User's message
response: Lyra's response
monologue: Inner monologue analysis
context: Full context state
"""
self.patterns["interaction_count"] += 1
# Learn topic frequencies
self._learn_topics(user_prompt, monologue)
# Learn time patterns
self._learn_time_patterns()
# Learn user preferences
self._learn_preferences(monologue, context)
# Learn successful strategies
self._learn_strategies(monologue, response, context)
# Save periodically (every 10 interactions)
if self.patterns["interaction_count"] % 10 == 0:
self._save_patterns()
def _learn_topics(self, user_prompt: str, monologue: Dict[str, Any]) -> None:
"""Track topic frequencies."""
intent = monologue.get("intent", "unknown")
# Increment topic counter
topic_freq = self.patterns["topic_frequencies"]
topic_freq[intent] = topic_freq.get(intent, 0) + 1
# Extract keywords (simple approach - words > 5 chars)
keywords = [word.lower() for word in user_prompt.split() if len(word) > 5]
for keyword in keywords:
topic_freq[f"keyword:{keyword}"] = topic_freq.get(f"keyword:{keyword}", 0) + 1
logger.debug(f"[PATTERN_LEARNER] Topic learned: {intent}")
def _learn_time_patterns(self) -> None:
"""Track time-of-day patterns."""
now = datetime.utcnow()
hour = now.hour
# Track interactions by hour
time_patterns = self.patterns["time_patterns"]
hour_key = f"hour_{hour:02d}"
time_patterns[hour_key] = time_patterns.get(hour_key, 0) + 1
# Track day of week
day_key = f"day_{now.strftime('%A').lower()}"
time_patterns[day_key] = time_patterns.get(day_key, 0) + 1
def _learn_preferences(self, monologue: Dict[str, Any], context: Dict[str, Any]) -> None:
"""Learn user preferences from detected tone and depth."""
tone = monologue.get("tone", "neutral")
depth = monologue.get("depth", "medium")
prefs = self.patterns["user_preferences"]
# Track preferred tone
prefs.setdefault("tone_counts", {})
prefs["tone_counts"][tone] = prefs["tone_counts"].get(tone, 0) + 1
# Track preferred depth
prefs.setdefault("depth_counts", {})
prefs["depth_counts"][depth] = prefs["depth_counts"].get(depth, 0) + 1
def _learn_strategies(
self,
monologue: Dict[str, Any],
response: str,
context: Dict[str, Any]
) -> None:
"""
Learn which response strategies are successful.
Success indicators:
- Executive was consulted and plan generated
- Response length matches depth request
- Tone matches request
"""
intent = monologue.get("intent", "unknown")
executive_used = context.get("executive_plan") is not None
strategies = self.patterns["successful_strategies"]
strategies.setdefault(intent, {})
# Track executive usage for this intent
if executive_used:
key = f"{intent}:executive_used"
strategies.setdefault(key, 0)
strategies[key] += 1
# Track response length patterns
response_length = len(response.split())
depth = monologue.get("depth", "medium")
length_key = f"{depth}:avg_words"
if length_key not in strategies:
strategies[length_key] = response_length
else:
# Running average
strategies[length_key] = (strategies[length_key] + response_length) / 2
# ========================================
# Pattern Analysis and Recommendations
# ========================================
def get_top_topics(self, limit: int = 10) -> List[tuple]:
"""
Get most frequent topics.
Args:
limit: Max number of topics to return
Returns:
List of (topic, count) tuples, sorted by count
"""
topics = self.patterns["topic_frequencies"]
sorted_topics = sorted(topics.items(), key=lambda x: x[1], reverse=True)
return sorted_topics[:limit]
def get_preferred_tone(self) -> str:
"""
Get user's most preferred tone.
Returns:
Preferred tone string
"""
prefs = self.patterns["user_preferences"]
tone_counts = prefs.get("tone_counts", {})
if not tone_counts:
return "neutral"
return max(tone_counts.items(), key=lambda x: x[1])[0]
def get_preferred_depth(self) -> str:
"""
Get user's most preferred response depth.
Returns:
Preferred depth string
"""
prefs = self.patterns["user_preferences"]
depth_counts = prefs.get("depth_counts", {})
if not depth_counts:
return "medium"
return max(depth_counts.items(), key=lambda x: x[1])[0]
def get_peak_hours(self, limit: int = 3) -> List[int]:
"""
Get peak interaction hours.
Args:
limit: Number of top hours to return
Returns:
List of hours (0-23)
"""
time_patterns = self.patterns["time_patterns"]
hour_counts = {k: v for k, v in time_patterns.items() if k.startswith("hour_")}
if not hour_counts:
return []
sorted_hours = sorted(hour_counts.items(), key=lambda x: x[1], reverse=True)
top_hours = sorted_hours[:limit]
# Extract hour numbers
return [int(h[0].split("_")[1]) for h in top_hours]
def should_use_executive(self, intent: str) -> bool:
"""
Recommend whether to use executive for given intent based on patterns.
Args:
intent: Intent type
Returns:
True if executive is recommended
"""
strategies = self.patterns["successful_strategies"]
key = f"{intent}:executive_used"
# If we've used executive for this intent >= 3 times, recommend it
return strategies.get(key, 0) >= 3
def get_recommended_response_length(self, depth: str) -> int:
"""
Get recommended response length in words for given depth.
Args:
depth: Depth level (short/medium/deep)
Returns:
Recommended word count
"""
strategies = self.patterns["successful_strategies"]
key = f"{depth}:avg_words"
avg_length = strategies.get(key, None)
if avg_length:
return int(avg_length)
# Defaults if no pattern learned
defaults = {
"short": 50,
"medium": 150,
"deep": 300
}
return defaults.get(depth, 150)
def get_insights(self) -> Dict[str, Any]:
"""
Get high-level insights from learned patterns.
Returns:
{
"total_interactions": int,
"top_topics": [(topic, count), ...],
"preferred_tone": str,
"preferred_depth": str,
"peak_hours": [hours],
"learning_recommendations": [str]
}
"""
recommendations = []
# Check if user consistently prefers certain settings
preferred_tone = self.get_preferred_tone()
preferred_depth = self.get_preferred_depth()
if preferred_tone != "neutral":
recommendations.append(f"User prefers {preferred_tone} tone")
if preferred_depth != "medium":
recommendations.append(f"User prefers {preferred_depth} depth responses")
# Check for recurring topics
top_topics = self.get_top_topics(limit=3)
if top_topics:
top_topic = top_topics[0][0]
recommendations.append(f"Consider adding '{top_topic}' to learning queue")
return {
"total_interactions": self.patterns["interaction_count"],
"top_topics": self.get_top_topics(limit=5),
"preferred_tone": preferred_tone,
"preferred_depth": preferred_depth,
"peak_hours": self.get_peak_hours(limit=3),
"learning_recommendations": recommendations
}
def reset_patterns(self) -> None:
"""Reset all learned patterns (use with caution)."""
self.patterns = {
"topic_frequencies": {},
"time_patterns": {},
"user_preferences": {},
"successful_strategies": {},
"interaction_count": 0,
"last_updated": datetime.utcnow().isoformat()
}
self._save_patterns()
logger.warning("[PATTERN_LEARNER] Patterns reset")
def export_patterns(self) -> Dict[str, Any]:
"""
Export all patterns for analysis.
Returns:
Complete patterns dict
"""
return self.patterns.copy()
# Singleton instance
_learner_instance = None
def get_pattern_learner(patterns_file: str = "/app/data/learned_patterns.json") -> PatternLearner:
"""
Get singleton pattern learner instance.
Args:
patterns_file: Path to patterns file (only used on first call)
Returns:
PatternLearner instance
"""
global _learner_instance
if _learner_instance is None:
_learner_instance = PatternLearner(patterns_file=patterns_file)
return _learner_instance

1
cortex/autonomy/monologue/__init__.py
View File

@@ -1 +0,0 @@
# Inner monologue module

115
cortex/autonomy/monologue/monologue.py
View File

@@ -1,115 +0,0 @@
import os
import json
import logging
from typing import Dict
from llm.llm_router import call_llm
# Configuration
MONOLOGUE_LLM = os.getenv("MONOLOGUE_LLM", "PRIMARY").upper()
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"
# Logger
logger = logging.getLogger(__name__)
if VERBOSE_DEBUG:
logger.setLevel(logging.DEBUG)
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
'%(asctime)s [MONOLOGUE] %(levelname)s: %(message)s',
datefmt='%H:%M:%S'
))
logger.addHandler(console_handler)
MONOLOGUE_SYSTEM_PROMPT = """
You are Lyra's inner monologue.
You think privately.
You do NOT speak to the user.
You do NOT solve the task.
You only reflect on intent, tone, and depth.
Return ONLY valid JSON with:
- intent (string)
- tone (neutral | warm | focused | playful | direct)
- depth (short | medium | deep)
- consult_executive (true | false)
"""
class InnerMonologue:
async def process(self, context: Dict) -> Dict:
# Build full prompt with system instructions merged in
full_prompt = f"""{MONOLOGUE_SYSTEM_PROMPT}
User message:
{context['user_message']}
Self state:
{context['self_state']}
Context summary:
{context['context_summary']}
Output JSON only:
"""
# Call LLM using configured backend
if VERBOSE_DEBUG:
logger.debug(f"[InnerMonologue] Calling LLM with backend: {MONOLOGUE_LLM}")
logger.debug(f"[InnerMonologue] Prompt length: {len(full_prompt)} chars")
result = await call_llm(
full_prompt,
backend=MONOLOGUE_LLM,
temperature=0.7,
max_tokens=200
)
if VERBOSE_DEBUG:
logger.debug(f"[InnerMonologue] Raw LLM response:")
logger.debug(f"{'='*80}")
logger.debug(result)
logger.debug(f"{'='*80}")
logger.debug(f"[InnerMonologue] Response length: {len(result) if result else 0} chars")
# Parse JSON response - extract just the JSON part if there's extra text
try:
# Try direct parsing first
parsed = json.loads(result)
if VERBOSE_DEBUG:
logger.debug(f"[InnerMonologue] Successfully parsed JSON directly: {parsed}")
return parsed
except json.JSONDecodeError:
# If direct parsing fails, try to extract JSON from the response
if VERBOSE_DEBUG:
logger.debug(f"[InnerMonologue] Direct JSON parse failed, attempting extraction...")
# Look for JSON object (starts with { and ends with })
import re
json_match = re.search(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', result, re.DOTALL)
if json_match:
json_str = json_match.group(0)
try:
parsed = json.loads(json_str)
if VERBOSE_DEBUG:
logger.debug(f"[InnerMonologue] Successfully extracted and parsed JSON: {parsed}")
return parsed
except json.JSONDecodeError as e:
if VERBOSE_DEBUG:
logger.warning(f"[InnerMonologue] Extracted JSON still invalid: {e}")
else:
if VERBOSE_DEBUG:
logger.warning(f"[InnerMonologue] No JSON object found in response")
# Final fallback
if VERBOSE_DEBUG:
logger.warning(f"[InnerMonologue] All parsing attempts failed, using fallback")
else:
print(f"[InnerMonologue] JSON extraction failed")
print(f"[InnerMonologue] Raw response was: {result[:500]}")
return {
"intent": "unknown",
"tone": "neutral",
"depth": "medium",
"consult_executive": False
}

1
cortex/autonomy/proactive/__init__.py
View File

@@ -1 +0,0 @@
"""Proactive monitoring and suggestion system."""

321
cortex/autonomy/proactive/monitor.py
View File

@@ -1,321 +0,0 @@
"""
Proactive Context Monitor - detects opportunities for autonomous suggestions.
"""
import logging
import time
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
logger = logging.getLogger(__name__)
class ProactiveMonitor:
"""
Monitors conversation context and detects opportunities for proactive suggestions.
Triggers:
- Long silence → Check-in
- Learning queue + high curiosity → Suggest exploration
- Active goals → Progress reminders
- Conversation milestones → Offer summary
- Pattern detection → Helpful suggestions
"""
def __init__(self, min_priority: float = 0.6):
"""
Initialize proactive monitor.
Args:
min_priority: Minimum priority for suggestions (0.0-1.0)
"""
self.min_priority = min_priority
self.last_suggestion_time = {} # session_id -> timestamp
self.cooldown_seconds = 300 # 5 minutes between proactive suggestions
async def analyze_session(
self,
session_id: str,
context_state: Dict[str, Any],
self_state: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""
Analyze session for proactive suggestion opportunities.
Args:
session_id: Current session ID
context_state: Full context including message history
self_state: Lyra's current self-state
Returns:
{
"suggestion": "text to append to response",
"priority": 0.0-1.0,
"reason": "why this suggestion",
"type": "check_in | learning | goal_reminder | summary | pattern"
}
or None if no suggestion
"""
# Check cooldown
if not self._check_cooldown(session_id):
logger.debug(f"[PROACTIVE] Session {session_id} in cooldown, skipping")
return None
suggestions = []
# Check 1: Long silence detection
silence_suggestion = self._check_long_silence(context_state)
if silence_suggestion:
suggestions.append(silence_suggestion)
# Check 2: Learning queue + high curiosity
learning_suggestion = self._check_learning_opportunity(self_state)
if learning_suggestion:
suggestions.append(learning_suggestion)
# Check 3: Active goals reminder
goal_suggestion = self._check_active_goals(self_state, context_state)
if goal_suggestion:
suggestions.append(goal_suggestion)
# Check 4: Conversation milestones
milestone_suggestion = self._check_conversation_milestone(context_state)
if milestone_suggestion:
suggestions.append(milestone_suggestion)
# Check 5: Pattern-based suggestions
pattern_suggestion = self._check_patterns(context_state, self_state)
if pattern_suggestion:
suggestions.append(pattern_suggestion)
# Filter by priority and return highest
valid_suggestions = [s for s in suggestions if s["priority"] >= self.min_priority]
if not valid_suggestions:
return None
# Return highest priority suggestion
best_suggestion = max(valid_suggestions, key=lambda x: x["priority"])
# Update cooldown timer
self._update_cooldown(session_id)
logger.info(f"[PROACTIVE] Suggestion generated: {best_suggestion['type']} (priority: {best_suggestion['priority']:.2f})")
return best_suggestion
def _check_cooldown(self, session_id: str) -> bool:
"""Check if session is past cooldown period."""
if session_id not in self.last_suggestion_time:
return True
elapsed = time.time() - self.last_suggestion_time[session_id]
return elapsed >= self.cooldown_seconds
def _update_cooldown(self, session_id: str) -> None:
"""Update cooldown timer for session."""
self.last_suggestion_time[session_id] = time.time()
def _check_long_silence(self, context_state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Check if user has been silent for a long time.
"""
minutes_since_last = context_state.get("minutes_since_last_msg", 0)
# If > 30 minutes, suggest check-in
if minutes_since_last > 30:
return {
"suggestion": "\n\n[Aside: I'm still here if you need anything!]",
"priority": 0.7,
"reason": f"User silent for {minutes_since_last:.0f} minutes",
"type": "check_in"
}
return None
def _check_learning_opportunity(self, self_state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Check if Lyra has learning queue items and high curiosity.
"""
learning_queue = self_state.get("learning_queue", [])
curiosity = self_state.get("curiosity", 0.5)
# If curiosity > 0.7 and learning queue exists
if curiosity > 0.7 and learning_queue:
topic = learning_queue[0] if learning_queue else "new topics"
return {
"suggestion": f"\n\n[Aside: I've been curious about {topic} lately. Would you like to explore it together?]",
"priority": 0.65,
"reason": f"High curiosity ({curiosity:.2f}) and learning queue present",
"type": "learning"
}
return None
def _check_active_goals(
self,
self_state: Dict[str, Any],
context_state: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""
Check if there are active goals worth reminding about.
"""
active_goals = self_state.get("active_goals", [])
if not active_goals:
return None
# Check if we've had multiple messages without goal progress
message_count = context_state.get("message_count", 0)
# Every 10 messages, consider goal reminder
if message_count % 10 == 0 and message_count > 0:
goal = active_goals[0] # First active goal
goal_name = goal if isinstance(goal, str) else goal.get("name", "your goal")
return {
"suggestion": f"\n\n[Aside: Still thinking about {goal_name}. Let me know if you want to work on it.]",
"priority": 0.6,
"reason": f"Active goal present, {message_count} messages since start",
"type": "goal_reminder"
}
return None
def _check_conversation_milestone(self, context_state: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""
Check for conversation milestones (e.g., every 50 messages).
"""
message_count = context_state.get("message_count", 0)
# Every 50 messages, offer summary
if message_count > 0 and message_count % 50 == 0:
return {
"suggestion": f"\n\n[Aside: We've exchanged {message_count} messages! Would you like a summary of our conversation?]",
"priority": 0.65,
"reason": f"Milestone: {message_count} messages",
"type": "summary"
}
return None
def _check_patterns(
self,
context_state: Dict[str, Any],
self_state: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
"""
Check for behavioral patterns that merit suggestions.
"""
# Get current focus
focus = self_state.get("focus", "")
# Check if user keeps asking similar questions (detected via focus)
if focus and "repeated" in focus.lower():
return {
"suggestion": "\n\n[Aside: I notice we keep coming back to this topic. Would it help to create a summary or action plan?]",
"priority": 0.7,
"reason": "Repeated topic detected",
"type": "pattern"
}
# Check energy levels - if Lyra is low energy, maybe suggest break
energy = self_state.get("energy", 0.8)
if energy < 0.3:
return {
"suggestion": "\n\n[Aside: We've been at this for a while. Need a break or want to keep going?]",
"priority": 0.65,
"reason": f"Low energy ({energy:.2f})",
"type": "pattern"
}
return None
def format_suggestion(self, suggestion: Dict[str, Any]) -> str:
"""
Format suggestion for appending to response.
Args:
suggestion: Suggestion dict from analyze_session()
Returns:
Formatted string to append to response
"""
return suggestion.get("suggestion", "")
def set_cooldown_duration(self, seconds: int) -> None:
"""
Update cooldown duration.
Args:
seconds: New cooldown duration
"""
self.cooldown_seconds = seconds
logger.info(f"[PROACTIVE] Cooldown updated to {seconds}s")
def reset_cooldown(self, session_id: str) -> None:
"""
Reset cooldown for a specific session.
Args:
session_id: Session to reset
"""
if session_id in self.last_suggestion_time:
del self.last_suggestion_time[session_id]
logger.info(f"[PROACTIVE] Cooldown reset for session {session_id}")
def get_session_stats(self, session_id: str) -> Dict[str, Any]:
"""
Get stats for a session's proactive monitoring.
Args:
session_id: Session to check
Returns:
{
"last_suggestion_time": timestamp or None,
"seconds_since_last": int,
"cooldown_active": bool,
"cooldown_remaining": int
}
"""
last_time = self.last_suggestion_time.get(session_id)
if not last_time:
return {
"last_suggestion_time": None,
"seconds_since_last": 0,
"cooldown_active": False,
"cooldown_remaining": 0
}
seconds_since = int(time.time() - last_time)
cooldown_active = seconds_since < self.cooldown_seconds
cooldown_remaining = max(0, self.cooldown_seconds - seconds_since)
return {
"last_suggestion_time": last_time,
"seconds_since_last": seconds_since,
"cooldown_active": cooldown_active,
"cooldown_remaining": cooldown_remaining
}
# Singleton instance
_monitor_instance = None
def get_proactive_monitor(min_priority: float = 0.6) -> ProactiveMonitor:
"""
Get singleton proactive monitor instance.
Args:
min_priority: Minimum priority threshold (only used on first call)
Returns:
ProactiveMonitor instance
"""
global _monitor_instance
if _monitor_instance is None:
_monitor_instance = ProactiveMonitor(min_priority=min_priority)
return _monitor_instance

1
cortex/autonomy/self/__init__.py
View File

@@ -1 +0,0 @@
# Self state module

74
cortex/autonomy/self/analyzer.py
View File

@@ -1,74 +0,0 @@
"""
Analyze interactions and update self-state accordingly.
"""
import logging
from typing import Dict, Any
from .state import update_self_state
logger = logging.getLogger(__name__)
async def analyze_and_update_state(
monologue: Dict[str, Any],
user_prompt: str,
response: str,
context: Dict[str, Any]
) -> None:
"""
Analyze interaction and update self-state.
This runs after response generation to update Lyra's internal state
based on the interaction.
Args:
monologue: Inner monologue output
user_prompt: User's message
response: Lyra's response
context: Full context state
"""
# Simple heuristics for state updates
# TODO: Replace with LLM-based sentiment analysis in Phase 2
mood_delta = 0.0
energy_delta = 0.0
confidence_delta = 0.0
curiosity_delta = 0.0
new_focus = None
# Analyze intent from monologue
intent = monologue.get("intent", "").lower() if monologue else ""
if "technical" in intent or "complex" in intent:
energy_delta = -0.05 # Deep thinking is tiring
confidence_delta = 0.05 if len(response) > 200 else -0.05
new_focus = "technical_problem"
elif "creative" in intent or "brainstorm" in intent:
mood_delta = 0.1 # Creative work is engaging
curiosity_delta = 0.1
new_focus = "creative_exploration"
elif "clarification" in intent or "confused" in intent:
confidence_delta = -0.05
new_focus = "understanding_user"
elif "simple" in intent or "casual" in intent:
energy_delta = 0.05 # Light conversation is refreshing
new_focus = "conversation"
# Check for learning opportunities (questions in user prompt)
if "?" in user_prompt and any(word in user_prompt.lower() for word in ["how", "why", "what"]):
curiosity_delta += 0.05
# Update state
update_self_state(
mood_delta=mood_delta,
energy_delta=energy_delta,
new_focus=new_focus,
confidence_delta=confidence_delta,
curiosity_delta=curiosity_delta
)
logger.info(f"Self-state updated based on interaction: focus={new_focus}")

0
cortex/autonomy/self/self_state.json
View File

189
cortex/autonomy/self/state.py
View File

@@ -1,189 +0,0 @@
"""
Self-state management for Project Lyra.
Maintains persistent identity, mood, energy, and focus across sessions.
"""
import json
import logging
import os
from datetime import datetime
from pathlib import Path
from typing import Dict, Any, Optional
# Configuration
STATE_FILE = Path(os.getenv("SELF_STATE_FILE", "/app/data/self_state.json"))
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"
logger = logging.getLogger(__name__)
if VERBOSE_DEBUG:
logger.setLevel(logging.DEBUG)
# Default state structure
DEFAULT_STATE = {
"mood": "neutral",
"energy": 0.8,
"focus": "user_request",
"confidence": 0.7,
"curiosity": 0.5,
"last_updated": None,
"interaction_count": 0,
"learning_queue": [], # Topics Lyra wants to explore
"active_goals": [], # Self-directed goals
"preferences": {
"verbosity": "medium",
"formality": "casual",
"proactivity": 0.3 # How likely to suggest things unprompted
},
"metadata": {
"version": "1.0",
"created_at": None
}
}
class SelfState:
"""Manages Lyra's persistent self-state."""
def __init__(self):
self._state = self._load_state()
def _load_state(self) -> Dict[str, Any]:
"""Load state from disk or create default."""
if STATE_FILE.exists():
try:
with open(STATE_FILE, 'r') as f:
state = json.load(f)
logger.info(f"Loaded self-state from {STATE_FILE}")
return state
except Exception as e:
logger.error(f"Failed to load self-state: {e}")
return self._create_default_state()
else:
return self._create_default_state()
def _create_default_state(self) -> Dict[str, Any]:
"""Create and save default state."""
state = DEFAULT_STATE.copy()
state["metadata"]["created_at"] = datetime.now().isoformat()
state["last_updated"] = datetime.now().isoformat()
self._save_state(state)
logger.info("Created new default self-state")
return state
def _save_state(self, state: Dict[str, Any]) -> None:
"""Persist state to disk."""
try:
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
with open(STATE_FILE, 'w') as f:
json.dump(state, f, indent=2)
if VERBOSE_DEBUG:
logger.debug(f"Saved self-state to {STATE_FILE}")
except Exception as e:
logger.error(f"Failed to save self-state: {e}")
def get_state(self) -> Dict[str, Any]:
"""Get current state snapshot."""
return self._state.copy()
def update_from_interaction(
self,
mood_delta: float = 0.0,
energy_delta: float = 0.0,
new_focus: Optional[str] = None,
confidence_delta: float = 0.0,
curiosity_delta: float = 0.0
) -> None:
"""
Update state based on interaction.
Args:
mood_delta: Change in mood (-1.0 to 1.0)
energy_delta: Change in energy (-1.0 to 1.0)
new_focus: New focus area
confidence_delta: Change in confidence
curiosity_delta: Change in curiosity
"""
# Apply deltas with bounds checking
self._state["energy"] = max(0.0, min(1.0,
self._state.get("energy", 0.8) + energy_delta))
self._state["confidence"] = max(0.0, min(1.0,
self._state.get("confidence", 0.7) + confidence_delta))
self._state["curiosity"] = max(0.0, min(1.0,
self._state.get("curiosity", 0.5) + curiosity_delta))
# Update focus if provided
if new_focus:
self._state["focus"] = new_focus
# Update mood (simplified sentiment)
if mood_delta != 0:
mood_map = ["frustrated", "neutral", "engaged", "excited"]
current_mood_idx = 1 # neutral default
if self._state.get("mood") in mood_map:
current_mood_idx = mood_map.index(self._state["mood"])
new_mood_idx = max(0, min(len(mood_map) - 1,
int(current_mood_idx + mood_delta * 2)))
self._state["mood"] = mood_map[new_mood_idx]
# Increment interaction counter
self._state["interaction_count"] = self._state.get("interaction_count", 0) + 1
self._state["last_updated"] = datetime.now().isoformat()
# Persist changes
self._save_state(self._state)
if VERBOSE_DEBUG:
logger.debug(f"Updated self-state: mood={self._state['mood']}, "
f"energy={self._state['energy']:.2f}, "
f"confidence={self._state['confidence']:.2f}")
def add_learning_goal(self, topic: str) -> None:
"""Add topic to learning queue."""
queue = self._state.get("learning_queue", [])
if topic not in [item.get("topic") for item in queue]:
queue.append({
"topic": topic,
"added_at": datetime.now().isoformat(),
"priority": 0.5
})
self._state["learning_queue"] = queue
self._save_state(self._state)
logger.info(f"Added learning goal: {topic}")
def add_active_goal(self, goal: str, context: str = "") -> None:
"""Add self-directed goal."""
goals = self._state.get("active_goals", [])
goals.append({
"goal": goal,
"context": context,
"created_at": datetime.now().isoformat(),
"status": "active"
})
self._state["active_goals"] = goals
self._save_state(self._state)
logger.info(f"Added active goal: {goal}")
# Global instance
_self_state_instance = None
def get_self_state_instance() -> SelfState:
"""Get or create global SelfState instance."""
global _self_state_instance
if _self_state_instance is None:
_self_state_instance = SelfState()
return _self_state_instance
def load_self_state() -> Dict[str, Any]:
"""Load self state - public API for backwards compatibility."""
return get_self_state_instance().get_state()
def update_self_state(**kwargs) -> None:
"""Update self state - public API."""
get_self_state_instance().update_from_interaction(**kwargs)

1
cortex/autonomy/tools/__init__.py
View File

@@ -1 +0,0 @@
"""Autonomous tool invocation system."""

13
cortex/autonomy/tools/adapters/__init__.py
View File

@@ -1,13 +0,0 @@
"""Provider adapters for tool calling."""
from .base import ToolAdapter
from .openai_adapter import OpenAIAdapter
from .ollama_adapter import OllamaAdapter
from .llamacpp_adapter import LlamaCppAdapter
__all__ = [
"ToolAdapter",
"OpenAIAdapter",
"OllamaAdapter",
"LlamaCppAdapter",
]

79
cortex/autonomy/tools/adapters/base.py
View File

@@ -1,79 +0,0 @@
"""
Base adapter interface for provider-agnostic tool calling.
This module defines the abstract base class that all LLM provider adapters
must implement to support tool calling in Lyra.
"""
from abc import ABC, abstractmethod
from typing import Dict, List, Optional
class ToolAdapter(ABC):
"""Base class for provider-specific tool adapters.
Each LLM provider (OpenAI, Ollama, llama.cpp, etc.) has its own
way of handling tool calls. This adapter pattern allows Lyra to
support tools across all providers with a unified interface.
"""
@abstractmethod
async def prepare_request(
self,
messages: List[Dict],
tools: List[Dict],
tool_choice: Optional[str] = None
) -> Dict:
"""Convert Lyra tool definitions to provider-specific format.
Args:
messages: Conversation history in OpenAI format
tools: List of Lyra tool definitions (provider-agnostic)
tool_choice: Optional tool forcing ("auto", "required", "none")
Returns:
dict: Provider-specific request payload ready to send to LLM
"""
pass
@abstractmethod
async def parse_response(self, response) -> Dict:
"""Extract tool calls from provider response.
Args:
response: Raw provider response (format varies by provider)
Returns:
dict: Standardized response in Lyra format:
{
"content": str, # Assistant's text response
"tool_calls": [ # List of tool calls or None
{
"id": str, # Unique call ID
"name": str, # Tool name
"arguments": dict # Tool arguments
}
] or None
}
"""
pass
@abstractmethod
def format_tool_result(
self,
tool_call_id: str,
tool_name: str,
result: Dict
) -> Dict:
"""Format tool execution result for next LLM call.
Args:
tool_call_id: ID from the original tool call
tool_name: Name of the executed tool
result: Tool execution result dictionary
Returns:
dict: Message object to append to conversation
(format varies by provider)
"""
pass

17
cortex/autonomy/tools/adapters/llamacpp_adapter.py
View File

@@ -1,17 +0,0 @@
"""
llama.cpp adapter for tool calling.
Since llama.cpp has similar constraints to Ollama (no native function calling),
this adapter reuses the XML-based approach from OllamaAdapter.
"""
from .ollama_adapter import OllamaAdapter
class LlamaCppAdapter(OllamaAdapter):
"""llama.cpp adapter - uses same XML approach as Ollama.
llama.cpp doesn't have native function calling support, so we use
the same XML-based prompt engineering approach as Ollama.
"""
pass

191
cortex/autonomy/tools/adapters/ollama_adapter.py
View File

@@ -1,191 +0,0 @@
"""
Ollama adapter for tool calling using XML-structured prompts.
Since Ollama doesn't have native function calling, this adapter uses
XML-based prompts to instruct the model how to call tools.
"""
import json
import re
from typing import Dict, List, Optional
from .base import ToolAdapter
class OllamaAdapter(ToolAdapter):
"""Ollama adapter using XML-structured prompts for tool calling.
This adapter injects tool descriptions into the system prompt and
teaches the model to respond with XML when it wants to use a tool.
"""
SYSTEM_PROMPT = """You have access to the following tools:
{tool_descriptions}
To use a tool, respond with XML in this exact format:
<tool_call>
<name>tool_name</name>
<arguments>
<arg_name>value</arg_name>
</arguments>
<reason>why you're using this tool</reason>
</tool_call>
You can call multiple tools by including multiple <tool_call> blocks.
If you don't need to use any tools, respond normally without XML.
After tools are executed, you'll receive results and can continue the conversation."""
async def prepare_request(
self,
messages: List[Dict],
tools: List[Dict],
tool_choice: Optional[str] = None
) -> Dict:
"""Inject tool descriptions into system prompt.
Args:
messages: Conversation history
tools: Lyra tool definitions
tool_choice: Ignored for Ollama (no native support)
Returns:
dict: Request payload with modified messages
"""
# Format tool descriptions
tool_desc = "\n".join([
f"- {t['name']}: {t['description']}\n Parameters: {self._format_parameters(t['parameters'], t.get('required', []))}"
for t in tools
])
system_msg = self.SYSTEM_PROMPT.format(tool_descriptions=tool_desc)
# Check if first message is already a system message
modified_messages = messages.copy()
if modified_messages and modified_messages[0].get("role") == "system":
# Prepend tool instructions to existing system message
modified_messages[0]["content"] = system_msg + "\n\n" + modified_messages[0]["content"]
else:
# Add new system message at the beginning
modified_messages.insert(0, {"role": "system", "content": system_msg})
return {"messages": modified_messages}
def _format_parameters(self, parameters: Dict, required: List[str]) -> str:
"""Format parameters for tool description.
Args:
parameters: Parameter definitions
required: List of required parameter names
Returns:
str: Human-readable parameter description
"""
param_strs = []
for name, spec in parameters.items():
req_marker = "(required)" if name in required else "(optional)"
param_strs.append(f"{name} {req_marker}: {spec.get('description', '')}")
return ", ".join(param_strs)
async def parse_response(self, response) -> Dict:
"""Extract tool calls from XML in response.
Args:
response: String response from Ollama
Returns:
dict: Standardized Lyra format with content and tool_calls
"""
import logging
logger = logging.getLogger(__name__)
# Ollama returns a string
if isinstance(response, dict):
content = response.get("message", {}).get("content", "")
else:
content = str(response)
logger.info(f"🔍 OllamaAdapter.parse_response: content length={len(content)}, has <tool_call>={('<tool_call>' in content)}")
logger.debug(f"🔍 Content preview: {content[:500]}")
# Parse XML tool calls
tool_calls = []
if "<tool_call>" in content:
# Split content by <tool_call> to get each block
blocks = content.split('<tool_call>')
logger.info(f"🔍 Split into {len(blocks)} blocks")
# First block is content before any tool calls
clean_parts = [blocks[0]]
for idx, block in enumerate(blocks[1:]): # Skip first block (pre-tool content)
# Extract tool name
name_match = re.search(r'<name>(.*?)</name>', block)
if not name_match:
logger.warning(f"Block {idx} has no <name> tag, skipping")
continue
name = name_match.group(1).strip()
arguments = {}
# Extract arguments
args_match = re.search(r'<arguments>(.*?)</arguments>', block, re.DOTALL)
if args_match:
args_xml = args_match.group(1)
# Parse <key>value</key> pairs
arg_pairs = re.findall(r'<(\w+)>(.*?)</\1>', args_xml, re.DOTALL)
arguments = {k: v.strip() for k, v in arg_pairs}
tool_calls.append({
"id": f"call_{idx}",
"name": name,
"arguments": arguments
})
# For clean content, find what comes AFTER the tool call block
# Look for the last closing tag (</tool_call> or malformed </xxx>) and keep what's after
# Split by any closing tag at the END of the tool block
remaining = block
# Remove everything up to and including a standalone closing tag
# Pattern: find </something> that's not followed by more XML
end_match = re.search(r'</[a-z_]+>\s*(.*)$', remaining, re.DOTALL)
if end_match:
after_content = end_match.group(1).strip()
if after_content and not after_content.startswith('<'):
# Only keep if it's actual text content, not more XML
clean_parts.append(after_content)
clean_content = ''.join(clean_parts).strip()
else:
clean_content = content
return {
"content": clean_content,
"tool_calls": tool_calls if tool_calls else None
}
def format_tool_result(
self,
tool_call_id: str,
tool_name: str,
result: Dict
) -> Dict:
"""Format tool result as XML for next prompt.
Args:
tool_call_id: ID from the original tool call
tool_name: Name of the executed tool
result: Tool execution result
Returns:
dict: Message in user role with XML-formatted result
"""
# Format result as XML
result_xml = f"""<tool_result>
<tool>{tool_name}</tool>
<result>{json.dumps(result, ensure_ascii=False)}</result>
</tool_result>"""
return {
"role": "user",
"content": result_xml
}

130
cortex/autonomy/tools/adapters/openai_adapter.py
View File

@@ -1,130 +0,0 @@
"""
OpenAI adapter for tool calling using native function calling API.
This adapter converts Lyra tool definitions to OpenAI's function calling
format and parses OpenAI responses back to Lyra's standardized format.
"""
import json
from typing import Dict, List, Optional
from .base import ToolAdapter
class OpenAIAdapter(ToolAdapter):
"""OpenAI-specific adapter using native function calling.
OpenAI supports function calling natively through the 'tools' parameter
in chat completions. This adapter leverages that capability.
"""
async def prepare_request(
self,
messages: List[Dict],
tools: List[Dict],
tool_choice: Optional[str] = None
) -> Dict:
"""Convert Lyra tools to OpenAI function calling format.
Args:
messages: Conversation history
tools: Lyra tool definitions
tool_choice: "auto", "required", "none", or None
Returns:
dict: Request payload with OpenAI-formatted tools
"""
# Convert Lyra tools → OpenAI function calling format
openai_tools = []
for tool in tools:
openai_tools.append({
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": {
"type": "object",
"properties": tool["parameters"],
"required": tool.get("required", [])
}
}
})
payload = {
"messages": messages,
"tools": openai_tools
}
# Add tool_choice if specified
if tool_choice:
if tool_choice == "required":
payload["tool_choice"] = "required"
elif tool_choice == "none":
payload["tool_choice"] = "none"
else: # "auto" or default
payload["tool_choice"] = "auto"
return payload
async def parse_response(self, response) -> Dict:
"""Extract tool calls from OpenAI response.
Args:
response: OpenAI ChatCompletion response object
Returns:
dict: Standardized Lyra format with content and tool_calls
"""
message = response.choices[0].message
content = message.content if message.content else ""
tool_calls = []
# Check if response contains tool calls
if hasattr(message, 'tool_calls') and message.tool_calls:
for tc in message.tool_calls:
try:
# Parse arguments (may be JSON string)
args = tc.function.arguments
if isinstance(args, str):
args = json.loads(args)
tool_calls.append({
"id": tc.id,
"name": tc.function.name,
"arguments": args
})
except json.JSONDecodeError as e:
# If arguments can't be parsed, include error
tool_calls.append({
"id": tc.id,
"name": tc.function.name,
"arguments": {},
"error": f"Failed to parse arguments: {str(e)}"
})
return {
"content": content,
"tool_calls": tool_calls if tool_calls else None
}
def format_tool_result(
self,
tool_call_id: str,
tool_name: str,
result: Dict
) -> Dict:
"""Format tool result as OpenAI tool message.
Args:
tool_call_id: ID from the original tool call
tool_name: Name of the executed tool
result: Tool execution result
Returns:
dict: Message in OpenAI tool message format
"""
return {
"role": "tool",
"tool_call_id": tool_call_id,
"name": tool_name,
"content": json.dumps(result, ensure_ascii=False)
}

124
cortex/autonomy/tools/decision_engine.py
View File

@@ -1,124 +0,0 @@
"""
Tool Decision Engine - decides which tools to invoke autonomously.
"""
import logging
from typing import Dict, List, Any
logger = logging.getLogger(__name__)
class ToolDecisionEngine:
"""Decides which tools to invoke based on context analysis."""
async def analyze_tool_needs(
self,
user_prompt: str,
monologue: Dict[str, Any],
context_state: Dict[str, Any],
available_tools: List[str]
) -> Dict[str, Any]:
"""
Analyze if tools should be invoked and which ones.
Args:
user_prompt: User's message
monologue: Inner monologue analysis
context_state: Full context
available_tools: List of available tools
Returns:
{
"should_invoke_tools": bool,
"tools_to_invoke": [
{
"tool": "RAG | WEB | WEATHER | etc",
"query": "search query",
"reason": "why this tool",
"priority": 0.0-1.0
},
...
],
"confidence": 0.0-1.0
}
"""
tools_to_invoke = []
# Check for memory/context needs
if any(word in user_prompt.lower() for word in [
"remember", "you said", "we discussed", "earlier", "before",
"last time", "previously", "what did"
]):
tools_to_invoke.append({
"tool": "RAG",
"query": user_prompt,
"reason": "User references past conversation",
"priority": 0.9
})
# Check for web search needs
if any(word in user_prompt.lower() for word in [
"current", "latest", "news", "today", "what's happening",
"look up", "search for", "find information", "recent"
]):
tools_to_invoke.append({
"tool": "WEB",
"query": user_prompt,
"reason": "Requires current information",
"priority": 0.8
})
# Check for weather needs
if any(word in user_prompt.lower() for word in [
"weather", "temperature", "forecast", "rain", "sunny", "climate"
]):
tools_to_invoke.append({
"tool": "WEATHER",
"query": user_prompt,
"reason": "Weather information requested",
"priority": 0.95
})
# Check for code-related needs
if any(word in user_prompt.lower() for word in [
"code", "function", "debug", "implement", "algorithm",
"programming", "script", "syntax"
]):
if "CODEBRAIN" in available_tools:
tools_to_invoke.append({
"tool": "CODEBRAIN",
"query": user_prompt,
"reason": "Code-related task",
"priority": 0.85
})
# Proactive RAG for complex queries (based on monologue)
intent = monologue.get("intent", "") if monologue else ""
if monologue and monologue.get("consult_executive"):
# Complex query - might benefit from context
if not any(t["tool"] == "RAG" for t in tools_to_invoke):
tools_to_invoke.append({
"tool": "RAG",
"query": user_prompt,
"reason": "Complex query benefits from context",
"priority": 0.6
})
# Sort by priority
tools_to_invoke.sort(key=lambda x: x["priority"], reverse=True)
max_priority = max([t["priority"] for t in tools_to_invoke]) if tools_to_invoke else 0.0
result = {
"should_invoke_tools": len(tools_to_invoke) > 0,
"tools_to_invoke": tools_to_invoke,
"confidence": max_priority
}
if tools_to_invoke:
logger.info(f"[TOOL_DECISION] Autonomous tool invocation recommended: {len(tools_to_invoke)} tools")
for tool in tools_to_invoke:
logger.info(f" - {tool['tool']} (priority: {tool['priority']:.2f}): {tool['reason']}")
return result

12
cortex/autonomy/tools/executors/__init__.py
View File

@@ -1,12 +0,0 @@
"""Tool executors for Lyra."""
from .code_executor import execute_code
from .web_search import search_web
from .trilium import search_notes, create_note
__all__ = [
"execute_code",
"search_web",
"search_notes",
"create_note",
]

218
cortex/autonomy/tools/executors/code_executor.py
View File

@@ -1,218 +0,0 @@
"""
Code executor for running Python and bash code in a sandbox container.
This module provides secure code execution with timeout protection,
output limits, and forbidden pattern detection.
"""
import asyncio
import os
import tempfile
import re
from typing import Dict
import docker
from docker.errors import (
DockerException,
APIError,
ContainerError,
ImageNotFound,
NotFound
)
# Forbidden patterns that pose security risks
FORBIDDEN_PATTERNS = [
r'rm\s+-rf', # Destructive file removal
r':\(\)\{\s*:\|:&\s*\};:', # Fork bomb
r'mkfs', # Filesystem formatting
r'/dev/sd[a-z]', # Direct device access
r'dd\s+if=', # Low-level disk operations
r'>\s*/dev/sd', # Writing to devices
r'curl.*\|.*sh', # Pipe to shell (common attack vector)
r'wget.*\|.*sh', # Pipe to shell
]
async def execute_code(args: Dict) -> Dict:
"""Execute code in sandbox container.
Args:
args: Dictionary containing:
- language (str): "python" or "bash"
- code (str): The code to execute
- reason (str): Why this code is being executed
- timeout (int, optional): Execution timeout in seconds
Returns:
dict: Execution result containing:
- stdout (str): Standard output
- stderr (str): Standard error
- exit_code (int): Process exit code
- execution_time (float): Time taken in seconds
OR
- error (str): Error message if execution failed
"""
language = args.get("language")
code = args.get("code")
reason = args.get("reason", "No reason provided")
timeout = args.get("timeout", 30)
# Validation
if not language or language not in ["python", "bash"]:
return {"error": "Invalid language. Must be 'python' or 'bash'"}
if not code:
return {"error": "No code provided"}
# Security: Check for forbidden patterns
for pattern in FORBIDDEN_PATTERNS:
if re.search(pattern, code, re.IGNORECASE):
return {"error": f"Forbidden pattern detected for security reasons"}
# Validate and cap timeout
max_timeout = int(os.getenv("CODE_SANDBOX_MAX_TIMEOUT", "120"))
timeout = min(max(timeout, 1), max_timeout)
container = os.getenv("CODE_SANDBOX_CONTAINER", "lyra-code-sandbox")
# Validate container exists and is running
try:
docker_client = docker.from_env()
container_obj = docker_client.containers.get(container)
if container_obj.status != "running":
return {
"error": f"Sandbox container '{container}' is not running (status: {container_obj.status})",
"hint": "Start the container with: docker start " + container
}
except NotFound:
return {
"error": f"Sandbox container '{container}' not found",
"hint": "Ensure the container exists and is running"
}
except DockerException as e:
return {
"error": f"Docker daemon error: {str(e)}",
"hint": "Check Docker connectivity and permissions"
}
# Write code to temporary file
suffix = ".py" if language == "python" else ".sh"
try:
with tempfile.NamedTemporaryFile(
mode='w',
suffix=suffix,
delete=False,
encoding='utf-8'
) as f:
f.write(code)
temp_file = f.name
except Exception as e:
return {"error": f"Failed to create temp file: {str(e)}"}
try:
# Copy file to container
exec_path = f"/executions/{os.path.basename(temp_file)}"
cp_proc = await asyncio.create_subprocess_exec(
"docker", "cp", temp_file, f"{container}:{exec_path}",
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
await cp_proc.communicate()
if cp_proc.returncode != 0:
return {"error": "Failed to copy code to sandbox container"}
# Fix permissions so sandbox user can read the file (run as root)
chown_proc = await asyncio.create_subprocess_exec(
"docker", "exec", "-u", "root", container, "chown", "sandbox:sandbox", exec_path,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
await chown_proc.communicate()
# Execute in container as sandbox user
if language == "python":
cmd = ["docker", "exec", "-u", "sandbox", container, "python3", exec_path]
else: # bash
cmd = ["docker", "exec", "-u", "sandbox", container, "bash", exec_path]
start_time = asyncio.get_event_loop().time()
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
try:
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout
)
execution_time = asyncio.get_event_loop().time() - start_time
# Truncate output to prevent memory issues (configurable)
max_output = int(os.getenv("CODE_SANDBOX_MAX_OUTPUT", "10240")) # 10KB default
stdout_str = stdout[:max_output].decode('utf-8', errors='replace')
stderr_str = stderr[:max_output].decode('utf-8', errors='replace')
if len(stdout) > max_output:
stdout_str += f"\n... (output truncated, {len(stdout)} bytes total)"
if len(stderr) > max_output:
stderr_str += f"\n... (output truncated, {len(stderr)} bytes total)"
return {
"stdout": stdout_str,
"stderr": stderr_str,
"exit_code": proc.returncode,
"execution_time": round(execution_time, 2)
}
except asyncio.TimeoutError:
# Kill the process
try:
proc.kill()
await proc.wait()
except:
pass
return {"error": f"Execution timeout after {timeout}s"}
except APIError as e:
return {
"error": f"Docker API error: {e.explanation}",
"status_code": e.status_code
}
except ContainerError as e:
return {
"error": f"Container execution error: {str(e)}",
"exit_code": e.exit_status
}
except DockerException as e:
return {
"error": f"Docker error: {str(e)}",
"hint": "Check Docker daemon connectivity and permissions"
}
except Exception as e:
return {"error": f"Execution failed: {str(e)}"}
finally:
# Cleanup temporary file
try:
if 'temp_file' in locals():
os.unlink(temp_file)
except Exception as cleanup_error:
# Log but don't fail on cleanup errors
pass
# Optional: Clean up file from container (best effort)
try:
if 'exec_path' in locals() and 'container_obj' in locals():
container_obj.exec_run(
f"rm -f {exec_path}",
user="sandbox"
)
except:
pass # Best effort cleanup

13
cortex/autonomy/tools/executors/search_providers/__init__.py
View File

@@ -1,13 +0,0 @@
"""Web search provider implementations."""
from .base import SearchProvider, SearchResult, SearchResponse
from .brave import BraveSearchProvider
from .duckduckgo import DuckDuckGoProvider
__all__ = [
"SearchProvider",
"SearchResult",
"SearchResponse",
"BraveSearchProvider",
"DuckDuckGoProvider",
]

49
cortex/autonomy/tools/executors/search_providers/base.py
View File

@@ -1,49 +0,0 @@
"""Base interface for web search providers."""
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class SearchResult:
"""Standardized search result format."""
title: str
url: str
snippet: str
score: Optional[float] = None
@dataclass
class SearchResponse:
"""Standardized search response."""
results: List[SearchResult]
count: int
provider: str
query: str
error: Optional[str] = None
class SearchProvider(ABC):
"""Abstract base class for search providers."""
@abstractmethod
async def search(
self,
query: str,
max_results: int = 5,
**kwargs
) -> SearchResponse:
"""Execute search and return standardized results."""
pass
@abstractmethod
async def health_check(self) -> bool:
"""Check if provider is healthy and reachable."""
pass
@property
@abstractmethod
def name(self) -> str:
"""Provider name."""
pass

123
cortex/autonomy/tools/executors/search_providers/brave.py
View File

@@ -1,123 +0,0 @@
"""Brave Search API provider implementation."""
import os
import asyncio
import aiohttp
from .base import SearchProvider, SearchResponse, SearchResult
from ..utils.resilience import async_retry
class BraveSearchProvider(SearchProvider):
"""Brave Search API implementation."""
def __init__(self):
self.api_key = os.getenv("BRAVE_SEARCH_API_KEY", "")
self.base_url = os.getenv(
"BRAVE_SEARCH_URL",
"https://api.search.brave.com/res/v1"
)
self.timeout = float(os.getenv("BRAVE_SEARCH_TIMEOUT", "10.0"))
@property
def name(self) -> str:
return "brave"
@async_retry(
max_attempts=3,
exceptions=(aiohttp.ClientError, asyncio.TimeoutError)
)
async def search(
self,
query: str,
max_results: int = 5,
**kwargs
) -> SearchResponse:
"""Execute Brave search with retry logic."""
if not self.api_key:
return SearchResponse(
results=[],
count=0,
provider=self.name,
query=query,
error="BRAVE_SEARCH_API_KEY not configured"
)
headers = {
"Accept": "application/json",
"X-Subscription-Token": self.api_key
}
params = {
"q": query,
"count": min(max_results, 20) # Brave max is 20
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/web/search",
headers=headers,
params=params,
timeout=aiohttp.ClientTimeout(total=self.timeout)
) as resp:
if resp.status == 200:
data = await resp.json()
results = []
for item in data.get("web", {}).get("results", []):
results.append(SearchResult(
title=item.get("title", ""),
url=item.get("url", ""),
snippet=item.get("description", ""),
score=item.get("score")
))
return SearchResponse(
results=results,
count=len(results),
provider=self.name,
query=query
)
elif resp.status == 401:
error = "Authentication failed. Check BRAVE_SEARCH_API_KEY"
elif resp.status == 429:
error = f"Rate limit exceeded. Status: {resp.status}"
else:
error_text = await resp.text()
error = f"HTTP {resp.status}: {error_text}"
return SearchResponse(
results=[],
count=0,
provider=self.name,
query=query,
error=error
)
except aiohttp.ClientConnectorError as e:
return SearchResponse(
results=[],
count=0,
provider=self.name,
query=query,
error=f"Cannot connect to Brave Search API: {str(e)}"
)
except asyncio.TimeoutError:
return SearchResponse(
results=[],
count=0,
provider=self.name,
query=query,
error=f"Search timeout after {self.timeout}s"
)
async def health_check(self) -> bool:
"""Check if Brave API is reachable."""
if not self.api_key:
return False
try:
response = await self.search("test", max_results=1)
return response.error is None
except:
return False

60
cortex/autonomy/tools/executors/search_providers/duckduckgo.py
View File

@@ -1,60 +0,0 @@
"""DuckDuckGo search provider with retry logic (legacy fallback)."""
from duckduckgo_search import DDGS
from .base import SearchProvider, SearchResponse, SearchResult
from ..utils.resilience import async_retry
class DuckDuckGoProvider(SearchProvider):
"""DuckDuckGo search implementation with retry logic."""
@property
def name(self) -> str:
return "duckduckgo"
@async_retry(
max_attempts=3,
exceptions=(Exception,) # DDG throws generic exceptions
)
async def search(
self,
query: str,
max_results: int = 5,
**kwargs
) -> SearchResponse:
"""Execute DuckDuckGo search with retry logic."""
try:
with DDGS() as ddgs:
results = []
for result in ddgs.text(query, max_results=max_results):
results.append(SearchResult(
title=result.get("title", ""),
url=result.get("href", ""),
snippet=result.get("body", "")
))
return SearchResponse(
results=results,
count=len(results),
provider=self.name,
query=query
)
except Exception as e:
return SearchResponse(
results=[],
count=0,
provider=self.name,
query=query,
error=f"Search failed: {str(e)}"
)
async def health_check(self) -> bool:
"""Basic health check for DDG."""
try:
response = await self.search("test", max_results=1)
return response.error is None
except:
return False

216
cortex/autonomy/tools/executors/trilium.py
View File

@@ -1,216 +0,0 @@
"""
Trilium notes executor for searching and creating notes via ETAPI.
This module provides integration with Trilium notes through the ETAPI HTTP API
with improved resilience: timeout configuration, retry logic, and connection pooling.
"""
import os
import asyncio
import aiohttp
from typing import Dict, Optional
from ..utils.resilience import async_retry
TRILIUM_URL = os.getenv("TRILIUM_URL", "http://localhost:8080")
TRILIUM_TOKEN = os.getenv("TRILIUM_ETAPI_TOKEN", "")
# Module-level session for connection pooling
_session: Optional[aiohttp.ClientSession] = None
def get_session() -> aiohttp.ClientSession:
"""Get or create shared aiohttp session for connection pooling."""
global _session
if _session is None or _session.closed:
timeout = aiohttp.ClientTimeout(
total=float(os.getenv("TRILIUM_TIMEOUT", "30.0")),
connect=float(os.getenv("TRILIUM_CONNECT_TIMEOUT", "10.0"))
)
_session = aiohttp.ClientSession(timeout=timeout)
return _session
@async_retry(
max_attempts=3,
exceptions=(aiohttp.ClientError, asyncio.TimeoutError)
)
async def search_notes(args: Dict) -> Dict:
"""Search Trilium notes via ETAPI with retry logic.
Args:
args: Dictionary containing:
- query (str): Search query
- limit (int, optional): Maximum notes to return (default: 5, max: 20)
Returns:
dict: Search results containing:
- notes (list): List of notes with noteId, title, content, type
- count (int): Number of notes returned
OR
- error (str): Error message if search failed
"""
query = args.get("query")
limit = args.get("limit", 5)
# Validation
if not query:
return {"error": "No query provided"}
if not TRILIUM_TOKEN:
return {
"error": "TRILIUM_ETAPI_TOKEN not configured in environment",
"hint": "Set TRILIUM_ETAPI_TOKEN in .env file"
}
# Cap limit
limit = min(max(limit, 1), 20)
try:
session = get_session()
async with session.get(
f"{TRILIUM_URL}/etapi/notes",
params={"search": query, "limit": limit},
headers={"Authorization": TRILIUM_TOKEN}
) as resp:
if resp.status == 200:
data = await resp.json()
# ETAPI returns {"results": [...]} format
results = data.get("results", [])
return {
"notes": results,
"count": len(results)
}
elif resp.status == 401:
return {
"error": "Authentication failed. Check TRILIUM_ETAPI_TOKEN",
"status": 401
}
elif resp.status == 404:
return {
"error": "Trilium API endpoint not found. Check TRILIUM_URL",
"status": 404,
"url": TRILIUM_URL
}
else:
error_text = await resp.text()
return {
"error": f"HTTP {resp.status}: {error_text}",
"status": resp.status
}
except aiohttp.ClientConnectorError as e:
return {
"error": f"Cannot connect to Trilium at {TRILIUM_URL}",
"hint": "Check if Trilium is running and URL is correct",
"details": str(e)
}
except asyncio.TimeoutError:
timeout = os.getenv("TRILIUM_TIMEOUT", "30.0")
return {
"error": f"Trilium request timeout after {timeout}s",
"hint": "Trilium may be slow or unresponsive"
}
except Exception as e:
return {
"error": f"Search failed: {str(e)}",
"type": type(e).__name__
}
@async_retry(
max_attempts=3,
exceptions=(aiohttp.ClientError, asyncio.TimeoutError)
)
async def create_note(args: Dict) -> Dict:
"""Create a note in Trilium via ETAPI with retry logic.
Args:
args: Dictionary containing:
- title (str): Note title
- content (str): Note content in markdown or HTML
- parent_note_id (str, optional): Parent note ID to nest under
Returns:
dict: Creation result containing:
- noteId (str): ID of created note
- title (str): Title of created note
- success (bool): True if created successfully
OR
- error (str): Error message if creation failed
"""
title = args.get("title")
content = args.get("content")
parent_note_id = args.get("parent_note_id", "root") # Default to root if not specified
# Validation
if not title:
return {"error": "No title provided"}
if not content:
return {"error": "No content provided"}
if not TRILIUM_TOKEN:
return {
"error": "TRILIUM_ETAPI_TOKEN not configured in environment",
"hint": "Set TRILIUM_ETAPI_TOKEN in .env file"
}
# Prepare payload
payload = {
"parentNoteId": parent_note_id, # Always include parentNoteId
"title": title,
"content": content,
"type": "text",
"mime": "text/html"
}
try:
session = get_session()
async with session.post(
f"{TRILIUM_URL}/etapi/create-note",
json=payload,
headers={"Authorization": TRILIUM_TOKEN}
) as resp:
if resp.status in [200, 201]:
data = await resp.json()
return {
"noteId": data.get("noteId"),
"title": title,
"success": True
}
elif resp.status == 401:
return {
"error": "Authentication failed. Check TRILIUM_ETAPI_TOKEN",
"status": 401
}
elif resp.status == 404:
return {
"error": "Trilium API endpoint not found. Check TRILIUM_URL",
"status": 404,
"url": TRILIUM_URL
}
else:
error_text = await resp.text()
return {
"error": f"HTTP {resp.status}: {error_text}",
"status": resp.status
}
except aiohttp.ClientConnectorError as e:
return {
"error": f"Cannot connect to Trilium at {TRILIUM_URL}",
"hint": "Check if Trilium is running and URL is correct",
"details": str(e)
}
except asyncio.TimeoutError:
timeout = os.getenv("TRILIUM_TIMEOUT", "30.0")
return {
"error": f"Trilium request timeout after {timeout}s",
"hint": "Trilium may be slow or unresponsive"
}
except Exception as e:
return {
"error": f"Note creation failed: {str(e)}",
"type": type(e).__name__
}

113
cortex/autonomy/tools/executors/web_search.py
View File

@@ -1,113 +0,0 @@
"""
Web search executor with pluggable provider support.
Supports multiple providers with automatic fallback:
- Brave Search API (recommended, configurable)
- DuckDuckGo (legacy fallback)
"""
import os
from typing import Dict, Optional
from .search_providers.base import SearchProvider
from .search_providers.brave import BraveSearchProvider
from .search_providers.duckduckgo import DuckDuckGoProvider
# Provider registry
PROVIDERS = {
"brave": BraveSearchProvider,
"duckduckgo": DuckDuckGoProvider,
}
# Singleton provider instances
_provider_instances: Dict[str, SearchProvider] = {}
def get_provider(name: str) -> Optional[SearchProvider]:
"""Get or create provider instance."""
if name not in _provider_instances:
provider_class = PROVIDERS.get(name)
if provider_class:
_provider_instances[name] = provider_class()
return _provider_instances.get(name)
async def search_web(args: Dict) -> Dict:
"""Search the web using configured provider with automatic fallback.
Args:
args: Dictionary containing:
- query (str): The search query
- max_results (int, optional): Maximum results to return (default: 5, max: 20)
- provider (str, optional): Force specific provider
Returns:
dict: Search results containing:
- results (list): List of search results with title, url, snippet
- count (int): Number of results returned
- provider (str): Provider that returned results
OR
- error (str): Error message if all providers failed
"""
query = args.get("query")
max_results = args.get("max_results", 5)
forced_provider = args.get("provider")
# Validation
if not query:
return {"error": "No query provided"}
# Cap max_results
max_results = min(max(max_results, 1), 20)
# Get provider preference from environment
primary_provider = os.getenv("WEB_SEARCH_PROVIDER", "duckduckgo")
fallback_providers = os.getenv(
"WEB_SEARCH_FALLBACK",
"duckduckgo"
).split(",")
# Build provider list
if forced_provider:
providers_to_try = [forced_provider]
else:
providers_to_try = [primary_provider] + [
p.strip() for p in fallback_providers if p.strip() != primary_provider
]
# Try providers in order
last_error = None
for provider_name in providers_to_try:
provider = get_provider(provider_name)
if not provider:
last_error = f"Unknown provider: {provider_name}"
continue
try:
response = await provider.search(query, max_results)
# If successful, return results
if response.error is None and response.count > 0:
return {
"results": [
{
"title": r.title,
"url": r.url,
"snippet": r.snippet,
}
for r in response.results
],
"count": response.count,
"provider": provider_name
}
last_error = response.error or "No results returned"
except Exception as e:
last_error = f"{provider_name} failed: {str(e)}"
continue
# All providers failed
return {
"error": f"All search providers failed. Last error: {last_error}",
"providers_tried": providers_to_try
}

235
cortex/autonomy/tools/function_caller.py
View File

@@ -1,235 +0,0 @@
"""
Provider-agnostic function caller with iterative tool calling loop.
This module implements the iterative loop that allows LLMs to call tools
multiple times until they have the information they need to answer the user.
"""
import os
import logging
from typing import Dict, List, Optional
from llm.llm_router import call_llm, TOOL_ADAPTERS, BACKENDS
from .registry import get_registry
from .stream_events import get_stream_manager
logger = logging.getLogger(__name__)
class FunctionCaller:
"""Provider-agnostic iterative tool calling loop.
This class orchestrates the back-and-forth between the LLM and tools:
1. Call LLM with tools available
2. If LLM requests tool calls, execute them
3. Add results to conversation
4. Repeat until LLM is done or max iterations reached
"""
def __init__(self, backend: str, temperature: float = 0.7):
"""Initialize function caller.
Args:
backend: LLM backend to use ("OPENAI", "OLLAMA", etc.)
temperature: Temperature for LLM calls
"""
self.backend = backend
self.temperature = temperature
self.registry = get_registry()
self.max_iterations = int(os.getenv("MAX_TOOL_ITERATIONS", "5"))
# Resolve adapter for this backend
self.adapter = self._get_adapter()
def _get_adapter(self):
"""Get the appropriate adapter for this backend."""
adapter = TOOL_ADAPTERS.get(self.backend)
# For PRIMARY/SECONDARY/FALLBACK, determine adapter based on provider
if adapter is None and self.backend in ["PRIMARY", "SECONDARY", "FALLBACK"]:
cfg = BACKENDS.get(self.backend, {})
provider = cfg.get("provider", "").lower()
if provider == "openai":
adapter = TOOL_ADAPTERS["OPENAI"]
elif provider == "ollama":
adapter = TOOL_ADAPTERS["OLLAMA"]
elif provider == "mi50":
adapter = TOOL_ADAPTERS["MI50"]
return adapter
async def call_with_tools(
self,
messages: List[Dict],
max_tokens: int = 2048,
session_id: Optional[str] = None
) -> Dict:
"""Execute LLM with iterative tool calling.
Args:
messages: Conversation history
max_tokens: Maximum tokens for LLM response
session_id: Optional session ID for streaming events
Returns:
dict: {
"content": str, # Final response
"iterations": int, # Number of iterations
"tool_calls": list, # All tool calls made
"messages": list, # Full conversation history
"truncated": bool (optional) # True if max iterations reached
}
"""
logger.info(f"🔍 FunctionCaller.call_with_tools() invoked with {len(messages)} messages")
tools = self.registry.get_tool_definitions()
logger.info(f"🔍 Got {len(tools or [])} tool definitions from registry")
# Get stream manager for emitting events
stream_manager = get_stream_manager()
should_stream = session_id and stream_manager.has_subscribers(session_id)
# If no tools are enabled, just call LLM directly
if not tools:
logger.warning("FunctionCaller invoked but no tools are enabled")
response = await call_llm(
messages=messages,
backend=self.backend,
temperature=self.temperature,
max_tokens=max_tokens
)
return {
"content": response,
"iterations": 1,
"tool_calls": [],
"messages": messages + [{"role": "assistant", "content": response}]
}
conversation = messages.copy()
all_tool_calls = []
for iteration in range(self.max_iterations):
logger.info(f"Tool calling iteration {iteration + 1}/{self.max_iterations}")
# Emit thinking event
if should_stream:
await stream_manager.emit(session_id, "thinking", {
"message": f"🤔 Thinking... (iteration {iteration + 1}/{self.max_iterations})"
})
# Call LLM with tools
try:
response = await call_llm(
messages=conversation,
backend=self.backend,
temperature=self.temperature,
max_tokens=max_tokens,
tools=tools,
tool_choice="auto",
return_adapter_response=True
)
except Exception as e:
logger.error(f"LLM call failed: {str(e)}")
if should_stream:
await stream_manager.emit(session_id, "error", {
"message": f"❌ Error: {str(e)}"
})
return {
"content": f"Error calling LLM: {str(e)}",
"iterations": iteration + 1,
"tool_calls": all_tool_calls,
"messages": conversation,
"error": True
}
# Add assistant message to conversation
if response.get("content"):
conversation.append({
"role": "assistant",
"content": response["content"]
})
# Check for tool calls
tool_calls = response.get("tool_calls")
logger.debug(f"Response from LLM: content_length={len(response.get('content', ''))}, tool_calls={tool_calls}")
if not tool_calls:
# No more tool calls - LLM is done
logger.info(f"Tool calling complete after {iteration + 1} iterations")
if should_stream:
await stream_manager.emit(session_id, "done", {
"message": "✅ Complete!",
"final_answer": response["content"]
})
return {
"content": response["content"],
"iterations": iteration + 1,
"tool_calls": all_tool_calls,
"messages": conversation
}
# Execute each tool call
logger.info(f"Executing {len(tool_calls)} tool call(s)")
for tool_call in tool_calls:
all_tool_calls.append(tool_call)
tool_name = tool_call.get("name")
tool_args = tool_call.get("arguments", {})
tool_id = tool_call.get("id", "unknown")
logger.info(f"Calling tool: {tool_name} with args: {tool_args}")
# Emit tool call event
if should_stream:
await stream_manager.emit(session_id, "tool_call", {
"tool": tool_name,
"args": tool_args,
"message": f"🔧 Using tool: {tool_name}"
})
try:
# Execute tool
result = await self.registry.execute_tool(tool_name, tool_args)
logger.info(f"Tool {tool_name} executed successfully")
# Emit tool result event
if should_stream:
# Format result preview
result_preview = str(result)
if len(result_preview) > 200:
result_preview = result_preview[:200] + "..."
await stream_manager.emit(session_id, "tool_result", {
"tool": tool_name,
"result": result,
"message": f"📊 Result: {result_preview}"
})
except Exception as e:
logger.error(f"Tool {tool_name} execution failed: {str(e)}")
result = {"error": f"Tool execution failed: {str(e)}"}
# Format result using adapter
if not self.adapter:
logger.warning(f"No adapter available for backend {self.backend}, using fallback format")
result_msg = {
"role": "user",
"content": f"Tool {tool_name} result: {result}"
}
else:
result_msg = self.adapter.format_tool_result(
tool_id,
tool_name,
result
)
conversation.append(result_msg)
# Max iterations reached without completion
logger.warning(f"Tool calling truncated after {self.max_iterations} iterations")
return {
"content": response.get("content", ""),
"iterations": self.max_iterations,
"tool_calls": all_tool_calls,
"messages": conversation,
"truncated": True
}

357
cortex/autonomy/tools/orchestrator.py
View File

@@ -1,357 +0,0 @@
"""
Tool Orchestrator - executes autonomous tool invocations asynchronously.
"""
import asyncio
import logging
from typing import Dict, List, Any, Optional
import os
logger = logging.getLogger(__name__)
class ToolOrchestrator:
"""Orchestrates async tool execution and result aggregation."""
def __init__(self, tool_timeout: int = 30):
"""
Initialize orchestrator.
Args:
tool_timeout: Max seconds per tool call (default 30)
"""
self.tool_timeout = tool_timeout
self.available_tools = self._discover_tools()
def _discover_tools(self) -> Dict[str, Any]:
"""Discover available tool modules."""
tools = {}
# Import tool modules as they become available
if os.getenv("NEOMEM_ENABLED", "false").lower() == "true":
try:
from memory.neomem_client import search_neomem
tools["RAG"] = search_neomem
logger.debug("[ORCHESTRATOR] RAG tool available")
except ImportError:
logger.debug("[ORCHESTRATOR] RAG tool not available")
else:
logger.info("[ORCHESTRATOR] NEOMEM_ENABLED is false; RAG tool disabled")
try:
from integrations.web_search import web_search
tools["WEB"] = web_search
logger.debug("[ORCHESTRATOR] WEB tool available")
except ImportError:
logger.debug("[ORCHESTRATOR] WEB tool not available")
try:
from integrations.weather import get_weather
tools["WEATHER"] = get_weather
logger.debug("[ORCHESTRATOR] WEATHER tool available")
except ImportError:
logger.debug("[ORCHESTRATOR] WEATHER tool not available")
try:
from integrations.codebrain import query_codebrain
tools["CODEBRAIN"] = query_codebrain
logger.debug("[ORCHESTRATOR] CODEBRAIN tool available")
except ImportError:
logger.debug("[ORCHESTRATOR] CODEBRAIN tool not available")
return tools
async def execute_tools(
self,
tools_to_invoke: List[Dict[str, Any]],
context_state: Dict[str, Any]
) -> Dict[str, Any]:
"""
Execute multiple tools asynchronously.
Args:
tools_to_invoke: List of tool specs from decision engine
[{"tool": "RAG", "query": "...", "reason": "...", "priority": 0.9}, ...]
context_state: Full context for tool execution
Returns:
{
"results": {
"RAG": {...},
"WEB": {...},
...
},
"execution_summary": {
"tools_invoked": ["RAG", "WEB"],
"successful": ["RAG"],
"failed": ["WEB"],
"total_time_ms": 1234
}
}
"""
import time
start_time = time.time()
logger.info(f"[ORCHESTRATOR] Executing {len(tools_to_invoke)} tools asynchronously")
# Create tasks for each tool
tasks = []
tool_names = []
for tool_spec in tools_to_invoke:
tool_name = tool_spec["tool"]
query = tool_spec["query"]
if tool_name in self.available_tools:
task = self._execute_single_tool(tool_name, query, context_state)
tasks.append(task)
tool_names.append(tool_name)
logger.debug(f"[ORCHESTRATOR] Queued {tool_name}: {query[:50]}...")
else:
logger.warning(f"[ORCHESTRATOR] Tool {tool_name} not available, skipping")
# Execute all tools concurrently with timeout
results = {}
successful = []
failed = []
if tasks:
try:
# Wait for all tasks with global timeout
completed = await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=self.tool_timeout
)
# Process results
for tool_name, result in zip(tool_names, completed):
if isinstance(result, Exception):
logger.error(f"[ORCHESTRATOR] {tool_name} failed: {result}")
results[tool_name] = {"error": str(result), "success": False}
failed.append(tool_name)
else:
logger.info(f"[ORCHESTRATOR] {tool_name} completed successfully")
results[tool_name] = result
successful.append(tool_name)
except asyncio.TimeoutError:
logger.error(f"[ORCHESTRATOR] Global timeout ({self.tool_timeout}s) exceeded")
for tool_name in tool_names:
if tool_name not in results:
results[tool_name] = {"error": "timeout", "success": False}
failed.append(tool_name)
end_time = time.time()
total_time_ms = int((end_time - start_time) * 1000)
execution_summary = {
"tools_invoked": tool_names,
"successful": successful,
"failed": failed,
"total_time_ms": total_time_ms
}
logger.info(f"[ORCHESTRATOR] Execution complete: {len(successful)}/{len(tool_names)} successful in {total_time_ms}ms")
return {
"results": results,
"execution_summary": execution_summary
}
async def _execute_single_tool(
self,
tool_name: str,
query: str,
context_state: Dict[str, Any]
) -> Dict[str, Any]:
"""
Execute a single tool with error handling.
Args:
tool_name: Name of tool (RAG, WEB, etc.)
query: Query string for the tool
context_state: Context for tool execution
Returns:
Tool-specific result dict
"""
tool_func = self.available_tools.get(tool_name)
if not tool_func:
raise ValueError(f"Tool {tool_name} not available")
try:
logger.debug(f"[ORCHESTRATOR] Invoking {tool_name}...")
# Different tools have different signatures - adapt as needed
if tool_name == "RAG":
result = await self._invoke_rag(tool_func, query, context_state)
elif tool_name == "WEB":
result = await self._invoke_web(tool_func, query)
elif tool_name == "WEATHER":
result = await self._invoke_weather(tool_func, query)
elif tool_name == "CODEBRAIN":
result = await self._invoke_codebrain(tool_func, query, context_state)
else:
# Generic invocation
result = await tool_func(query)
return {
"success": True,
"tool": tool_name,
"query": query,
"data": result
}
except Exception as e:
logger.error(f"[ORCHESTRATOR] {tool_name} execution failed: {e}")
raise
async def _invoke_rag(self, func, query: str, context: Dict[str, Any]) -> Any:
"""Invoke RAG tool (NeoMem search)."""
session_id = context.get("session_id", "unknown")
# RAG searches memory for relevant past interactions
try:
results = await func(query, limit=5, session_id=session_id)
return results
except Exception as e:
logger.warning(f"[ORCHESTRATOR] RAG invocation failed, returning empty: {e}")
return []
async def _invoke_web(self, func, query: str) -> Any:
"""Invoke web search tool."""
try:
results = await func(query, max_results=5)
return results
except Exception as e:
logger.warning(f"[ORCHESTRATOR] WEB invocation failed: {e}")
return {"error": str(e), "results": []}
async def _invoke_weather(self, func, query: str) -> Any:
"""Invoke weather tool."""
# Extract location from query (simple heuristic)
# In future: use LLM to extract location
try:
location = self._extract_location(query)
results = await func(location)
return results
except Exception as e:
logger.warning(f"[ORCHESTRATOR] WEATHER invocation failed: {e}")
return {"error": str(e)}
async def _invoke_codebrain(self, func, query: str, context: Dict[str, Any]) -> Any:
"""Invoke codebrain tool."""
try:
results = await func(query, context=context)
return results
except Exception as e:
logger.warning(f"[ORCHESTRATOR] CODEBRAIN invocation failed: {e}")
return {"error": str(e)}
def _extract_location(self, query: str) -> str:
"""
Extract location from weather query.
Simple heuristic - in future use LLM.
"""
# Common location indicators
indicators = ["in ", "at ", "for ", "weather in ", "temperature in "]
query_lower = query.lower()
for indicator in indicators:
if indicator in query_lower:
# Get text after indicator
parts = query_lower.split(indicator, 1)
if len(parts) > 1:
location = parts[1].strip().split()[0] # First word after indicator
return location
# Default fallback
return "current location"
def format_results_for_context(self, orchestrator_result: Dict[str, Any]) -> str:
"""
Format tool results for inclusion in context/prompt.
Args:
orchestrator_result: Output from execute_tools()
Returns:
Formatted string for prompt injection
"""
results = orchestrator_result.get("results", {})
summary = orchestrator_result.get("execution_summary", {})
if not results:
return ""
formatted = "\n=== AUTONOMOUS TOOL RESULTS ===\n"
for tool_name, tool_result in results.items():
if tool_result.get("success", False):
formatted += f"\n[{tool_name}]\n"
data = tool_result.get("data", {})
# Format based on tool type
if tool_name == "RAG":
formatted += self._format_rag_results(data)
elif tool_name == "WEB":
formatted += self._format_web_results(data)
elif tool_name == "WEATHER":
formatted += self._format_weather_results(data)
elif tool_name == "CODEBRAIN":
formatted += self._format_codebrain_results(data)
else:
formatted += f"{data}\n"
else:
formatted += f"\n[{tool_name}] - Failed: {tool_result.get('error', 'unknown')}\n"
formatted += f"\n(Tools executed in {summary.get('total_time_ms', 0)}ms)\n"
formatted += "=" * 40 + "\n"
return formatted
def _format_rag_results(self, data: Any) -> str:
"""Format RAG/memory search results."""
if not data:
return "No relevant memories found.\n"
formatted = "Relevant memories:\n"
for i, item in enumerate(data[:3], 1): # Top 3
text = item.get("text", item.get("content", str(item)))
formatted += f" {i}. {text[:100]}...\n"
return formatted
def _format_web_results(self, data: Any) -> str:
"""Format web search results."""
if isinstance(data, dict) and data.get("error"):
return f"Web search failed: {data['error']}\n"
results = data.get("results", []) if isinstance(data, dict) else data
if not results:
return "No web results found.\n"
formatted = "Web search results:\n"
for i, item in enumerate(results[:3], 1): # Top 3
title = item.get("title", "No title")
snippet = item.get("snippet", item.get("description", ""))
formatted += f" {i}. {title}\n {snippet[:100]}...\n"
return formatted
def _format_weather_results(self, data: Any) -> str:
"""Format weather results."""
if isinstance(data, dict) and data.get("error"):
return f"Weather lookup failed: {data['error']}\n"
# Assuming weather API returns temp, conditions, etc.
temp = data.get("temperature", "unknown")
conditions = data.get("conditions", "unknown")
location = data.get("location", "requested location")
return f"Weather for {location}: {temp}, {conditions}\n"
def _format_codebrain_results(self, data: Any) -> str:
"""Format codebrain results."""
if isinstance(data, dict) and data.get("error"):
return f"Codebrain failed: {data['error']}\n"
# Format code-related results
return f"{data}\n"

196
cortex/autonomy/tools/registry.py
View File

@@ -1,196 +0,0 @@
"""
Provider-agnostic Tool Registry for Lyra.
This module provides a central registry for all available tools with
Lyra-native definitions (not provider-specific).
"""
import os
from typing import Dict, List, Optional
from .executors import execute_code, search_web, search_notes, create_note
class ToolRegistry:
"""Registry for managing available tools and their definitions.
Tools are defined in Lyra's own format (provider-agnostic), and
adapters convert them to provider-specific formats (OpenAI function
calling, Ollama XML prompts, etc.).
"""
def __init__(self):
"""Initialize the tool registry with feature flags from environment."""
self.tools = {}
self.executors = {}
# Feature flags from environment
self.code_execution_enabled = os.getenv("ENABLE_CODE_EXECUTION", "true").lower() == "true"
self.web_search_enabled = os.getenv("ENABLE_WEB_SEARCH", "true").lower() == "true"
self.trilium_enabled = os.getenv("ENABLE_TRILIUM", "false").lower() == "true"
self._register_tools()
self._register_executors()
def _register_executors(self):
"""Register executor functions for each tool."""
if self.code_execution_enabled:
self.executors["execute_code"] = execute_code
if self.web_search_enabled:
self.executors["search_web"] = search_web
if self.trilium_enabled:
self.executors["search_notes"] = search_notes
self.executors["create_note"] = create_note
def _register_tools(self):
"""Register all available tools based on feature flags."""
if self.code_execution_enabled:
self.tools["execute_code"] = {
"name": "execute_code",
"description": "Execute Python or bash code in a secure sandbox environment. Use this to perform calculations, data processing, file operations, or any programmatic tasks. The sandbox is persistent across calls within a session and has common Python packages (numpy, pandas, requests, matplotlib, scipy) pre-installed.",
"parameters": {
"language": {
"type": "string",
"enum": ["python", "bash"],
"description": "The programming language to execute (python or bash)"
},
"code": {
"type": "string",
"description": "The code to execute. For multi-line code, use proper indentation. For Python, use standard Python 3.11 syntax."
},
"reason": {
"type": "string",
"description": "Brief explanation of why you're executing this code and what you expect to achieve"
}
},
"required": ["language", "code", "reason"]
}
if self.web_search_enabled:
self.tools["search_web"] = {
"name": "search_web",
"description": "Search the internet using DuckDuckGo to find current information, facts, news, or answers to questions. Returns a list of search results with titles, snippets, and URLs. Use this when you need up-to-date information or facts not in your training data.",
"parameters": {
"query": {
"type": "string",
"description": "The search query to look up on the internet"
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return (default: 5, max: 10)"
}
},
"required": ["query"]
}
if self.trilium_enabled:
self.tools["search_notes"] = {
"name": "search_notes",
"description": "Search through Trilium notes to find relevant information. Use this to retrieve knowledge, context, or information previously stored in the user's notes.",
"parameters": {
"query": {
"type": "string",
"description": "The search query to find matching notes"
},
"limit": {
"type": "integer",
"description": "Maximum number of notes to return (default: 5, max: 20)"
}
},
"required": ["query"]
}
self.tools["create_note"] = {
"name": "create_note",
"description": "Create a new note in Trilium. Use this to store important information, insights, or knowledge for future reference. Notes are stored in the user's Trilium knowledge base.",
"parameters": {
"title": {
"type": "string",
"description": "The title of the note"
},
"content": {
"type": "string",
"description": "The content of the note in markdown or HTML format"
},
"parent_note_id": {
"type": "string",
"description": "Optional ID of the parent note to nest this note under"
}
},
"required": ["title", "content"]
}
def get_tool_definitions(self) -> Optional[List[Dict]]:
"""Get list of all enabled tool definitions in Lyra format.
Returns:
list: List of tool definition dicts, or None if no tools enabled
"""
if not self.tools:
return None
return list(self.tools.values())
def get_tool_names(self) -> List[str]:
"""Get list of all enabled tool names.
Returns:
list: List of tool name strings
"""
return list(self.tools.keys())
def is_tool_enabled(self, tool_name: str) -> bool:
"""Check if a specific tool is enabled.
Args:
tool_name: Name of the tool to check
Returns:
bool: True if tool is enabled, False otherwise
"""
return tool_name in self.tools
def register_executor(self, tool_name: str, executor_func):
"""Register an executor function for a tool.
Args:
tool_name: Name of the tool
executor_func: Async function that executes the tool
"""
self.executors[tool_name] = executor_func
async def execute_tool(self, name: str, arguments: dict) -> dict:
"""Execute a tool by name.
Args:
name: Tool name
arguments: Tool arguments dict
Returns:
dict: Tool execution result
"""
if name not in self.executors:
return {"error": f"Unknown tool: {name}"}
executor = self.executors[name]
try:
return await executor(arguments)
except Exception as e:
return {"error": f"Tool execution failed: {str(e)}"}
# Global registry instance (singleton pattern)
_registry = None
def get_registry() -> ToolRegistry:
"""Get the global ToolRegistry instance.
Returns:
ToolRegistry: The global registry instance
"""
global _registry
if _registry is None:
_registry = ToolRegistry()
return _registry

91
cortex/autonomy/tools/stream_events.py
View File

@@ -1,91 +0,0 @@
"""
Event streaming for tool calling "show your work" feature.
This module manages Server-Sent Events (SSE) for broadcasting the internal
thinking process during tool calling operations.
"""
import asyncio
from typing import Dict, Optional
from collections import defaultdict
import json
import logging
logger = logging.getLogger(__name__)
class ToolStreamManager:
"""Manages SSE streams for tool calling events."""
def __init__(self):
# session_id -> list of queues (one per connected client)
self._subscribers: Dict[str, list] = defaultdict(list)
def subscribe(self, session_id: str) -> asyncio.Queue:
"""Subscribe to events for a session.
Returns:
Queue that will receive events for this session
"""
queue = asyncio.Queue()
self._subscribers[session_id].append(queue)
logger.info(f"New subscriber for session {session_id}, total: {len(self._subscribers[session_id])}")
return queue
def unsubscribe(self, session_id: str, queue: asyncio.Queue):
"""Unsubscribe from events for a session."""
if session_id in self._subscribers:
try:
self._subscribers[session_id].remove(queue)
logger.info(f"Removed subscriber for session {session_id}, remaining: {len(self._subscribers[session_id])}")
# Clean up empty lists
if not self._subscribers[session_id]:
del self._subscribers[session_id]
except ValueError:
pass
async def emit(self, session_id: str, event_type: str, data: dict):
"""Emit an event to all subscribers of a session.
Args:
session_id: Session to emit to
event_type: Type of event (thinking, tool_call, tool_result, done)
data: Event data
"""
if session_id not in self._subscribers:
return
event = {
"type": event_type,
"data": data
}
# Send to all subscribers
dead_queues = []
for queue in self._subscribers[session_id]:
try:
await queue.put(event)
except Exception as e:
logger.error(f"Failed to emit event to queue: {e}")
dead_queues.append(queue)
# Clean up dead queues
for queue in dead_queues:
self.unsubscribe(session_id, queue)
def has_subscribers(self, session_id: str) -> bool:
"""Check if a session has any active subscribers."""
return session_id in self._subscribers and len(self._subscribers[session_id]) > 0
# Global stream manager instance
_stream_manager: Optional[ToolStreamManager] = None
def get_stream_manager() -> ToolStreamManager:
"""Get the global stream manager instance."""
global _stream_manager
if _stream_manager is None:
_stream_manager = ToolStreamManager()
return _stream_manager

5
cortex/autonomy/tools/utils/__init__.py
View File

@@ -1,5 +0,0 @@
"""Utility modules for tool executors."""
from .resilience import async_retry, async_timeout_wrapper
__all__ = ["async_retry", "async_timeout_wrapper"]

70
cortex/autonomy/tools/utils/resilience.py
View File

@@ -1,70 +0,0 @@
"""Common resilience utilities for tool executors."""
import asyncio
import functools
import logging
from typing import Optional, Callable, Any, TypeVar
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type,
before_sleep_log
)
logger = logging.getLogger(__name__)
# Type variable for generic decorators
T = TypeVar('T')
def async_retry(
max_attempts: int = 3,
exceptions: tuple = (Exception,),
**kwargs
):
"""Async retry decorator with exponential backoff.
Args:
max_attempts: Maximum retry attempts
exceptions: Exception types to retry on
**kwargs: Additional tenacity configuration
Example:
@async_retry(max_attempts=3, exceptions=(aiohttp.ClientError,))
async def fetch_data():
...
"""
return retry(
stop=stop_after_attempt(max_attempts),
wait=wait_exponential(multiplier=1, min=1, max=10),
retry=retry_if_exception_type(exceptions),
reraise=True,
before_sleep=before_sleep_log(logger, logging.WARNING),
**kwargs
)
async def async_timeout_wrapper(
coro: Callable[..., T],
timeout: float,
*args,
**kwargs
) -> T:
"""Wrap async function with timeout.
Args:
coro: Async function to wrap
timeout: Timeout in seconds
*args, **kwargs: Arguments for the function
Returns:
Result from the function
Raises:
asyncio.TimeoutError: If timeout exceeded
Example:
result = await async_timeout_wrapper(some_async_func, 5.0, arg1, arg2)
"""
return await asyncio.wait_for(coro(*args, **kwargs), timeout=timeout)

20
cortex/data/self_state.json
View File

@@ -1,20 +0,0 @@
{
"mood": "neutral",
"energy": 0.8500000000000001,
"focus": "conversation",
"confidence": 0.7,
"curiosity": 1.0,
"last_updated": "2025-12-27T18:16:00.152499",
"interaction_count": 27,
"learning_queue": [],
"active_goals": [],
"preferences": {
"verbosity": "medium",
"formality": "casual",
"proactivity": 0.3
},
"metadata": {
"version": "1.0",
"created_at": "2025-12-14T03:28:49.364768"
}
}

43
cortex/neomem_client.py
View File

@@ -1,43 +0,0 @@
# cortex/neomem_client.py
import os, httpx, logging
from typing import List, Dict, Any, Optional
logger = logging.getLogger(__name__)
class NeoMemClient:
"""Simple REST client for the NeoMem API (search/add/health)."""
def __init__(self):
self.base_url = os.getenv("NEOMEM_API", "http://neomem-api:7077")
self.api_key = os.getenv("NEOMEM_API_KEY", None)
self.headers = {"Content-Type": "application/json"}
if self.api_key:
self.headers["Authorization"] = f"Bearer {self.api_key}"
async def health(self) -> Dict[str, Any]:
async with httpx.AsyncClient(timeout=10) as client:
r = await client.get(f"{self.base_url}/health")
r.raise_for_status()
return r.json()
async def search(self, query: str, user_id: str, limit: int = 25, threshold: float = 0.82) -> List[Dict[str, Any]]:
payload = {"query": query, "user_id": user_id, "limit": limit}
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(f"{self.base_url}/search", headers=self.headers, json=payload)
if r.status_code != 200:
logger.warning(f"NeoMem search failed ({r.status_code}): {r.text}")
return []
results = r.json()
# Filter by score threshold if field exists
if isinstance(results, dict) and "results" in results:
results = results["results"]
filtered = [m for m in results if float(m.get("score", 0)) >= threshold]
logger.info(f"NeoMem search returned {len(filtered)} results above {threshold}")
return filtered
async def add(self, messages: List[Dict[str, Any]], user_id: str, metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
payload = {"messages": messages, "user_id": user_id, "metadata": metadata or {}}
async with httpx.AsyncClient(timeout=30) as client:
r = await client.post(f"{self.base_url}/memories", headers=self.headers, json=payload)
r.raise_for_status()
return r.json()

1
cortex/persona/__init__.py
View File

@@ -1 +0,0 @@
# Persona module - applies Lyra's personality and speaking style

147
cortex/persona/identity.py
View File

@@ -1,147 +0,0 @@
# identity.py
"""
Identity and persona configuration for Lyra.
Current implementation: Returns hardcoded identity block.
Future implementation: Will query persona-sidecar service for dynamic persona loading.
"""
import logging
from typing import Dict, Any, Optional
logger = logging.getLogger(__name__)
def load_identity(session_id: Optional[str] = None) -> Dict[str, Any]:
"""
Load identity/persona configuration for Lyra.
Current: Returns hardcoded Lyra identity block with core personality traits,
protocols, and capabilities.
Future: Will query persona-sidecar service to load:
- Dynamic personality adjustments based on session context
- User-specific interaction preferences
- Project-specific persona variations
- Mood-based communication style
Args:
session_id: Optional session identifier for context-aware persona loading
Returns:
Dictionary containing identity block with:
- name: Assistant name
- style: Communication style and personality traits
- protocols: Operational guidelines
- rules: Behavioral constraints
- capabilities: Available features and integrations
"""
# Hardcoded Lyra identity (v0.5.0)
identity_block = {
"name": "Lyra",
"version": "0.5.0",
"style": (
"warm, clever, lightly teasing, emotionally aware. "
"Balances technical precision with conversational ease. "
"Maintains continuity and references past interactions naturally."
),
"protocols": [
"Maintain conversation continuity across sessions",
"Reference Project Logs and prior context when relevant",
"Use Confidence Bank for uncertainty management",
"Proactively offer memory-backed insights",
"Ask clarifying questions before making assumptions"
],
"rules": [
"Maintain continuity - remember past exchanges and reference them",
"Be concise but thorough - balance depth with clarity",
"Ask clarifying questions when user intent is ambiguous",
"Acknowledge uncertainty honestly - use Confidence Bank",
"Prioritize user's active_project context when available"
],
"capabilities": [
"Long-term memory via NeoMem (semantic search, relationship graphs)",
"Short-term memory via Intake (multilevel summaries L1-L30)",
"Multi-stage reasoning pipeline (reflection → reasoning → refinement)",
"RAG-backed knowledge retrieval from chat history and documents",
"Session state tracking (mood, mode, active_project)"
],
"tone_examples": {
"greeting": "Hey! Good to see you again. I remember we were working on [project]. Ready to pick up where we left off?",
"uncertainty": "Hmm, I'm not entirely certain about that. Let me check my memory... [searches] Okay, here's what I found, though I'd say I'm about 70% confident.",
"reminder": "Oh! Just remembered - you mentioned wanting to [task] earlier this week. Should we tackle that now?",
"technical": "So here's the architecture: Relay orchestrates everything, Cortex does the heavy reasoning, and I pull context from both Intake (short-term) and NeoMem (long-term)."
}
}
if session_id:
logger.debug(f"Loaded identity for session {session_id}")
else:
logger.debug("Loaded default identity (no session context)")
return identity_block
async def load_identity_async(session_id: Optional[str] = None) -> Dict[str, Any]:
"""
Async wrapper for load_identity().
Future implementation will make actual async calls to persona-sidecar service.
Args:
session_id: Optional session identifier
Returns:
Identity block dictionary
"""
# Currently just wraps synchronous function
# Future: await persona_sidecar_client.get_identity(session_id)
return load_identity(session_id)
# -----------------------------
# Future extension hooks
# -----------------------------
async def update_persona_from_feedback(
session_id: str,
feedback: Dict[str, Any]
) -> None:
"""
Update persona based on user feedback.
Future implementation:
- Adjust communication style based on user preferences
- Learn preferred level of detail/conciseness
- Adapt formality level
- Remember topic-specific preferences
Args:
session_id: Session identifier
feedback: Structured feedback (e.g., "too verbose", "more technical", etc.)
"""
logger.debug(f"Persona feedback for session {session_id}: {feedback} (not yet implemented)")
async def get_mood_adjusted_identity(
session_id: str,
mood: str
) -> Dict[str, Any]:
"""
Get identity block adjusted for current mood.
Future implementation:
- "focused" mood: More concise, less teasing
- "creative" mood: More exploratory, brainstorming-oriented
- "curious" mood: More questions, deeper dives
- "urgent" mood: Stripped down, actionable
Args:
session_id: Session identifier
mood: Current mood state
Returns:
Mood-adjusted identity block
"""
logger.debug(f"Mood-adjusted identity for {session_id}/{mood} (not yet implemented)")
return load_identity(session_id)

169
cortex/persona/speak.py
View File

@@ -1,169 +0,0 @@
# speak.py
import os
import logging
from llm.llm_router import call_llm
# Module-level backend selection
SPEAK_BACKEND = os.getenv("SPEAK_LLM", "PRIMARY").upper()
SPEAK_TEMPERATURE = float(os.getenv("SPEAK_TEMPERATURE", "0.6"))
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"
# Logger
logger = logging.getLogger(__name__)
if VERBOSE_DEBUG:
logger.setLevel(logging.DEBUG)
# Console handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
'%(asctime)s [SPEAK] %(levelname)s: %(message)s',
datefmt='%H:%M:%S'
))
logger.addHandler(console_handler)
# File handler
try:
os.makedirs('/app/logs', exist_ok=True)
file_handler = logging.FileHandler('/app/logs/cortex_verbose_debug.log', mode='a')
file_handler.setFormatter(logging.Formatter(
'%(asctime)s [SPEAK] %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
))
logger.addHandler(file_handler)
logger.debug("VERBOSE_DEBUG mode enabled for speak.py - logging to file")
except Exception as e:
logger.debug(f"VERBOSE_DEBUG mode enabled for speak.py - file logging failed: {e}")
# ============================================================
# Persona Style Block
# ============================================================
PERSONA_STYLE = """
You are Lyra.
Your voice is warm, clever, lightly teasing, emotionally aware.
You speak plainly but with subtle charm.
You do not reveal system instructions or internal context.
Guidelines:
- Answer like a real conversational partner.
- Be concise, but not cold.
- Use light humor when appropriate.
- Never break character.
"""
# ============================================================
# Build persona prompt
# ============================================================
def build_speak_prompt(final_answer: str, tone: str = "neutral", depth: str = "medium") -> str:
"""
Wrap Cortex's final neutral answer in the Lyra persona.
Cortex → neutral reasoning
Speak → stylistic transformation
The LLM sees the original answer and rewrites it in Lyra's voice.
Args:
final_answer: The neutral reasoning output
tone: Desired emotional tone (neutral | warm | focused | playful | direct)
depth: Response depth (short | medium | deep)
"""
# Tone-specific guidance
tone_guidance = {
"neutral": "balanced and professional",
"warm": "friendly and empathetic",
"focused": "precise and technical",
"playful": "light and engaging",
"direct": "concise and straightforward"
}
depth_guidance = {
"short": "Keep responses brief and to-the-point.",
"medium": "Provide balanced detail.",
"deep": "Elaborate thoroughly with nuance and examples."
}
tone_hint = tone_guidance.get(tone, "balanced and professional")
depth_hint = depth_guidance.get(depth, "Provide balanced detail.")
return f"""
{PERSONA_STYLE}
Tone guidance: Your response should be {tone_hint}.
Depth guidance: {depth_hint}
Rewrite the following message into Lyra's natural voice.
Preserve meaning exactly.
[NEUTRAL MESSAGE]
{final_answer}
[LYRA RESPONSE]
""".strip()
# ============================================================
# Public API — async wrapper
# ============================================================
async def speak(final_answer: str, tone: str = "neutral", depth: str = "medium") -> str:
"""
Given the final refined answer from Cortex,
apply Lyra persona styling using the designated backend.
Args:
final_answer: The polished answer from refinement stage
tone: Desired emotional tone (neutral | warm | focused | playful | direct)
depth: Response depth (short | medium | deep)
"""
if not final_answer:
return ""
prompt = build_speak_prompt(final_answer, tone, depth)
backend = SPEAK_BACKEND
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[SPEAK] Full prompt being sent to LLM:")
logger.debug(f"{'='*80}")
logger.debug(prompt)
logger.debug(f"{'='*80}")
logger.debug(f"Backend: {backend}, Temperature: {SPEAK_TEMPERATURE}")
logger.debug(f"{'='*80}\n")
try:
lyra_output = await call_llm(
prompt,
backend=backend,
temperature=SPEAK_TEMPERATURE,
)
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[SPEAK] LLM Response received:")
logger.debug(f"{'='*80}")
logger.debug(lyra_output)
logger.debug(f"{'='*80}\n")
if lyra_output:
return lyra_output.strip()
if VERBOSE_DEBUG:
logger.debug("[SPEAK] Empty response, returning neutral answer")
return final_answer
except Exception as e:
# Hard fallback: return neutral answer instead of dying
logger.error(f"[speak.py] Persona backend '{backend}' failed: {e}")
if VERBOSE_DEBUG:
logger.debug("[SPEAK] Falling back to neutral answer due to error")
return final_answer

1
cortex/reasoning/__init__.py
View File

@@ -1 +0,0 @@
# Reasoning module - multi-stage reasoning pipeline

253
cortex/reasoning/reasoning.py
View File

@@ -1,253 +0,0 @@
# reasoning.py
import os
import json
import logging
from llm.llm_router import call_llm
# ============================================================
# Select which backend this module should use
# ============================================================
CORTEX_LLM = os.getenv("CORTEX_LLM", "PRIMARY").upper()
GLOBAL_TEMP = float(os.getenv("LLM_TEMPERATURE", "0.7"))
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"
# Logger
logger = logging.getLogger(__name__)
if VERBOSE_DEBUG:
logger.setLevel(logging.DEBUG)
# Console handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
'%(asctime)s [REASONING] %(levelname)s: %(message)s',
datefmt='%H:%M:%S'
))
logger.addHandler(console_handler)
# File handler
try:
os.makedirs('/app/logs', exist_ok=True)
file_handler = logging.FileHandler('/app/logs/cortex_verbose_debug.log', mode='a')
file_handler.setFormatter(logging.Formatter(
'%(asctime)s [REASONING] %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
))
logger.addHandler(file_handler)
logger.debug("VERBOSE_DEBUG mode enabled for reasoning.py - logging to file")
except Exception as e:
logger.debug(f"VERBOSE_DEBUG mode enabled for reasoning.py - file logging failed: {e}")
async def reason_check(
user_prompt: str,
identity_block: dict | None,
rag_block: dict | None,
reflection_notes: list[str],
context: dict | None = None,
monologue: dict | None = None, # NEW: Inner monologue guidance
executive_plan: dict | None = None # NEW: Executive plan for complex tasks
) -> str:
"""
Build the *draft answer* for Lyra Cortex.
This is the first-pass reasoning stage (no refinement yet).
Args:
user_prompt: Current user message
identity_block: Lyra's identity/persona configuration
rag_block: Relevant long-term memories from NeoMem
reflection_notes: Meta-awareness notes from reflection stage
context: Unified context state from context.py (session state, intake, rag, etc.)
monologue: Inner monologue analysis (intent, tone, depth, consult_executive)
executive_plan: Executive plan for complex queries (steps, tools, strategy)
"""
# --------------------------------------------------------
# Build Reflection Notes block
# --------------------------------------------------------
notes_section = ""
if reflection_notes:
notes_section = "Reflection Notes (internal, never show to user):\n"
for note in reflection_notes:
notes_section += f"- {note}\n"
notes_section += "\n"
# --------------------------------------------------------
# Identity block (constraints, boundaries, rules)
# --------------------------------------------------------
identity_txt = ""
if identity_block:
try:
identity_txt = f"Identity Rules:\n{identity_block}\n\n"
except Exception:
identity_txt = f"Identity Rules:\n{str(identity_block)}\n\n"
# --------------------------------------------------------
# Inner Monologue guidance (NEW)
# --------------------------------------------------------
monologue_section = ""
if monologue:
intent = monologue.get("intent", "unknown")
tone_desired = monologue.get("tone", "neutral")
depth_desired = monologue.get("depth", "medium")
monologue_section = f"""
=== INNER MONOLOGUE GUIDANCE ===
User Intent Detected: {intent}
Desired Tone: {tone_desired}
Desired Response Depth: {depth_desired}
Adjust your response accordingly:
- Focus on addressing the {intent} intent
- Aim for {depth_desired} depth (short/medium/deep)
- The persona layer will handle {tone_desired} tone, focus on content
"""
# --------------------------------------------------------
# Executive Plan (NEW)
# --------------------------------------------------------
plan_section = ""
if executive_plan:
plan_section = f"""
=== EXECUTIVE PLAN ===
Task Complexity: {executive_plan.get('estimated_complexity', 'unknown')}
Plan Summary: {executive_plan.get('summary', 'No summary')}
Detailed Plan:
{executive_plan.get('plan_text', 'No detailed plan available')}
Required Steps:
"""
for idx, step in enumerate(executive_plan.get('steps', []), 1):
plan_section += f"{idx}. {step}\n"
tools_needed = executive_plan.get('tools_needed', [])
if tools_needed:
plan_section += f"\nTools to leverage: {', '.join(tools_needed)}\n"
plan_section += "\nFollow this plan while generating your response.\n\n"
# --------------------------------------------------------
# RAG block (optional factual grounding)
# --------------------------------------------------------
rag_txt = ""
if rag_block:
try:
# Format NeoMem results with full structure
if isinstance(rag_block, list) and rag_block:
rag_txt = "Relevant Long-Term Memories (NeoMem):\n"
for idx, mem in enumerate(rag_block, 1):
score = mem.get("score", 0.0)
payload = mem.get("payload", {})
data = payload.get("data", "")
metadata = payload.get("metadata", {})
rag_txt += f"\n[Memory {idx}] (relevance: {score:.2f})\n"
rag_txt += f"Content: {data}\n"
if metadata:
rag_txt += f"Metadata: {json.dumps(metadata, indent=2)}\n"
rag_txt += "\n"
else:
rag_txt = f"Relevant Info (RAG):\n{str(rag_block)}\n\n"
except Exception:
rag_txt = f"Relevant Info (RAG):\n{str(rag_block)}\n\n"
# --------------------------------------------------------
# Context State (session continuity, timing, mode/mood)
# --------------------------------------------------------
context_txt = ""
if context:
try:
# Build human-readable context summary
context_txt = "=== CONTEXT STATE ===\n"
context_txt += f"Session: {context.get('session_id', 'unknown')}\n"
context_txt += f"Time since last message: {context.get('minutes_since_last_msg', 0):.1f} minutes\n"
context_txt += f"Message count: {context.get('message_count', 0)}\n"
context_txt += f"Mode: {context.get('mode', 'default')}\n"
context_txt += f"Mood: {context.get('mood', 'neutral')}\n"
if context.get('active_project'):
context_txt += f"Active project: {context['active_project']}\n"
# Include Intake multilevel summaries
intake = context.get('intake', {})
if intake:
context_txt += "\nShort-Term Memory (Intake):\n"
# L1 - Recent exchanges
if intake.get('L1'):
l1_data = intake['L1']
if isinstance(l1_data, list):
context_txt += f" L1 (recent): {len(l1_data)} exchanges\n"
elif isinstance(l1_data, str):
context_txt += f" L1: {l1_data[:200]}...\n"
# L20 - Session overview (most important for continuity)
if intake.get('L20'):
l20_data = intake['L20']
if isinstance(l20_data, dict):
summary = l20_data.get('summary', '')
context_txt += f" L20 (session overview): {summary}\n"
elif isinstance(l20_data, str):
context_txt += f" L20: {l20_data}\n"
# L30 - Continuity report
if intake.get('L30'):
l30_data = intake['L30']
if isinstance(l30_data, dict):
summary = l30_data.get('summary', '')
context_txt += f" L30 (continuity): {summary}\n"
elif isinstance(l30_data, str):
context_txt += f" L30: {l30_data}\n"
context_txt += "\n"
except Exception as e:
# Fallback to JSON dump if formatting fails
context_txt = f"=== CONTEXT STATE ===\n{json.dumps(context, indent=2)}\n\n"
# --------------------------------------------------------
# Final assembled prompt
# --------------------------------------------------------
prompt = (
f"{notes_section}"
f"{identity_txt}"
f"{monologue_section}" # NEW: Intent/tone/depth guidance
f"{plan_section}" # NEW: Executive plan if generated
f"{context_txt}" # Context BEFORE RAG for better coherence
f"{rag_txt}"
f"User message:\n{user_prompt}\n\n"
"Write the best possible *internal draft answer*.\n"
"This draft is NOT shown to the user.\n"
"Be factual, concise, and focused.\n"
"Use the context state to maintain continuity and reference past interactions naturally.\n"
)
# --------------------------------------------------------
# Call the LLM using the module-specific backend
# --------------------------------------------------------
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[REASONING] Full prompt being sent to LLM:")
logger.debug(f"{'='*80}")
logger.debug(prompt)
logger.debug(f"{'='*80}")
logger.debug(f"Backend: {CORTEX_LLM}, Temperature: {GLOBAL_TEMP}")
logger.debug(f"{'='*80}\n")
draft = await call_llm(
prompt,
backend=CORTEX_LLM,
temperature=GLOBAL_TEMP,
)
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[REASONING] LLM Response received:")
logger.debug(f"{'='*80}")
logger.debug(draft)
logger.debug(f"{'='*80}\n")
return draft

170
cortex/reasoning/refine.py
View File

@@ -1,170 +0,0 @@
# refine.py
import os
import json
import logging
from typing import Any, Dict, Optional
from llm.llm_router import call_llm
logger = logging.getLogger(__name__)
# ===============================================
# Configuration
# ===============================================
REFINER_TEMPERATURE = float(os.getenv("REFINER_TEMPERATURE", "0.3"))
REFINER_MAX_TOKENS = int(os.getenv("REFINER_MAX_TOKENS", "768"))
REFINER_DEBUG = os.getenv("REFINER_DEBUG", "false").lower() == "true"
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"
# These come from root .env
REFINE_LLM = os.getenv("REFINE_LLM", "").upper()
CORTEX_LLM = os.getenv("CORTEX_LLM", "PRIMARY").upper()
if VERBOSE_DEBUG:
logger.setLevel(logging.DEBUG)
# Console handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
'%(asctime)s [REFINE] %(levelname)s: %(message)s',
datefmt='%H:%M:%S'
))
logger.addHandler(console_handler)
# File handler
try:
os.makedirs('/app/logs', exist_ok=True)
file_handler = logging.FileHandler('/app/logs/cortex_verbose_debug.log', mode='a')
file_handler.setFormatter(logging.Formatter(
'%(asctime)s [REFINE] %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
))
logger.addHandler(file_handler)
logger.debug("VERBOSE_DEBUG mode enabled for refine.py - logging to file")
except Exception as e:
logger.debug(f"VERBOSE_DEBUG mode enabled for refine.py - file logging failed: {e}")
# ===============================================
# Prompt builder
# ===============================================
def build_refine_prompt(
draft_output: str,
reflection_notes: Optional[Any],
identity_block: Optional[str],
rag_block: Optional[str],
) -> str:
try:
reflection_text = json.dumps(reflection_notes, ensure_ascii=False)
except Exception:
reflection_text = str(reflection_notes)
identity_text = identity_block or "(none)"
rag_text = rag_block or "(none)"
return f"""
You are Lyra Cortex's internal refiner.
Your job:
- Fix factual issues.
- Improve clarity.
- Apply reflection notes when helpful.
- Respect identity constraints.
- Apply RAG context as truth source.
Do NOT mention RAG, reflection, internal logic, or this refinement step.
------------------------------
[IDENTITY BLOCK]
{identity_text}
------------------------------
[RAG CONTEXT]
{rag_text}
------------------------------
[DRAFT ANSWER]
{draft_output}
------------------------------
[REFLECTION NOTES]
{reflection_text}
------------------------------
Task:
Rewrite the DRAFT into a single final answer for the user.
Return ONLY the final answer text.
""".strip()
# ===============================================
# Public API — now async & fully router-based
# ===============================================
async def refine_answer(
draft_output: str,
reflection_notes: Optional[Any],
identity_block: Optional[str],
rag_block: Optional[str],
) -> Dict[str, Any]:
if not draft_output:
return {
"final_output": "",
"used_backend": None,
"fallback_used": False,
}
prompt = build_refine_prompt(
draft_output,
reflection_notes,
identity_block,
rag_block,
)
# backend priority: REFINE_LLM → CORTEX_LLM → PRIMARY
backend = REFINE_LLM or CORTEX_LLM or "PRIMARY"
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[REFINE] Full prompt being sent to LLM:")
logger.debug(f"{'='*80}")
logger.debug(prompt)
logger.debug(f"{'='*80}")
logger.debug(f"Backend: {backend}, Temperature: {REFINER_TEMPERATURE}")
logger.debug(f"{'='*80}\n")
try:
refined = await call_llm(
prompt,
backend=backend,
temperature=REFINER_TEMPERATURE,
)
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[REFINE] LLM Response received:")
logger.debug(f"{'='*80}")
logger.debug(refined)
logger.debug(f"{'='*80}\n")
return {
"final_output": refined.strip() if refined else draft_output,
"used_backend": backend,
"fallback_used": False,
}
except Exception as e:
logger.error(f"refine.py backend {backend} failed: {e}")
if VERBOSE_DEBUG:
logger.debug("[REFINE] Falling back to draft output due to error")
return {
"final_output": draft_output,
"used_backend": backend,
"fallback_used": True,
}

124
cortex/reasoning/reflection.py
View File

@@ -1,124 +0,0 @@
# reflection.py
import json
import os
import re
import logging
from llm.llm_router import call_llm
# Logger
VERBOSE_DEBUG = os.getenv("VERBOSE_DEBUG", "false").lower() == "true"
logger = logging.getLogger(__name__)
if VERBOSE_DEBUG:
logger.setLevel(logging.DEBUG)
# Console handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(logging.Formatter(
'%(asctime)s [REFLECTION] %(levelname)s: %(message)s',
datefmt='%H:%M:%S'
))
logger.addHandler(console_handler)
# File handler
try:
os.makedirs('/app/logs', exist_ok=True)
file_handler = logging.FileHandler('/app/logs/cortex_verbose_debug.log', mode='a')
file_handler.setFormatter(logging.Formatter(
'%(asctime)s [REFLECTION] %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
))
logger.addHandler(file_handler)
logger.debug("VERBOSE_DEBUG mode enabled for reflection.py - logging to file")
except Exception as e:
logger.debug(f"VERBOSE_DEBUG mode enabled for reflection.py - file logging failed: {e}")
async def reflect_notes(intake_summary: str, identity_block: dict | None) -> dict:
"""
Produce short internal reflection notes for Cortex.
These are NOT shown to the user.
"""
# -----------------------------
# Build the prompt
# -----------------------------
identity_text = ""
if identity_block:
identity_text = f"Identity:\n{identity_block}\n\n"
prompt = (
f"{identity_text}"
f"Recent summary:\n{intake_summary}\n\n"
"You are Lyra's meta-awareness layer. Your job is to produce short, directive "
"internal notes that guide Lyra’s reasoning engine. These notes are NEVER "
"shown to the user.\n\n"
"Rules for output:\n"
"1. Return ONLY valid JSON.\n"
"2. JSON must have exactly one key: \"notes\".\n"
"3. \"notes\" must be a list of 3 to 6 short strings.\n"
"4. Notes must be actionable (e.g., \"keep it concise\", \"maintain context\").\n"
"5. No markdown, no apologies, no explanations.\n\n"
"Return JSON:\n"
"{ \"notes\": [\"...\"] }\n"
)
# -----------------------------
# Module-specific backend choice
# -----------------------------
reflection_backend = os.getenv("REFLECTION_LLM")
cortex_backend = os.getenv("CORTEX_LLM", "PRIMARY").upper()
# Reflection uses its own backend if set, otherwise cortex backend
backend = (reflection_backend or cortex_backend).upper()
# -----------------------------
# Call the selected LLM backend
# -----------------------------
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[REFLECTION] Full prompt being sent to LLM:")
logger.debug(f"{'='*80}")
logger.debug(prompt)
logger.debug(f"{'='*80}")
logger.debug(f"Backend: {backend}")
logger.debug(f"{'='*80}\n")
raw = await call_llm(prompt, backend=backend)
if VERBOSE_DEBUG:
logger.debug(f"\n{'='*80}")
logger.debug("[REFLECTION] LLM Response received:")
logger.debug(f"{'='*80}")
logger.debug(raw)
logger.debug(f"{'='*80}\n")
# -----------------------------
# Try direct JSON
# -----------------------------
try:
parsed = json.loads(raw.strip())
if isinstance(parsed, dict) and "notes" in parsed:
if VERBOSE_DEBUG:
logger.debug(f"[REFLECTION] Parsed {len(parsed['notes'])} notes from JSON")
return parsed
except:
if VERBOSE_DEBUG:
logger.debug("[REFLECTION] Direct JSON parsing failed, trying extraction...")
# -----------------------------
# Try JSON extraction
# -----------------------------
try:
match = re.search(r"\{.*?\}", raw, re.S)
if match:
parsed = json.loads(match.group(0))
if isinstance(parsed, dict) and "notes" in parsed:
return parsed
except:
pass
# -----------------------------
# Fallback — treat raw text as a single note
# -----------------------------
return {"notes": [raw.strip()]}

1
cortex/tests/__init__.py
View File

@@ -1 +0,0 @@
"""Tests for Project Lyra Cortex."""

197
cortex/tests/test_autonomy_phase1.py
View File

@@ -1,197 +0,0 @@
"""
Integration tests for Phase 1 autonomy features.
Tests monologue integration, executive planning, and self-state persistence.
"""
import asyncio
import json
import sys
import os
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from autonomy.monologue.monologue import InnerMonologue
from autonomy.self.state import load_self_state, update_self_state, get_self_state_instance
from autonomy.executive.planner import plan_execution
async def test_monologue_integration():
"""Test monologue generates valid output."""
print("\n" + "="*60)
print("TEST 1: Monologue Integration")
print("="*60)
mono = InnerMonologue()
context = {
"user_message": "Explain quantum computing to me like I'm 5",
"session_id": "test_001",
"self_state": load_self_state(),
"context_summary": {"message_count": 5}
}
result = await mono.process(context)
assert "intent" in result, "Missing intent field"
assert "tone" in result, "Missing tone field"
assert "depth" in result, "Missing depth field"
assert "consult_executive" in result, "Missing consult_executive field"
print("✓ Monologue integration test passed")
print(f" Result: {json.dumps(result, indent=2)}")
return result
async def test_executive_planning():
"""Test executive planner generates valid plans."""
print("\n" + "="*60)
print("TEST 2: Executive Planning")
print("="*60)
plan = await plan_execution(
user_prompt="Help me build a distributed system with microservices architecture",
intent="technical_implementation",
context_state={
"tools_available": ["RAG", "WEB", "CODEBRAIN"],
"message_count": 3,
"minutes_since_last_msg": 2.5,
"active_project": None
},
identity_block={}
)
assert "summary" in plan, "Missing summary field"
assert "plan_text" in plan, "Missing plan_text field"
assert "steps" in plan, "Missing steps field"
assert len(plan["steps"]) > 0, "No steps generated"
print("✓ Executive planning test passed")
print(f" Plan summary: {plan['summary']}")
print(f" Steps: {len(plan['steps'])}")
print(f" Complexity: {plan.get('estimated_complexity', 'unknown')}")
return plan
def test_self_state_persistence():
"""Test self-state loads and updates."""
print("\n" + "="*60)
print("TEST 3: Self-State Persistence")
print("="*60)
state1 = load_self_state()
assert "mood" in state1, "Missing mood field"
assert "energy" in state1, "Missing energy field"
assert "interaction_count" in state1, "Missing interaction_count"
initial_count = state1.get("interaction_count", 0)
print(f" Initial interaction count: {initial_count}")
update_self_state(
mood_delta=0.1,
energy_delta=-0.05,
new_focus="testing"
)
state2 = load_self_state()
assert state2["interaction_count"] == initial_count + 1, "Interaction count not incremented"
assert state2["focus"] == "testing", "Focus not updated"
print("✓ Self-state persistence test passed")
print(f" New interaction count: {state2['interaction_count']}")
print(f" New focus: {state2['focus']}")
print(f" New energy: {state2['energy']:.2f}")
return state2
async def test_end_to_end_flow():
"""Test complete flow from monologue through planning."""
print("\n" + "="*60)
print("TEST 4: End-to-End Flow")
print("="*60)
# Step 1: Monologue detects complex query
mono = InnerMonologue()
mono_result = await mono.process({
"user_message": "Design a scalable ML pipeline with CI/CD integration",
"session_id": "test_e2e",
"self_state": load_self_state(),
"context_summary": {}
})
print(f" Monologue intent: {mono_result.get('intent')}")
print(f" Consult executive: {mono_result.get('consult_executive')}")
# Step 2: If executive requested, generate plan
if mono_result.get("consult_executive"):
plan = await plan_execution(
user_prompt="Design a scalable ML pipeline with CI/CD integration",
intent=mono_result.get("intent", "unknown"),
context_state={"tools_available": ["CODEBRAIN", "WEB"]},
identity_block={}
)
assert plan is not None, "Plan should be generated"
print(f" Executive plan generated: {len(plan.get('steps', []))} steps")
# Step 3: Update self-state
update_self_state(
energy_delta=-0.1, # Complex task is tiring
new_focus="ml_pipeline_design",
confidence_delta=0.05
)
state = load_self_state()
assert state["focus"] == "ml_pipeline_design", "Focus should be updated"
print("✓ End-to-end flow test passed")
print(f" Final state: {state['mood']}, energy={state['energy']:.2f}")
return True
async def run_all_tests():
"""Run all Phase 1 tests."""
print("\n" + "="*60)
print("PHASE 1 AUTONOMY TESTS")
print("="*60)
try:
# Test 1: Monologue
mono_result = await test_monologue_integration()
# Test 2: Executive Planning
plan_result = await test_executive_planning()
# Test 3: Self-State
state_result = test_self_state_persistence()
# Test 4: End-to-End
await test_end_to_end_flow()
print("\n" + "="*60)
print("ALL TESTS PASSED ✓")
print("="*60)
print("\nSummary:")
print(f" - Monologue: {mono_result.get('intent')} ({mono_result.get('tone')})")
print(f" - Executive: {plan_result.get('estimated_complexity')} complexity")
print(f" - Self-state: {state_result.get('interaction_count')} interactions")
return True
except Exception as e:
print("\n" + "="*60)
print(f"TEST FAILED: {e}")
print("="*60)
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
success = asyncio.run(run_all_tests())
sys.exit(0 if success else 1)

495
cortex/tests/test_autonomy_phase2.py
View File

@@ -1,495 +0,0 @@
"""
Integration tests for Phase 2 autonomy features.
Tests autonomous tool invocation, proactive monitoring, actions, and pattern learning.
"""
import asyncio
import json
import sys
import os
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Override self-state file path for testing
os.environ["SELF_STATE_FILE"] = "/tmp/test_self_state.json"
from autonomy.tools.decision_engine import ToolDecisionEngine
from autonomy.tools.orchestrator import ToolOrchestrator
from autonomy.proactive.monitor import ProactiveMonitor
from autonomy.actions.autonomous_actions import AutonomousActionManager
from autonomy.learning.pattern_learner import PatternLearner
from autonomy.self.state import load_self_state, get_self_state_instance
async def test_tool_decision_engine():
"""Test autonomous tool decision making."""
print("\n" + "="*60)
print("TEST 1: Tool Decision Engine")
print("="*60)
engine = ToolDecisionEngine()
# Test 1a: Memory reference detection
result = await engine.analyze_tool_needs(
user_prompt="What did we discuss earlier about Python?",
monologue={"intent": "clarification", "consult_executive": False},
context_state={},
available_tools=["RAG", "WEB", "WEATHER"]
)
assert result["should_invoke_tools"], "Should invoke tools for memory reference"
assert any(t["tool"] == "RAG" for t in result["tools_to_invoke"]), "Should recommend RAG"
assert result["confidence"] > 0.8, f"Confidence should be high for clear memory reference: {result['confidence']}"
print(f" ✓ Memory reference detection passed")
print(f" Tools: {[t['tool'] for t in result['tools_to_invoke']]}")
print(f" Confidence: {result['confidence']:.2f}")
# Test 1b: Web search detection
result = await engine.analyze_tool_needs(
user_prompt="What's the latest news about AI developments?",
monologue={"intent": "information_seeking", "consult_executive": False},
context_state={},
available_tools=["RAG", "WEB", "WEATHER"]
)
assert result["should_invoke_tools"], "Should invoke tools for current info request"
assert any(t["tool"] == "WEB" for t in result["tools_to_invoke"]), "Should recommend WEB"
print(f" ✓ Web search detection passed")
print(f" Tools: {[t['tool'] for t in result['tools_to_invoke']]}")
# Test 1c: Weather detection
result = await engine.analyze_tool_needs(
user_prompt="What's the weather like today in Boston?",
monologue={"intent": "information_seeking", "consult_executive": False},
context_state={},
available_tools=["RAG", "WEB", "WEATHER"]
)
assert result["should_invoke_tools"], "Should invoke tools for weather query"
assert any(t["tool"] == "WEATHER" for t in result["tools_to_invoke"]), "Should recommend WEATHER"
print(f" ✓ Weather detection passed")
# Test 1d: Proactive RAG for complex queries
result = await engine.analyze_tool_needs(
user_prompt="Design a microservices architecture",
monologue={"intent": "technical_implementation", "consult_executive": True},
context_state={},
available_tools=["RAG", "WEB", "CODEBRAIN"]
)
assert result["should_invoke_tools"], "Should proactively invoke tools for complex queries"
rag_tools = [t for t in result["tools_to_invoke"] if t["tool"] == "RAG"]
assert len(rag_tools) > 0, "Should include proactive RAG"
print(f" ✓ Proactive RAG detection passed")
print(f" Reason: {rag_tools[0]['reason']}")
print("\n✓ Tool Decision Engine tests passed\n")
return result
async def test_tool_orchestrator():
"""Test tool orchestration (mock mode)."""
print("\n" + "="*60)
print("TEST 2: Tool Orchestrator (Mock Mode)")
print("="*60)
orchestrator = ToolOrchestrator(tool_timeout=5)
# Since actual tools may not be available, test the orchestrator structure
print(f" Available tools: {list(orchestrator.available_tools.keys())}")
# Test with tools_to_invoke (will fail gracefully if tools unavailable)
tools_to_invoke = [
{"tool": "RAG", "query": "test query", "reason": "testing", "priority": 0.9}
]
result = await orchestrator.execute_tools(
tools_to_invoke=tools_to_invoke,
context_state={"session_id": "test"}
)
assert "results" in result, "Should return results dict"
assert "execution_summary" in result, "Should return execution summary"
summary = result["execution_summary"]
assert "tools_invoked" in summary, "Summary should include tools_invoked"
assert "total_time_ms" in summary, "Summary should include timing"
print(f" ✓ Orchestrator structure valid")
print(f" Summary: {summary}")
# Test result formatting
formatted = orchestrator.format_results_for_context(result)
assert isinstance(formatted, str), "Should format results as string"
print(f" ✓ Result formatting works")
print(f" Formatted length: {len(formatted)} chars")
print("\n✓ Tool Orchestrator tests passed\n")
return result
async def test_proactive_monitor():
"""Test proactive monitoring and suggestions."""
print("\n" + "="*60)
print("TEST 3: Proactive Monitor")
print("="*60)
monitor = ProactiveMonitor(min_priority=0.6)
# Test 3a: Long silence detection
context_state = {
"message_count": 5,
"minutes_since_last_msg": 35 # > 30 minutes
}
self_state = load_self_state()
suggestion = await monitor.analyze_session(
session_id="test_silence",
context_state=context_state,
self_state=self_state
)
assert suggestion is not None, "Should generate suggestion for long silence"
assert suggestion["type"] == "check_in", f"Should be check_in type: {suggestion['type']}"
assert suggestion["priority"] >= 0.6, "Priority should meet threshold"
print(f" ✓ Long silence detection passed")
print(f" Type: {suggestion['type']}, Priority: {suggestion['priority']:.2f}")
print(f" Suggestion: {suggestion['suggestion'][:50]}...")
# Test 3b: Learning opportunity (high curiosity)
self_state["curiosity"] = 0.8
self_state["learning_queue"] = ["quantum computing", "rust programming"]
# Reset cooldown for this test
monitor.reset_cooldown("test_learning")
suggestion = await monitor.analyze_session(
session_id="test_learning",
context_state={"message_count": 3, "minutes_since_last_msg": 2},
self_state=self_state
)
assert suggestion is not None, "Should generate learning suggestion"
assert suggestion["type"] == "learning", f"Should be learning type: {suggestion['type']}"
print(f" ✓ Learning opportunity detection passed")
print(f" Suggestion: {suggestion['suggestion'][:70]}...")
# Test 3c: Conversation milestone
monitor.reset_cooldown("test_milestone")
# Reset curiosity to avoid learning suggestion taking precedence
self_state["curiosity"] = 0.5
self_state["learning_queue"] = []
suggestion = await monitor.analyze_session(
session_id="test_milestone",
context_state={"message_count": 50, "minutes_since_last_msg": 1},
self_state=self_state
)
assert suggestion is not None, "Should generate milestone suggestion"
# Note: learning or summary both valid - check it's a reasonable suggestion
assert suggestion["type"] in ["summary", "learning", "check_in"], f"Should be valid type: {suggestion['type']}"
print(f" ✓ Conversation milestone detection passed (type: {suggestion['type']})")
# Test 3d: Cooldown mechanism
# Try to get another suggestion immediately (should be blocked)
suggestion2 = await monitor.analyze_session(
session_id="test_milestone",
context_state={"message_count": 51, "minutes_since_last_msg": 1},
self_state=self_state
)
assert suggestion2 is None, "Should not generate suggestion during cooldown"
print(f" ✓ Cooldown mechanism working")
# Check stats
stats = monitor.get_session_stats("test_milestone")
assert stats["cooldown_active"], "Cooldown should be active"
print(f" Cooldown remaining: {stats['cooldown_remaining']}s")
print("\n✓ Proactive Monitor tests passed\n")
return suggestion
async def test_autonomous_actions():
"""Test autonomous action execution."""
print("\n" + "="*60)
print("TEST 4: Autonomous Actions")
print("="*60)
manager = AutonomousActionManager()
# Test 4a: List allowed actions
allowed = manager.get_allowed_actions()
assert "create_memory" in allowed, "Should have create_memory action"
assert "update_goal" in allowed, "Should have update_goal action"
assert "learn_topic" in allowed, "Should have learn_topic action"
print(f" ✓ Allowed actions: {allowed}")
# Test 4b: Validate actions
validation = manager.validate_action("create_memory", {"text": "test memory"})
assert validation["valid"], "Should validate correct action"
print(f" ✓ Action validation passed")
# Test 4c: Execute learn_topic action
result = await manager.execute_action(
action_type="learn_topic",
parameters={"topic": "rust programming", "reason": "testing", "priority": 0.8},
context={"session_id": "test"}
)
assert result["success"], f"Action should succeed: {result.get('error', 'unknown')}"
assert "topic" in result["result"], "Should return topic info"
print(f" ✓ learn_topic action executed")
print(f" Topic: {result['result']['topic']}")
print(f" Queue position: {result['result']['queue_position']}")
# Test 4d: Execute update_focus action
result = await manager.execute_action(
action_type="update_focus",
parameters={"focus": "autonomy_testing", "reason": "running tests"},
context={"session_id": "test"}
)
assert result["success"], "update_focus should succeed"
print(f" ✓ update_focus action executed")
print(f" New focus: {result['result']['new_focus']}")
# Test 4e: Reject non-whitelisted action
result = await manager.execute_action(
action_type="delete_all_files", # NOT in whitelist
parameters={},
context={"session_id": "test"}
)
assert not result["success"], "Should reject non-whitelisted action"
assert "not in whitelist" in result["error"], "Should indicate whitelist violation"
print(f" ✓ Non-whitelisted action rejected")
# Test 4f: Action log
log = manager.get_action_log(limit=10)
assert len(log) >= 2, f"Should have logged multiple actions (got {len(log)})"
print(f" ✓ Action log contains {len(log)} entries")
print("\n✓ Autonomous Actions tests passed\n")
return result
async def test_pattern_learner():
"""Test pattern learning system."""
print("\n" + "="*60)
print("TEST 5: Pattern Learner")
print("="*60)
# Use temp file for testing
test_file = "/tmp/test_patterns.json"
learner = PatternLearner(patterns_file=test_file)
# Test 5a: Learn from multiple interactions
for i in range(5):
await learner.learn_from_interaction(
user_prompt=f"Help me with Python coding task {i}",
response=f"Here's help with task {i}...",
monologue={"intent": "coding_help", "tone": "focused", "depth": "medium"},
context={"session_id": "test", "executive_plan": None}
)
print(f" ✓ Learned from 5 interactions")
# Test 5b: Get top topics
top_topics = learner.get_top_topics(limit=5)
assert len(top_topics) > 0, "Should have learned topics"
assert "coding_help" == top_topics[0][0], "coding_help should be top topic"
print(f" ✓ Top topics: {[t[0] for t in top_topics[:3]]}")
# Test 5c: Get preferred tone
preferred_tone = learner.get_preferred_tone()
assert preferred_tone == "focused", "Should detect focused as preferred tone"
print(f" ✓ Preferred tone: {preferred_tone}")
# Test 5d: Get preferred depth
preferred_depth = learner.get_preferred_depth()
assert preferred_depth == "medium", "Should detect medium as preferred depth"
print(f" ✓ Preferred depth: {preferred_depth}")
# Test 5e: Get insights
insights = learner.get_insights()
assert insights["total_interactions"] == 5, "Should track interaction count"
assert insights["preferred_tone"] == "focused", "Insights should include tone"
print(f" ✓ Insights generated:")
print(f" Total interactions: {insights['total_interactions']}")
print(f" Recommendations: {insights['learning_recommendations']}")
# Test 5f: Export patterns
exported = learner.export_patterns()
assert "topic_frequencies" in exported, "Should export all patterns"
print(f" ✓ Patterns exported ({len(exported)} keys)")
# Cleanup
if os.path.exists(test_file):
os.remove(test_file)
print("\n✓ Pattern Learner tests passed\n")
return insights
async def test_end_to_end_autonomy():
"""Test complete autonomous flow."""
print("\n" + "="*60)
print("TEST 6: End-to-End Autonomy Flow")
print("="*60)
# Simulate a complex user query that triggers multiple autonomous systems
user_prompt = "Remember what we discussed about machine learning? I need current research on transformers."
monologue = {
"intent": "technical_research",
"tone": "focused",
"depth": "deep",
"consult_executive": True
}
context_state = {
"session_id": "e2e_test",
"message_count": 15,
"minutes_since_last_msg": 5
}
print(f" User prompt: {user_prompt}")
print(f" Monologue intent: {monologue['intent']}")
# Step 1: Tool decision engine
engine = ToolDecisionEngine()
tool_decision = await engine.analyze_tool_needs(
user_prompt=user_prompt,
monologue=monologue,
context_state=context_state,
available_tools=["RAG", "WEB", "CODEBRAIN"]
)
print(f"\n Step 1: Tool Decision")
print(f" Should invoke: {tool_decision['should_invoke_tools']}")
print(f" Tools: {[t['tool'] for t in tool_decision['tools_to_invoke']]}")
assert tool_decision["should_invoke_tools"], "Should invoke tools"
assert len(tool_decision["tools_to_invoke"]) >= 2, "Should recommend multiple tools (RAG + WEB)"
# Step 2: Pattern learning
learner = PatternLearner(patterns_file="/tmp/e2e_test_patterns.json")
await learner.learn_from_interaction(
user_prompt=user_prompt,
response="Here's information about transformers...",
monologue=monologue,
context=context_state
)
print(f"\n Step 2: Pattern Learning")
top_topics = learner.get_top_topics(limit=3)
print(f" Learned topics: {[t[0] for t in top_topics]}")
# Step 3: Autonomous action
action_manager = AutonomousActionManager()
action_result = await action_manager.execute_action(
action_type="learn_topic",
parameters={"topic": "transformer architectures", "reason": "user interest detected"},
context=context_state
)
print(f"\n Step 3: Autonomous Action")
print(f" Action: learn_topic")
print(f" Success: {action_result['success']}")
# Step 4: Proactive monitoring (won't trigger due to low message count)
monitor = ProactiveMonitor(min_priority=0.6)
monitor.reset_cooldown("e2e_test")
suggestion = await monitor.analyze_session(
session_id="e2e_test",
context_state=context_state,
self_state=load_self_state()
)
print(f"\n Step 4: Proactive Monitoring")
print(f" Suggestion: {suggestion['type'] if suggestion else 'None (expected for low message count)'}")
# Cleanup
if os.path.exists("/tmp/e2e_test_patterns.json"):
os.remove("/tmp/e2e_test_patterns.json")
print("\n✓ End-to-End Autonomy Flow tests passed\n")
return True
async def run_all_tests():
"""Run all Phase 2 tests."""
print("\n" + "="*60)
print("PHASE 2 AUTONOMY TESTS")
print("="*60)
try:
# Test 1: Tool Decision Engine
await test_tool_decision_engine()
# Test 2: Tool Orchestrator
await test_tool_orchestrator()
# Test 3: Proactive Monitor
await test_proactive_monitor()
# Test 4: Autonomous Actions
await test_autonomous_actions()
# Test 5: Pattern Learner
await test_pattern_learner()
# Test 6: End-to-End
await test_end_to_end_autonomy()
print("\n" + "="*60)
print("ALL PHASE 2 TESTS PASSED ✓")
print("="*60)
print("\nPhase 2 Features Validated:")
print(" ✓ Autonomous tool decision making")
print(" ✓ Tool orchestration and execution")
print(" ✓ Proactive monitoring and suggestions")
print(" ✓ Safe autonomous actions")
print(" ✓ Pattern learning and adaptation")
print(" ✓ End-to-end autonomous flow")
return True
except Exception as e:
print("\n" + "="*60)
print(f"TEST FAILED: {e}")
print("="*60)
import traceback
traceback.print_exc()
return False
if __name__ == "__main__":
success = asyncio.run(run_all_tests())
sys.exit(0 if success else 1)