Compare commits

...
4 Commits
Author SHA1 Message Date
Mervin Praison 548b4afb88 Release v2.3.4 2025-12-15 16:39:58 +00:00
Mervin Praison ffa938b3df docs: add advanced rules features and new memory capabilities
- Add local override files support (PRAISON.local.md, CLAUDE.local.md) with higher priority
- Add @import syntax for including other files in rules (like Claude Code)
- Add git root discovery for monorepo support with fallback to subprocess
- Add auto-discovery of modular rules directories (.claude/rules/, .windsurf/rules/, .cursor/rules/)
- Add character limits (12000) for context window management with truncation
- Add AutoMemory for
2025-12-15 16:39:04 +00:00
Mervin Praison 8a9beed264 docs: add advanced memory features and rules management system
- Add session save/resume, context compression, and checkpointing (inspired by Gemini CLI)
- Add RulesManager for auto-discovery of instruction files (CLAUDE.md, AGENTS.md, GEMINI.md, .cursorrules, .windsurfrules)
- Add slash commands for memory management (/memory show, /memory save, /memory compress, etc.)
- Add rules context injection into agent system prompts with activation modes (always, glob, manual, ai_decision)
- Add program
2025-12-15 16:17:25 +00:00
Mervin Praison daca5a1762 test: add multi-agent memory sharing and integration tests
- Add TestMultiAgentMemorySharing class with tests for shared/isolated memory between agents
- Test agents with same user_id share memory (short_term, long_term, entities)
- Test agents with different user_id have isolated memory
- Test memory context injection into system prompt
- Test memory display info shows correct counts
- Add TestMultiAgentMemoryIntegration class for LLM integration tests
- Test multi-agent memory sharing with OpenA
2025-12-15 15:00:16 +00:00
25 changed files with 4996 additions and 48 deletions

No files matched your search

+123 -11
View File
@@ -178,10 +178,11 @@ result = agent.rewrite("What about cost?", chat_history=[...])
### 5. Agent Memory (Zero Dependencies)
Enable persistent memory for agents - works out of the box without any extra packages.
Enable persistent memory for agents - works out of the box without any extra packages. Features inspired by Claude, Gemini CLI, Codex CLI, Cursor, and Windsurf.
```python
from praisonaiagents import Agent
from praisonaiagents.memory import FileMemory
# Enable memory with a single parameter
agent = Agent(
@@ -191,17 +192,9 @@ agent = Agent(
user_id="user123" # Isolate memory per user
)
# Store memories programmatically
agent.store_memory("User prefers dark mode", memory_type="short_term")
agent.store_memory("User's name is John", memory_type="long_term", importance=0.9)
# Memory is automatically injected into conversations
result = agent.start("What do you know about me?")
# Agent will recall: "Your name is John and you prefer dark mode"
# Memory persists across sessions - create new agent with same user_id
agent2 = Agent(name="Assistant", memory=True, user_id="user123")
# agent2 automatically loads all previous memories!
result = agent.start("My name is John and I prefer Python")
# Agent will remember this for future conversations
```
**Memory Types:**
@@ -210,6 +203,28 @@ agent2 = Agent(name="Assistant", memory=True, user_id="user123")
- **Entity**: People, places, organizations with attributes
- **Episodic**: Date-based interaction history
**Advanced Features (like Gemini CLI):**
```python
from praisonaiagents.memory import FileMemory
memory = FileMemory(user_id="user123")
# Session Save/Resume
memory.save_session("project_session", conversation_history=[...])
memory.resume_session("project_session")
# Context Compression
memory.compress(llm_func=lambda p: agent.chat(p), max_items=10)
# Checkpointing
memory.create_checkpoint("before_refactor", include_files=["main.py"])
memory.restore_checkpoint("before_refactor", restore_files=True)
# Slash Commands
memory.handle_command("/memory show")
memory.handle_command("/memory save my_session")
```
**Storage Options:**
| Option | Dependencies | Description |
|--------|-------------|-------------|
@@ -218,6 +233,103 @@ agent2 = Agent(name="Assistant", memory=True, user_id="user123")
| `memory="sqlite"` | Built-in | SQLite with indexing |
| `memory="chromadb"` | chromadb | Vector/semantic search |
### 6. Rules & Instructions (like Cursor/Windsurf)
PraisonAI auto-discovers instruction files from your project root and git root:
| File | Description | Priority |
|------|-------------|----------|
| `PRAISON.md` | PraisonAI native instructions | High |
| `PRAISON.local.md` | Local overrides (gitignored) | Higher |
| `CLAUDE.md` | Claude Code memory file | High |
| `CLAUDE.local.md` | Local overrides (gitignored) | Higher |
| `AGENTS.md` | OpenAI Codex CLI instructions | High |
| `GEMINI.md` | Gemini CLI memory file | High |
| `.cursorrules` | Cursor IDE rules | High |
| `.windsurfrules` | Windsurf IDE rules | High |
| `.claude/rules/*.md` | Claude Code modular rules | Medium |
| `.windsurf/rules/*.md` | Windsurf modular rules | Medium |
| `.cursor/rules/*.mdc` | Cursor modular rules | Medium |
| `.praison/rules/*.md` | Workspace rules | Medium |
| `~/.praison/rules/*.md` | Global rules | Low |
```python
from praisonaiagents import Agent
# Agent auto-discovers CLAUDE.md, AGENTS.md, GEMINI.md, etc.
agent = Agent(name="Assistant", instructions="You are helpful.")
# Rules are injected into system prompt automatically
```
**@Import Syntax (like Claude Code):**
```markdown
# CLAUDE.md
See @README for project overview
See @docs/architecture.md for system design
@~/.praison/my-preferences.md
```
**Rule File Format (with YAML frontmatter):**
```markdown
---
description: Python coding guidelines
globs: ["**/*.py"]
activation: always # always, glob, manual, ai_decision
---
# Guidelines
- Use type hints
- Follow PEP 8
```
### 7. Auto-Generated Memories (like Windsurf Cascade)
```python
from praisonaiagents.memory import FileMemory, AutoMemory
memory = FileMemory(user_id="user123")
auto = AutoMemory(memory, enabled=True)
# Automatically extracts and stores memories from conversations
memories = auto.process_interaction(
"My name is John and I prefer Python for backend work"
)
# Extracts: name="John", preference="Python for backend"
```
### 8. Workflows (like Windsurf)
Create reusable multi-step workflows in `.praison/workflows/`:
```python
from praisonaiagents.memory import WorkflowManager
manager = WorkflowManager()
# Execute a workflow
result = manager.execute(
"deploy",
executor=lambda prompt: agent.chat(prompt),
variables={"environment": "production"}
)
```
### 9. Hooks (like Windsurf Cascade Hooks)
Configure in `.praison/hooks.json`:
```python
from praisonaiagents.memory import HooksManager
hooks = HooksManager()
# Register Python hooks
hooks.register("pre_write_code", lambda ctx: print(f"Writing {ctx['file']}"))
# Execute hooks
result = hooks.execute("pre_write_code", {"file": "main.py"})
```
## Using No Code
### Auto Mode:
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
flask \
"praisonai>=2.3.3" \
"praisonai>=2.3.4" \
"praisonai[api]" \
gunicorn \
markdown
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.3" \
"praisonai>=2.3.4" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.3" \
"praisonai>=2.3.4" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=2.3.3" \
"praisonai>=2.3.4" \
"praisonai[ui]" \
"praisonai[crewai]"
+289 -9
View File
@@ -216,27 +216,307 @@ router = RouterAgent(
## Memory
Persistent memory with short-term, long-term, and graph memory support.
PraisonAI provides a comprehensive memory system with features inspired by Claude, Gemini CLI, Codex CLI, Cursor, and Windsurf.
### Quick Start (Zero Dependencies)
```python
from praisonaiagents import Agent, Memory
from praisonaiagents import Agent
from praisonaiagents.memory import FileMemory
# Simple file-based memory (no extra dependencies)
memory = FileMemory(user_id="user123")
agent = Agent(
name="Memory Agent",
instructions="You are a helpful assistant with memory.",
memory=memory # or memory=True for auto FileMemory
)
response = agent.chat("My name is John and I prefer Python")
# Memory automatically stores this context
```
### Advanced Memory (with providers)
```python
from praisonaiagents.memory import Memory
memory = Memory(
provider="chroma", # or "mongodb", "rag"
provider="chroma", # or "mongodb", "mem0", "rag"
use_short_term=True,
use_long_term=True
)
agent = Agent(name="Memory Agent", memory=memory)
```
### Memory Types
| Type | Description | Storage |
|------|-------------|---------|
| **Short-term** | Rolling buffer of recent context | JSON/SQLite |
| **Long-term** | Persistent facts and knowledge | JSON/SQLite/Vector DB |
| **Entity** | Named entities (people, places, concepts) | JSON/SQLite |
| **Episodic** | Date-based interaction memories | JSON files |
| **Graph** | Relationship-based memory | Mem0/Neo4j |
### Session Save/Resume (like Gemini CLI)
```python
from praisonaiagents.memory import FileMemory
memory = FileMemory(user_id="user123")
# Add context during conversation
memory.add_short_term("User is working on ML project")
memory.add_long_term("User prefers Python", importance=0.9)
# Save session for later
memory.save_session("ml_project", conversation_history=[...])
# Resume later
session_data = memory.resume_session("ml_project")
# List all sessions
sessions = memory.list_sessions()
```
### Context Compression (like Gemini CLI)
```python
# Auto-compress when memory gets full
memory.auto_compress_if_needed(threshold_percent=0.7)
# Manual compression with LLM summarization
def llm_summarize(prompt):
return agent.chat(prompt)
summary = memory.compress(llm_func=llm_summarize, max_items=10)
```
### Checkpointing (like Gemini CLI)
```python
# Create checkpoint before risky operations
checkpoint_id = memory.create_checkpoint("before_refactor", include_files=["main.py"])
# Restore if needed
memory.restore_checkpoint(checkpoint_id, restore_files=True)
# List checkpoints
checkpoints = memory.list_checkpoints()
```
### Memory Slash Commands
```python
# Handle slash commands programmatically
result = memory.handle_command("/memory show")
result = memory.handle_command("/memory add User likes coffee")
result = memory.handle_command("/memory search Python")
result = memory.handle_command("/memory save my_session")
result = memory.handle_command("/memory compress")
result = memory.handle_command("/memory checkpoint")
result = memory.handle_command("/memory help")
```
**Available Commands:**
- `/memory show` - Display stats and recent items
- `/memory add <content>` - Add to long-term memory
- `/memory clear [short|all]` - Clear memory
- `/memory search <query>` - Search memories
- `/memory save <name>` - Save session
- `/memory resume <name>` - Resume session
- `/memory sessions` - List saved sessions
- `/memory compress` - Compress short-term memory
- `/memory checkpoint [name]` - Create checkpoint
- `/memory restore <id>` - Restore checkpoint
- `/memory checkpoints` - List checkpoints
- `/memory refresh` - Reload from disk
---
## Rules & Instructions (like Cursor/Windsurf)
PraisonAI automatically discovers and applies rules from multiple sources, similar to Cursor, Windsurf, Claude Code, and Codex CLI.
### Supported Instruction Files
| File | Description | Priority |
|------|-------------|----------|
| `PRAISON.md` | PraisonAI native instructions | High (500) |
| `PRAISON.local.md` | Local overrides (gitignored) | Higher (600) |
| `CLAUDE.md` | Claude Code memory file | High (500) |
| `CLAUDE.local.md` | Local overrides (gitignored) | Higher (600) |
| `AGENTS.md` | OpenAI Codex CLI instructions | High (500) |
| `GEMINI.md` | Gemini CLI memory file | High (500) |
| `.cursorrules` | Cursor IDE rules (legacy) | High (500) |
| `.windsurfrules` | Windsurf IDE rules (legacy) | High (500) |
| `.claude/rules/*.md` | Claude Code modular rules | Medium (50) |
| `.windsurf/rules/*.md` | Windsurf modular rules | Medium (50) |
| `.cursor/rules/*.mdc` | Cursor modular rules | Medium (50) |
| `.praison/rules/*.md` | Workspace rules | Medium (0) |
| `~/.praison/rules/*.md` | Global rules | Low (-1000) |
### Auto-Discovery
Rules are automatically loaded when you create an Agent:
```python
from praisonaiagents import Agent
# Agent auto-discovers CLAUDE.md, AGENTS.md, GEMINI.md, etc.
agent = Agent(
name="Memory Agent",
memory=memory
name="Assistant",
instructions="You are helpful."
)
# Rules are injected into system prompt automatically
```
### Rule File Format
Rules support YAML frontmatter for advanced configuration:
```markdown
---
description: Python coding guidelines
globs: ["**/*.py", "**/*.pyx"]
activation: always # always, glob, manual, ai_decision
priority: 10
---
# Python Guidelines
- Use type hints for all functions
- Follow PEP 8 style guide
- Write docstrings for public APIs
```
### Activation Modes
| Mode | Description |
|------|-------------|
| `always` | Always applied (default) |
| `glob` | Applied when file matches glob pattern |
| `manual` | Only when explicitly invoked via @mention |
| `ai_decision` | AI decides when to apply |
### Programmatic Rules Management
```python
from praisonaiagents.memory import RulesManager
rules = RulesManager(workspace_path="/path/to/project")
# Get all active rules
active = rules.get_active_rules()
# Get rules for specific file
python_rules = rules.get_rules_for_file("src/main.py")
# Build context for LLM
context = rules.build_rules_context(file_path="src/main.py")
# Create new rule
rules.create_rule(
name="testing",
content="Always write tests first",
globs=["**/*.test.*"],
activation="glob"
)
# Get stats
stats = rules.get_stats()
# {'total_rules': 5, 'root_rules': 3, 'workspace_rules': 2, ...}
```
### Storage Structure
```
project/
├── CLAUDE.md # Auto-loaded (Claude Code)
├── CLAUDE.local.md # Local overrides (gitignored)
├── AGENTS.md # Auto-loaded (Codex CLI)
├── GEMINI.md # Auto-loaded (Gemini CLI)
├── PRAISON.md # Auto-loaded (PraisonAI)
├── PRAISON.local.md # Local overrides (gitignored)
├── .claude/rules/ # Claude Code modular rules
├── .windsurf/rules/ # Windsurf rules
├── .cursor/rules/ # Cursor rules
├── .praison/
│ ├── rules/ # Workspace rules
│ │ ├── python.md
│ │ └── testing.md
│ ├── workflows/ # Reusable workflows
│ │ └── deploy.md
│ ├── hooks.json # Pre/post operation hooks
│ └── memory/
│ └── {user_id}/
│ ├── short_term.json
│ ├── long_term.json
│ ├── sessions/
│ └── checkpoints/
└── ~/.praison/
└── rules/ # Global rules
└── global.md
```
### @Import Syntax (like Claude Code)
Include other files in your rules:
```markdown
# CLAUDE.md
See @README for project overview
See @docs/architecture.md for system design
@~/.praison/my-preferences.md
```
### Auto-Generated Memories (like Windsurf Cascade)
```python
from praisonaiagents.memory import FileMemory, AutoMemory
memory = FileMemory(user_id="user123")
auto = AutoMemory(memory, enabled=True)
# Automatically extracts and stores memories from conversations
memories = auto.process_interaction(
"My name is John and I prefer Python for backend work"
)
# Extracts: name="John", preference="Python for backend"
```
### Workflows (like Windsurf)
Create reusable multi-step workflows:
```python
from praisonaiagents.memory import WorkflowManager
manager = WorkflowManager()
# Execute a workflow
result = manager.execute(
"deploy",
executor=lambda prompt: agent.chat(prompt),
variables={"environment": "production"}
)
```
**Memory Types:**
- **Short-term**: Recent conversation context
- **Long-term**: Persistent storage across sessions
- **Graph**: Relationship-based memory with Mem0
### Hooks (like Windsurf Cascade Hooks)
```python
from praisonaiagents.memory import HooksManager
hooks = HooksManager()
# Register Python hooks
hooks.register("pre_write_code", lambda ctx: print(f"Writing {ctx['file']}"))
# Or configure in .praison/hooks.json
```
---
@@ -529,6 +529,10 @@ Your Goal: {self.goal}
self.prompt_caching = prompt_caching
self.claude_memory = claude_memory
# Initialize rules manager for persistent context (like Cursor/Windsurf)
self._rules_manager = None
self._init_rules_manager()
# Handle web_search fallback: inject DuckDuckGo tool for unsupported models
if web_search and not self._model_supports_web_search():
from ..tools.duckduckgo_tools import internet_search
@@ -669,6 +673,57 @@ Your Goal: {self.goal}
return supports_prompt_caching(model_name)
def _init_rules_manager(self):
"""
Initialize RulesManager for persistent rules/instructions.
Automatically discovers rules from:
- ~/.praison/rules/ (global)
- .praison/rules/ (workspace)
- Subdirectory rules
"""
try:
from ..memory.rules_manager import RulesManager
import os
# Get workspace path (current working directory)
workspace_path = os.getcwd()
self._rules_manager = RulesManager(
workspace_path=workspace_path,
verbose=1 if self.verbose else 0
)
# Log discovered rules
stats = self._rules_manager.get_stats()
if stats["total_rules"] > 0:
logging.debug(f"RulesManager: Discovered {stats['total_rules']} rules")
except ImportError:
logging.debug("RulesManager not available")
self._rules_manager = None
except Exception as e:
logging.debug(f"Could not initialize RulesManager: {e}")
self._rules_manager = None
def get_rules_context(self, file_path: Optional[str] = None, include_manual: Optional[List[str]] = None) -> str:
"""
Get rules context for the current conversation.
Args:
file_path: Optional file path for glob-based rule matching
include_manual: Optional list of manual rule names to include (via @mention)
Returns:
Formatted rules context string
"""
if not self._rules_manager:
return ""
return self._rules_manager.build_rules_context(
file_path=file_path,
include_manual=include_manual
)
def _init_memory(self, memory, user_id: Optional[str] = None):
"""
Initialize memory based on the memory parameter.
@@ -1072,6 +1127,12 @@ Your Goal: {self.goal}
Your Role: {self.role}\n
Your Goal: {self.goal}"""
# Add rules context if rules manager is enabled
if self._rules_manager:
rules_context = self.get_rules_context()
if rules_context:
system_prompt += f"\n\n## Rules (Guidelines you must follow)\n{rules_context}"
# Add memory context if memory is enabled
if self._memory_instance:
memory_context = self.get_memory_context()
@@ -8,19 +8,83 @@ This module provides memory management capabilities including:
- User memory for preferences/history
- Quality-based storage decisions
- Graph memory support via Mem0
- Session save/resume (like Gemini CLI)
- Context compression (like Gemini CLI)
- Checkpointing (like Gemini CLI)
- Rules management (like Cursor/Windsurf)
- Auto-generated memories (like Windsurf Cascade)
- Workflows (like Windsurf)
- Hooks (like Windsurf Cascade Hooks)
Memory Providers:
- FileMemory: Zero-dependency JSON file-based storage (default)
- Memory: Full-featured with SQLite, ChromaDB, Mem0, MongoDB support
- RulesManager: Rules/instructions file management
- AutoMemory: Automatic memory extraction from conversations
- WorkflowManager: Multi-step workflow execution
- HooksManager: Pre/post operation hooks
"""
from .file_memory import FileMemory, create_memory
from .rules_manager import RulesManager, Rule, create_rules_manager
# Lazy import for Memory to avoid dependency issues
# Lazy imports for optional modules to avoid dependency issues and improve startup time
def __getattr__(name):
if name == "Memory":
from .memory import Memory
return Memory
if name == "AutoMemory":
from .auto_memory import AutoMemory
return AutoMemory
if name == "AutoMemoryExtractor":
from .auto_memory import AutoMemoryExtractor
return AutoMemoryExtractor
if name == "create_auto_memory":
from .auto_memory import create_auto_memory
return create_auto_memory
if name == "WorkflowManager":
from .workflows import WorkflowManager
return WorkflowManager
if name == "Workflow":
from .workflows import Workflow
return Workflow
if name == "WorkflowStep":
from .workflows import WorkflowStep
return WorkflowStep
if name == "create_workflow_manager":
from .workflows import create_workflow_manager
return create_workflow_manager
if name == "HooksManager":
from .hooks import HooksManager
return HooksManager
if name == "HookResult":
from .hooks import HookResult
return HookResult
if name == "create_hooks_manager":
from .hooks import create_hooks_manager
return create_hooks_manager
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
__all__ = ["FileMemory", "Memory", "create_memory"]
__all__ = [
# Core memory
"FileMemory",
"Memory",
"create_memory",
# Rules management
"RulesManager",
"Rule",
"create_rules_manager",
# Auto memory
"AutoMemory",
"AutoMemoryExtractor",
"create_auto_memory",
# Workflows
"WorkflowManager",
"Workflow",
"WorkflowStep",
"create_workflow_manager",
# Hooks
"HooksManager",
"HookResult",
"create_hooks_manager",
]
@@ -0,0 +1,382 @@
"""
Auto-Generated Memory System for PraisonAI Agents.
Provides automatic memory generation similar to Windsurf's Cascade,
where the AI automatically decides what context is worth remembering.
Features:
- Automatic extraction of important facts from conversations
- Entity recognition and storage
- Preference detection
- Lazy evaluation to avoid performance impact
- Configurable importance thresholds
This module is designed to be lightweight and only activates
when explicitly enabled to avoid impacting performance.
"""
import re
import logging
from typing import Any, Dict, List, Optional, Callable, TYPE_CHECKING
if TYPE_CHECKING:
from .file_memory import FileMemory
logger = logging.getLogger(__name__)
class AutoMemoryExtractor:
"""
Automatically extracts memorable content from conversations.
Uses pattern matching for fast extraction without LLM calls,
with optional LLM enhancement for better accuracy.
Example:
```python
extractor = AutoMemoryExtractor()
# Extract from conversation
memories = extractor.extract(
"My name is John and I prefer Python for backend work."
)
# Returns: [
# {"type": "entity", "name": "John", "entity_type": "person"},
# {"type": "preference", "content": "prefers Python for backend work"}
# ]
```
"""
# Patterns for fast extraction (no LLM needed)
PATTERNS = {
# Name patterns
"name": [
r"(?:my name is|i'm|i am|call me)\s+([A-Z][a-z]+)",
r"(?:name|Name):\s*([A-Z][a-z]+)",
],
# Preference patterns
"preference": [
r"i (?:prefer|like|love|enjoy|use)\s+(.+?)(?:\.|$)",
r"(?:my favorite|i always use)\s+(.+?)(?:\.|$)",
],
# Role/job patterns
"role": [
r"i (?:am|work as|'m)\s+(?:a|an)\s+([a-z\s]+?)(?:\.|,|$)",
r"(?:my job|my role|i do)\s+(?:is|as)?\s*(.+?)(?:\.|$)",
],
# Location patterns
"location": [
r"i (?:live|work|am based|am located)\s+(?:in|at)\s+([A-Z][a-z\s]+)",
r"(?:from|based in)\s+([A-Z][a-z\s,]+)",
],
# Project/task patterns
"project": [
r"(?:working on|building|developing|creating)\s+(?:a|an|the)?\s*(.+?)(?:\.|$)",
r"(?:my project|the project)\s+(?:is|called)?\s*(.+?)(?:\.|$)",
],
# Technology preferences
"technology": [
r"(?:using|use|prefer)\s+(Python|JavaScript|TypeScript|Go|Rust|Java|C\+\+|Ruby|PHP)",
r"(?:framework|library|tool)(?:s)?\s+(?:like|such as|including)\s+(.+?)(?:\.|$)",
],
}
# Importance scores by type
IMPORTANCE_SCORES = {
"name": 0.95,
"preference": 0.7,
"role": 0.85,
"location": 0.6,
"project": 0.75,
"technology": 0.7,
}
def __init__(
self,
min_importance: float = 0.5,
use_llm: bool = False,
llm_func: Optional[Callable[[str], str]] = None,
verbose: int = 0
):
"""
Initialize AutoMemoryExtractor.
Args:
min_importance: Minimum importance score to extract (0.0-1.0)
use_llm: Whether to use LLM for enhanced extraction
llm_func: LLM function for extraction (if use_llm=True)
verbose: Verbosity level
"""
self.min_importance = min_importance
self.use_llm = use_llm
self.llm_func = llm_func
self.verbose = verbose
# Compile patterns for performance
self._compiled_patterns = {}
for category, patterns in self.PATTERNS.items():
self._compiled_patterns[category] = [
re.compile(p, re.IGNORECASE) for p in patterns
]
def extract(self, text: str) -> List[Dict[str, Any]]:
"""
Extract memorable content from text.
Args:
text: Text to extract from (conversation, message, etc.)
Returns:
List of extracted memories with type, content, and importance
"""
memories = []
# Fast pattern-based extraction
for category, patterns in self._compiled_patterns.items():
importance = self.IMPORTANCE_SCORES.get(category, 0.5)
if importance < self.min_importance:
continue
for pattern in patterns:
matches = pattern.findall(text)
for match in matches:
if isinstance(match, tuple):
match = match[0]
match = match.strip()
if len(match) < 2 or len(match) > 200:
continue
memory = {
"type": category,
"content": match,
"importance": importance,
"source": "pattern"
}
# Avoid duplicates
if not any(m["content"].lower() == match.lower() for m in memories):
memories.append(memory)
# Optional LLM enhancement
if self.use_llm and self.llm_func and len(text) > 50:
llm_memories = self._extract_with_llm(text)
for mem in llm_memories:
if mem.get("importance", 0) >= self.min_importance:
if not any(m["content"].lower() == mem["content"].lower() for m in memories):
memories.append(mem)
return memories
def _extract_with_llm(self, text: str) -> List[Dict[str, Any]]:
"""Extract memories using LLM for better accuracy."""
if not self.llm_func:
return []
prompt = f"""Extract important facts from this text that should be remembered.
Return a JSON array of objects with: type (name/preference/role/location/project/technology), content, importance (0.0-1.0).
Only extract clear, specific facts. Be concise.
Text: {text[:1500]}
JSON array (no markdown, just the array):"""
try:
response = self.llm_func(prompt)
# Parse JSON from response
import json
# Find JSON array in response
start = response.find("[")
end = response.rfind("]") + 1
if start >= 0 and end > start:
return json.loads(response[start:end])
except Exception as e:
if self.verbose:
logger.warning(f"LLM extraction failed: {e}")
return []
def should_remember(self, text: str) -> bool:
"""
Quick check if text likely contains memorable content.
Useful for filtering before full extraction.
Args:
text: Text to check
Returns:
True if text likely contains memorable content
"""
# Quick keyword check
keywords = [
"my name", "i am", "i'm", "i prefer", "i like", "i use",
"i work", "i live", "working on", "building", "my project",
"favorite", "always use"
]
text_lower = text.lower()
return any(kw in text_lower for kw in keywords)
class AutoMemory:
"""
Wrapper that adds auto-generation capabilities to FileMemory.
Intercepts interactions and automatically extracts/stores memories.
Designed to be lightweight - only processes when enabled.
Example:
```python
from praisonaiagents.memory import FileMemory
base_memory = FileMemory(user_id="user123")
auto_memory = AutoMemory(base_memory, enabled=True)
# Process conversation and auto-store memories
auto_memory.process_interaction(
user_message="My name is John and I prefer Python",
assistant_response="Nice to meet you, John!"
)
```
"""
def __init__(
self,
memory: "FileMemory",
enabled: bool = True,
extractor: Optional[AutoMemoryExtractor] = None,
min_importance: float = 0.6,
verbose: int = 0
):
"""
Initialize AutoMemory wrapper.
Args:
memory: Base FileMemory instance
enabled: Whether auto-generation is enabled
extractor: Custom extractor (or use default)
min_importance: Minimum importance to auto-store
verbose: Verbosity level
"""
self.memory = memory
self.enabled = enabled
self.min_importance = min_importance
self.verbose = verbose
self.extractor = extractor or AutoMemoryExtractor(
min_importance=min_importance,
verbose=verbose
)
# Track what we've already processed to avoid duplicates
self._processed_hashes: set = set()
def process_interaction(
self,
user_message: str,
assistant_response: Optional[str] = None,
store: bool = True
) -> List[Dict[str, Any]]:
"""
Process an interaction and extract memories.
Args:
user_message: User's message
assistant_response: Optional assistant response
store: Whether to store extracted memories
Returns:
List of extracted memories
"""
if not self.enabled:
return []
# Combine messages for extraction
text = user_message
if assistant_response:
text += "\n" + assistant_response
# Check if already processed
import hashlib
text_hash = hashlib.md5(text.encode()).hexdigest()[:16]
if text_hash in self._processed_hashes:
return []
self._processed_hashes.add(text_hash)
# Quick filter
if not self.extractor.should_remember(text):
return []
# Extract memories
memories = self.extractor.extract(text)
if store and memories:
self._store_memories(memories)
return memories
def _store_memories(self, memories: List[Dict[str, Any]]):
"""Store extracted memories in the base memory."""
for mem in memories:
mem_type = mem.get("type", "")
content = mem.get("content", "")
importance = mem.get("importance", 0.5)
if not content:
continue
# Store based on type
if mem_type == "name":
# Store as entity
self.memory.add_entity(
name=content,
entity_type="person",
attributes={"relationship": "user"}
)
elif mem_type in ("role", "location"):
# Store as entity attribute
self.memory.add_long_term(
f"User's {mem_type}: {content}",
importance=importance
)
else:
# Store as long-term memory
self.memory.add_long_term(content, importance=importance)
if self.verbose:
logger.info(f"Auto-stored memory: {mem_type} - {content[:50]}...")
def enable(self):
"""Enable auto-generation."""
self.enabled = True
def disable(self):
"""Disable auto-generation."""
self.enabled = False
def set_llm(self, llm_func: Callable[[str], str]):
"""Set LLM function for enhanced extraction."""
self.extractor.use_llm = True
self.extractor.llm_func = llm_func
def create_auto_memory(
memory: "FileMemory",
enabled: bool = True,
**kwargs
) -> AutoMemory:
"""
Create an AutoMemory wrapper.
Args:
memory: Base FileMemory instance
enabled: Whether auto-generation is enabled
**kwargs: Additional configuration
Returns:
AutoMemory instance
"""
return AutoMemory(memory, enabled=enabled, **kwargs)
@@ -837,6 +837,445 @@ class FileMemory:
self._save_summaries()
self._log("Imported memory data")
# -------------------------------------------------------------------------
# Session Management
# -------------------------------------------------------------------------
def save_session(
self,
name: str,
conversation_history: Optional[List[Dict[str, Any]]] = None,
metadata: Optional[Dict[str, Any]] = None
) -> str:
"""
Save current session state for later resumption.
Similar to Gemini CLI's /chat save command.
Args:
name: Session name (used as filename)
conversation_history: Optional conversation messages to save
metadata: Optional additional metadata
Returns:
Path to saved session file
"""
sessions_path = self.user_path / "sessions"
sessions_path.mkdir(parents=True, exist_ok=True)
session_file = sessions_path / f"{name}.json"
session_data = {
"name": name,
"user_id": self.user_id,
"saved_at": time.time(),
"saved_at_iso": datetime.now().isoformat(),
"short_term": [item.to_dict() for item in self._short_term],
"long_term_snapshot": [item.to_dict() for item in self._long_term[-50:]], # Last 50
"entity_ids": list(self._entities.keys()),
"conversation_history": conversation_history or [],
"metadata": metadata or {},
"config": self.config
}
self._write_json(session_file, session_data)
self._log(f"Saved session '{name}' to {session_file}")
return str(session_file)
def resume_session(self, name: str) -> Dict[str, Any]:
"""
Resume a previously saved session.
Similar to Gemini CLI's /chat resume command.
Args:
name: Session name to resume
Returns:
Session data including conversation history
"""
sessions_path = self.user_path / "sessions"
session_file = sessions_path / f"{name}.json"
if not session_file.exists():
raise FileNotFoundError(f"Session '{name}' not found")
session_data = self._read_json(session_file, {})
# Restore short-term memory from session
if "short_term" in session_data:
self._short_term = [MemoryItem.from_dict(item) for item in session_data["short_term"]]
self._save_short_term()
self._log(f"Resumed session '{name}'")
return session_data
def list_sessions(self) -> List[Dict[str, Any]]:
"""
List all saved sessions.
Similar to Gemini CLI's /chat list command.
Returns:
List of session info dicts
"""
sessions_path = self.user_path / "sessions"
if not sessions_path.exists():
return []
sessions = []
for session_file in sessions_path.glob("*.json"):
try:
data = self._read_json(session_file, {})
sessions.append({
"name": data.get("name", session_file.stem),
"saved_at": data.get("saved_at_iso", ""),
"short_term_count": len(data.get("short_term", [])),
"has_conversation": len(data.get("conversation_history", [])) > 0,
"file_path": str(session_file)
})
except Exception:
continue
# Sort by saved_at descending
sessions.sort(key=lambda x: x.get("saved_at", ""), reverse=True)
return sessions
def delete_session(self, name: str) -> bool:
"""Delete a saved session."""
sessions_path = self.user_path / "sessions"
session_file = sessions_path / f"{name}.json"
if session_file.exists():
session_file.unlink()
self._log(f"Deleted session '{name}'")
return True
return False
# -------------------------------------------------------------------------
# Context Compression
# -------------------------------------------------------------------------
def compress(
self,
llm_func: Optional[callable] = None,
max_items: int = 10
) -> str:
"""
Compress short-term memory into a summary.
Similar to Gemini CLI's /compress command.
Args:
llm_func: Optional LLM function for summarization.
Should accept (prompt: str) -> str
max_items: Max items to keep after compression
Returns:
The generated summary
"""
if len(self._short_term) <= max_items:
return "" # No compression needed
# Gather content to compress
items_to_compress = self._short_term[:-max_items]
content_list = [item.content for item in items_to_compress]
# Generate summary
if llm_func:
prompt = f"""Summarize the following conversation context into key points.
Preserve important facts, decisions, and context.
Be concise but comprehensive.
Context to summarize:
{chr(10).join(f'- {c}' for c in content_list)}
Summary:"""
summary = llm_func(prompt)
else:
# Simple concatenation if no LLM
summary = "Compressed context: " + " | ".join(content_list[:5]) + "..."
# Add summary as a high-importance long-term memory
self.add_long_term(
content=f"[Session Summary] {summary}",
metadata={"type": "compression_summary", "items_compressed": len(items_to_compress)},
importance=0.9
)
# Keep only recent items
self._short_term = self._short_term[-max_items:]
self._save_short_term()
self._log(f"Compressed {len(items_to_compress)} items into summary")
return summary
def auto_compress_if_needed(
self,
threshold_percent: float = 0.7,
llm_func: Optional[callable] = None
) -> Optional[str]:
"""
Auto-compress if short-term memory exceeds threshold.
Args:
threshold_percent: Compress when STM is this % full (0.0-1.0)
llm_func: Optional LLM function for summarization
Returns:
Summary if compressed, None otherwise
"""
limit = self.config["short_term_limit"]
threshold = int(limit * threshold_percent)
if len(self._short_term) >= threshold:
return self.compress(llm_func=llm_func, max_items=int(limit * 0.3))
return None
# -------------------------------------------------------------------------
# Checkpointing
# -------------------------------------------------------------------------
def create_checkpoint(
self,
name: Optional[str] = None,
include_files: Optional[List[str]] = None
) -> str:
"""
Create a checkpoint of current memory state.
Similar to Gemini CLI's checkpoint before destructive operations.
Args:
name: Optional checkpoint name (auto-generated if not provided)
include_files: Optional list of file paths to snapshot
Returns:
Checkpoint ID
"""
checkpoints_path = self.user_path / "checkpoints"
checkpoints_path.mkdir(parents=True, exist_ok=True)
checkpoint_id = name or f"checkpoint_{int(time.time())}"
checkpoint_file = checkpoints_path / f"{checkpoint_id}.json"
checkpoint_data = {
"id": checkpoint_id,
"created_at": time.time(),
"created_at_iso": datetime.now().isoformat(),
"memory_export": self.export(),
"file_snapshots": {}
}
# Optionally snapshot files
if include_files:
for file_path in include_files:
try:
path = Path(file_path)
if path.exists() and path.is_file():
checkpoint_data["file_snapshots"][file_path] = path.read_text(encoding="utf-8")
except Exception as e:
self._log(f"Could not snapshot file {file_path}: {e}", logging.WARNING)
self._write_json(checkpoint_file, checkpoint_data)
self._log(f"Created checkpoint '{checkpoint_id}'")
return checkpoint_id
def restore_checkpoint(self, checkpoint_id: str, restore_files: bool = False) -> bool:
"""
Restore memory state from a checkpoint.
Args:
checkpoint_id: Checkpoint ID to restore
restore_files: Whether to restore file snapshots
Returns:
True if restored successfully
"""
checkpoints_path = self.user_path / "checkpoints"
checkpoint_file = checkpoints_path / f"{checkpoint_id}.json"
if not checkpoint_file.exists():
self._log(f"Checkpoint '{checkpoint_id}' not found", logging.ERROR)
return False
checkpoint_data = self._read_json(checkpoint_file, {})
# Restore memory
if "memory_export" in checkpoint_data:
self.import_data(checkpoint_data["memory_export"])
# Optionally restore files
if restore_files and "file_snapshots" in checkpoint_data:
for file_path, content in checkpoint_data["file_snapshots"].items():
try:
Path(file_path).write_text(content, encoding="utf-8")
self._log(f"Restored file: {file_path}")
except Exception as e:
self._log(f"Could not restore file {file_path}: {e}", logging.WARNING)
self._log(f"Restored checkpoint '{checkpoint_id}'")
return True
def list_checkpoints(self) -> List[Dict[str, Any]]:
"""List all checkpoints."""
checkpoints_path = self.user_path / "checkpoints"
if not checkpoints_path.exists():
return []
checkpoints = []
for checkpoint_file in checkpoints_path.glob("*.json"):
try:
data = self._read_json(checkpoint_file, {})
checkpoints.append({
"id": data.get("id", checkpoint_file.stem),
"created_at": data.get("created_at_iso", ""),
"has_file_snapshots": len(data.get("file_snapshots", {})) > 0,
"file_path": str(checkpoint_file)
})
except Exception:
continue
# Sort by created_at descending
checkpoints.sort(key=lambda x: x.get("created_at", ""), reverse=True)
return checkpoints
def delete_checkpoint(self, checkpoint_id: str) -> bool:
"""Delete a checkpoint."""
checkpoints_path = self.user_path / "checkpoints"
checkpoint_file = checkpoints_path / f"{checkpoint_id}.json"
if checkpoint_file.exists():
checkpoint_file.unlink()
self._log(f"Deleted checkpoint '{checkpoint_id}'")
return True
return False
# -------------------------------------------------------------------------
# Memory Commands (Slash Commands)
# -------------------------------------------------------------------------
def handle_command(self, command: str) -> Dict[str, Any]:
"""
Handle slash commands like /memory show, /memory add, etc.
Similar to Gemini CLI's /memory commands.
Args:
command: Full command string (e.g., "/memory show" or "/memory add User likes Python")
Returns:
Command result dict
"""
parts = command.strip().split(maxsplit=2)
if len(parts) < 2:
return {"error": "Invalid command. Use: /memory <action> [args]"}
action = parts[1].lower()
args = parts[2] if len(parts) > 2 else ""
if action == "show":
return {
"action": "show",
"stats": self.get_stats(),
"short_term": [item.content for item in self.get_short_term(limit=10)],
"long_term": [item.content for item in self.get_long_term(limit=10)]
}
elif action == "add":
if not args:
return {"error": "Usage: /memory add <content>"}
mem_id = self.add_long_term(args, importance=0.8)
return {"action": "add", "id": mem_id, "content": args}
elif action == "clear":
target = args.lower() if args else "short"
if target == "all":
self.clear_all()
return {"action": "clear", "target": "all"}
elif target == "short":
self.clear_short_term()
return {"action": "clear", "target": "short_term"}
else:
return {"error": "Usage: /memory clear [short|all]"}
elif action == "search":
if not args:
return {"error": "Usage: /memory search <query>"}
results = self.search(args, limit=10)
return {"action": "search", "query": args, "results": results}
elif action == "save":
if not args:
return {"error": "Usage: /memory save <session_name>"}
path = self.save_session(args)
return {"action": "save", "session": args, "path": path}
elif action == "resume":
if not args:
return {"error": "Usage: /memory resume <session_name>"}
try:
self.resume_session(args)
return {"action": "resume", "session": args, "restored": True}
except FileNotFoundError:
return {"error": f"Session '{args}' not found"}
elif action == "sessions":
sessions = self.list_sessions()
return {"action": "sessions", "sessions": sessions}
elif action == "compress":
summary = self.compress()
return {"action": "compress", "summary": summary}
elif action == "checkpoint":
checkpoint_id = self.create_checkpoint(args if args else None)
return {"action": "checkpoint", "id": checkpoint_id}
elif action == "restore":
if not args:
return {"error": "Usage: /memory restore <checkpoint_id>"}
success = self.restore_checkpoint(args)
return {"action": "restore", "checkpoint": args, "success": success}
elif action == "checkpoints":
checkpoints = self.list_checkpoints()
return {"action": "checkpoints", "checkpoints": checkpoints}
elif action == "refresh":
self._load_all()
return {"action": "refresh", "message": "Memory reloaded from disk"}
elif action == "help":
return {
"action": "help",
"commands": {
"/memory show": "Display memory stats and recent items",
"/memory add <content>": "Add to long-term memory",
"/memory clear [short|all]": "Clear memory",
"/memory search <query>": "Search memories",
"/memory save <name>": "Save session",
"/memory resume <name>": "Resume session",
"/memory sessions": "List saved sessions",
"/memory compress": "Compress short-term memory",
"/memory checkpoint [name]": "Create checkpoint",
"/memory restore <id>": "Restore checkpoint",
"/memory checkpoints": "List checkpoints",
"/memory refresh": "Reload from disk"
}
}
else:
return {"error": f"Unknown action: {action}. Use /memory help for available commands."}
# Convenience function for simple usage
@@ -0,0 +1,417 @@
"""
Cascade Hooks System for PraisonAI Agents.
Provides event hooks similar to Windsurf's Cascade Hooks,
allowing custom actions before/after agent operations.
Features:
- Pre/post hooks for read, write, command, prompt operations
- JSON configuration (.praison/hooks.json)
- Script execution with timeout
- Exit code handling
- Lazy loading for performance
Configuration (.praison/hooks.json):
{
"hooks": {
"pre_write_code": "./scripts/lint.sh",
"post_write_code": "./scripts/format.sh",
"pre_run_command": "./scripts/validate.sh",
"post_user_prompt": "./scripts/log.sh"
},
"timeout": 30,
"enabled": true
}
Hook Events:
- pre_read_code: Before reading a file
- post_read_code: After reading a file
- pre_write_code: Before writing to a file
- post_write_code: After writing to a file
- pre_run_command: Before running a terminal command
- post_run_command: After running a terminal command
- pre_user_prompt: Before processing user prompt
- post_user_prompt: After processing user prompt
"""
import os
import json
import logging
import subprocess
from pathlib import Path
from typing import Any, Dict, List, Optional, Callable, Literal
from dataclasses import dataclass
logger = logging.getLogger(__name__)
# Hook event types
HookEvent = Literal[
"pre_read_code",
"post_read_code",
"pre_write_code",
"post_write_code",
"pre_run_command",
"post_run_command",
"pre_user_prompt",
"post_user_prompt",
"pre_mcp_tool_use",
"post_mcp_tool_use"
]
@dataclass
class HookResult:
"""Result of a hook execution."""
success: bool
exit_code: int = 0
stdout: str = ""
stderr: str = ""
blocked: bool = False # If True, the operation should be blocked
modified_input: Optional[str] = None # Modified input if applicable
@dataclass
class HookConfig:
"""Configuration for a single hook."""
event: HookEvent
command: str
timeout: int = 30
enabled: bool = True
block_on_failure: bool = False
pass_input: bool = True
class HooksManager:
"""
Manages hook discovery, loading, and execution.
Hooks are configured in .praison/hooks.json and can execute
scripts before/after agent operations.
Example:
```python
hooks = HooksManager(workspace_path="/path/to/project")
# Execute pre-write hook
result = hooks.execute(
"pre_write_code",
context={"file_path": "main.py", "content": "..."}
)
if result.blocked:
print("Write blocked by hook")
```
"""
CONFIG_FILE = ".praison/hooks.json"
def __init__(
self,
workspace_path: Optional[str] = None,
verbose: int = 0
):
"""
Initialize HooksManager.
Args:
workspace_path: Path to workspace/project root
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.verbose = verbose
self._hooks: Dict[HookEvent, List[HookConfig]] = {}
self._global_timeout = 30
self._enabled = True
self._loaded = False
# Python callable hooks (for programmatic use)
self._callable_hooks: Dict[HookEvent, List[Callable]] = {}
def _log(self, msg: str, level: int = logging.INFO):
"""Log message if verbose."""
if self.verbose >= 1:
logger.log(level, msg)
def _ensure_loaded(self):
"""Lazy load hooks on first access."""
if not self._loaded:
self._load_config()
self._loaded = True
def _load_config(self):
"""Load hooks configuration from file."""
config_path = self.workspace_path / self.CONFIG_FILE.replace("/", os.sep)
if not config_path.exists():
return
try:
config = json.loads(config_path.read_text(encoding="utf-8"))
self._enabled = config.get("enabled", True)
self._global_timeout = config.get("timeout", 30)
hooks_config = config.get("hooks", {})
for event, command in hooks_config.items():
if event not in self._hooks:
self._hooks[event] = []
if isinstance(command, str):
# Simple command string
self._hooks[event].append(HookConfig(
event=event,
command=command,
timeout=self._global_timeout
))
elif isinstance(command, dict):
# Detailed config
self._hooks[event].append(HookConfig(
event=event,
command=command.get("command", ""),
timeout=command.get("timeout", self._global_timeout),
enabled=command.get("enabled", True),
block_on_failure=command.get("block_on_failure", False),
pass_input=command.get("pass_input", True)
))
elif isinstance(command, list):
# Multiple commands for same event
for cmd in command:
if isinstance(cmd, str):
self._hooks[event].append(HookConfig(
event=event,
command=cmd,
timeout=self._global_timeout
))
self._log(f"Loaded {sum(len(h) for h in self._hooks.values())} hooks")
except Exception as e:
self._log(f"Error loading hooks config: {e}", logging.WARNING)
def reload(self):
"""Reload hooks configuration."""
self._hooks = {}
self._loaded = False
self._ensure_loaded()
def register(
self,
event: HookEvent,
handler: Callable[[Dict[str, Any]], Optional[Dict[str, Any]]]
):
"""
Register a Python callable as a hook.
Args:
event: Hook event type
handler: Callable that receives context dict and optionally returns modified context
"""
if event not in self._callable_hooks:
self._callable_hooks[event] = []
self._callable_hooks[event].append(handler)
def unregister(self, event: HookEvent, handler: Callable):
"""Unregister a callable hook."""
if event in self._callable_hooks:
self._callable_hooks[event] = [
h for h in self._callable_hooks[event] if h != handler
]
def execute(
self,
event: HookEvent,
context: Optional[Dict[str, Any]] = None
) -> HookResult:
"""
Execute hooks for an event.
Args:
event: Hook event type
context: Context data to pass to hooks
Returns:
HookResult with execution status
"""
self._ensure_loaded()
if not self._enabled:
return HookResult(success=True)
context = context or {}
combined_result = HookResult(success=True)
# Execute callable hooks first
if event in self._callable_hooks:
for handler in self._callable_hooks[event]:
try:
result = handler(context)
if isinstance(result, dict):
context.update(result)
except Exception as e:
self._log(f"Callable hook error for {event}: {e}", logging.WARNING)
combined_result.success = False
# Execute script hooks
if event in self._hooks:
for hook in self._hooks[event]:
if not hook.enabled:
continue
result = self._execute_script(hook, context)
if not result.success:
combined_result.success = False
combined_result.stderr += result.stderr
if hook.block_on_failure:
combined_result.blocked = True
break
combined_result.stdout += result.stdout
if result.modified_input:
combined_result.modified_input = result.modified_input
return combined_result
def _execute_script(
self,
hook: HookConfig,
context: Dict[str, Any]
) -> HookResult:
"""Execute a script hook."""
try:
# Prepare environment
env = os.environ.copy()
# Pass context as environment variables
if hook.pass_input:
for key, value in context.items():
if isinstance(value, (str, int, float, bool)):
env[f"PRAISON_HOOK_{key.upper()}"] = str(value)
# Also pass as JSON
env["PRAISON_HOOK_CONTEXT"] = json.dumps(context)
# Resolve command path
command = hook.command
if not os.path.isabs(command) and not command.startswith("./"):
command = str(self.workspace_path / command)
# Execute
result = subprocess.run(
command,
shell=True,
cwd=str(self.workspace_path),
env=env,
capture_output=True,
text=True,
timeout=hook.timeout
)
success = result.returncode == 0
# Check for special exit codes
blocked = result.returncode == 1 and hook.block_on_failure
# Check stdout for modified input
modified_input = None
if result.stdout.startswith("MODIFIED:"):
modified_input = result.stdout[9:].strip()
self._log(f"Hook {hook.event} executed: exit={result.returncode}")
return HookResult(
success=success,
exit_code=result.returncode,
stdout=result.stdout,
stderr=result.stderr,
blocked=blocked,
modified_input=modified_input
)
except subprocess.TimeoutExpired:
self._log(f"Hook {hook.event} timed out after {hook.timeout}s", logging.WARNING)
return HookResult(
success=False,
exit_code=-1,
stderr=f"Timeout after {hook.timeout}s"
)
except Exception as e:
self._log(f"Hook {hook.event} error: {e}", logging.WARNING)
return HookResult(
success=False,
exit_code=-1,
stderr=str(e)
)
def has_hooks(self, event: HookEvent) -> bool:
"""Check if any hooks are registered for an event."""
self._ensure_loaded()
return (
event in self._hooks and len(self._hooks[event]) > 0 or
event in self._callable_hooks and len(self._callable_hooks[event]) > 0
)
def get_stats(self) -> Dict[str, Any]:
"""Get hooks statistics."""
self._ensure_loaded()
script_hooks = sum(len(h) for h in self._hooks.values())
callable_hooks = sum(len(h) for h in self._callable_hooks.values())
return {
"enabled": self._enabled,
"script_hooks": script_hooks,
"callable_hooks": callable_hooks,
"total_hooks": script_hooks + callable_hooks,
"events": list(set(list(self._hooks.keys()) + list(self._callable_hooks.keys())))
}
def create_config(
self,
hooks: Optional[Dict[str, str]] = None,
timeout: int = 30,
enabled: bool = True
):
"""
Create a hooks configuration file.
Args:
hooks: Dict mapping event names to commands
timeout: Global timeout in seconds
enabled: Whether hooks are enabled
"""
config_dir = self.workspace_path / ".praison"
config_dir.mkdir(parents=True, exist_ok=True)
config = {
"enabled": enabled,
"timeout": timeout,
"hooks": hooks or {}
}
config_path = config_dir / "hooks.json"
config_path.write_text(json.dumps(config, indent=2), encoding="utf-8")
self._log(f"Created hooks config at {config_path}")
self.reload()
def create_hooks_manager(
workspace_path: Optional[str] = None,
**kwargs
) -> HooksManager:
"""
Create a HooksManager instance.
Args:
workspace_path: Path to workspace
**kwargs: Additional configuration
Returns:
HooksManager instance
"""
return HooksManager(workspace_path=workspace_path, **kwargs)
@@ -0,0 +1,819 @@
"""
Rules Manager for PraisonAI Agents.
Provides persistent rules/instructions support similar to:
- Cursor (.cursor/rules/*.mdc)
- Windsurf (.windsurf/rules/)
- Claude Code (CLAUDE.md, .claude/rules/)
- Codex CLI (AGENTS.md)
- Gemini CLI (GEMINI.md)
Features:
- Auto-discovery of root instruction files (CLAUDE.md, AGENTS.md, etc.)
- Git root discovery for monorepo support
- @Import syntax for including other files
- Local override files (CLAUDE.local.md)
- Multiple activation modes (always, glob, manual, ai_decision)
- Character limits for context window management
- Symlink support for shared rules
Storage Structure:
.praison/rules/
├── global.md # Always applied (global)
├── python.md # Glob: **/*.py
├── testing.md # Glob: **/*.test.*
└── security.md # Activation: manual (@security)
Rule File Format (YAML frontmatter + Markdown):
---
description: Python coding guidelines
globs: ["**/*.py", "**/*.pyx"]
activation: always # always, glob, manual, ai_decision
priority: 10
---
# Python Guidelines
- Use type hints
- Follow PEP 8
"""
import os
import re
import fnmatch
import logging
import subprocess
from pathlib import Path
from typing import Any, Dict, List, Optional, Literal, Callable
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class Rule:
"""A single rule with metadata and content."""
name: str
content: str
description: str = ""
globs: List[str] = field(default_factory=list)
activation: Literal["always", "glob", "manual", "ai_decision"] = "always"
priority: int = 0
file_path: Optional[str] = None
def matches_file(self, file_path: str) -> bool:
"""Check if this rule matches a given file path."""
if self.activation == "always":
return True
if self.activation == "manual":
return False # Only activated via @mention
if self.activation == "glob" and self.globs:
for pattern in self.globs:
if fnmatch.fnmatch(file_path, pattern):
return True
# Also try with ** expansion
if "**" in pattern:
# Convert ** to regex for recursive matching
regex_pattern = pattern.replace("**", ".*").replace("*", "[^/]*")
if re.match(regex_pattern, file_path):
return True
return False
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"content": self.content,
"description": self.description,
"globs": self.globs,
"activation": self.activation,
"priority": self.priority,
"file_path": self.file_path
}
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> 'Rule':
return cls(**data)
class RulesManager:
"""
Manages rules/instructions for AI agents.
Supports multiple rule sources:
- Global rules (~/.praison/rules/)
- Workspace rules (.praison/rules/)
- Subdirectory rules (subdir/.praison/rules/)
- Root instruction files (CLAUDE.md, AGENTS.md, GEMINI.md, .cursorrules, .windsurfrules)
Example:
```python
rules = RulesManager(workspace_path="/path/to/project")
# Get all active rules
active_rules = rules.get_active_rules()
# Get rules for specific file
python_rules = rules.get_rules_for_file("src/main.py")
# Get manually invoked rule
security_rule = rules.get_rule_by_name("security")
# Build context string for LLM
context = rules.build_rules_context(file_path="src/main.py")
```
"""
RULES_DIR_NAME = ".praison/rules"
SUPPORTED_EXTENSIONS = [".md", ".mdc", ".txt"]
# Root-level instruction files (like Claude, Codex, Gemini, Cursor, Windsurf)
ROOT_INSTRUCTION_FILES = [
"CLAUDE.md", # Claude Code memory file
"CLAUDE.local.md", # Claude Code local overrides (gitignored)
"AGENTS.md", # OpenAI Codex CLI instructions
"GEMINI.md", # Gemini CLI memory file
".cursorrules", # Cursor legacy rules (deprecated but supported)
".windsurfrules", # Windsurf legacy rules
"PRAISON.md", # PraisonAI native instructions
"PRAISON.local.md", # PraisonAI local overrides (gitignored)
]
# Additional rules directories to discover (like Claude, Windsurf)
ADDITIONAL_RULES_DIRS = [
".claude/rules", # Claude Code modular rules
".windsurf/rules", # Windsurf rules
".cursor/rules", # Cursor rules
]
# Maximum characters per rule (like Windsurf's 12000 limit)
MAX_RULE_CHARS = 12000
# Import pattern for @path/to/file syntax
IMPORT_PATTERN = re.compile(r'(?<!`)@([\w./-]+)(?!`)')
def __init__(
self,
workspace_path: Optional[str] = None,
global_rules_path: Optional[str] = None,
verbose: int = 0
):
"""
Initialize RulesManager.
Args:
workspace_path: Path to workspace/project root
global_rules_path: Path to global rules (default: ~/.praison/rules)
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.global_rules_path = Path(global_rules_path) if global_rules_path else Path.home() / ".praison" / "rules"
self.verbose = verbose
self._rules: Dict[str, Rule] = {}
self._load_all_rules()
def _log(self, msg: str, level: int = logging.INFO):
"""Log message if verbose."""
if self.verbose >= 1:
logger.log(level, msg)
def _find_git_root(self, start_path: Optional[Path] = None) -> Optional[Path]:
"""
Find the git repository root from a starting path.
Uses pure Python (no gitpython dependency) for performance.
Falls back to subprocess if needed.
Args:
start_path: Starting path (defaults to workspace_path)
Returns:
Path to git root, or None if not in a git repo
"""
start = start_path or self.workspace_path
# Method 1: Walk up looking for .git directory (fastest, no subprocess)
current = Path(start).resolve()
while current != current.parent:
git_dir = current / ".git"
if git_dir.exists():
return current
current = current.parent
# Method 2: Fallback to git command (handles edge cases like worktrees)
try:
result = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
cwd=str(start),
capture_output=True,
text=True,
timeout=2 # Prevent hanging
)
if result.returncode == 0:
return Path(result.stdout.strip())
except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
pass
return None
def _process_imports(self, content: str, base_path: Path, depth: int = 0) -> str:
"""
Process @import syntax in rule content.
Supports:
- @path/to/file - Import relative to workspace
- @~/path/to/file - Import from home directory
- @./path/to/file - Import relative to current file
Args:
content: Rule content with potential @imports
base_path: Base path for relative imports
depth: Current recursion depth (max 5 to prevent loops)
Returns:
Content with imports resolved
"""
if depth > 5:
self._log("Import depth exceeded (max 5), stopping", logging.WARNING)
return content
def replace_import(match):
import_path = match.group(1)
# Skip if it looks like a mention (@username) or package (@org/pkg)
if "/" not in import_path and "." not in import_path:
return match.group(0)
# Resolve path
if import_path.startswith("~/"):
file_path = Path.home() / import_path[2:]
elif import_path.startswith("./"):
file_path = base_path / import_path[2:]
else:
file_path = self.workspace_path / import_path
# Try with and without .md extension
if not file_path.exists() and not file_path.suffix:
file_path = file_path.with_suffix(".md")
if file_path.exists() and file_path.is_file():
try:
imported_content = file_path.read_text(encoding="utf-8")
# Recursively process imports in imported content
imported_content = self._process_imports(
imported_content,
file_path.parent,
depth + 1
)
return imported_content
except Exception as e:
self._log(f"Failed to import {file_path}: {e}", logging.WARNING)
return match.group(0)
else:
# Not a file import, leave as-is (might be @mention)
return match.group(0)
return self.IMPORT_PATTERN.sub(replace_import, content)
def _truncate_rule_content(self, content: str) -> str:
"""Truncate rule content to MAX_RULE_CHARS if needed."""
if len(content) <= self.MAX_RULE_CHARS:
return content
truncated = content[:self.MAX_RULE_CHARS - 50]
return truncated + "\n\n... (truncated, exceeded 12000 char limit)"
def _parse_frontmatter(self, content: str) -> tuple[Dict[str, Any], str]:
"""Parse YAML frontmatter from markdown content."""
frontmatter = {}
body = content
# Check for YAML frontmatter (--- ... ---)
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
yaml_content = parts[1].strip()
body = parts[2].strip()
# Simple YAML parsing (avoid dependency)
for line in yaml_content.split("\n"):
line = line.strip()
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip()
# Handle lists
if value.startswith("[") and value.endswith("]"):
# Parse simple list: ["*.py", "*.pyx"]
value = [v.strip().strip('"\'') for v in value[1:-1].split(",") if v.strip()]
# Handle booleans
elif value.lower() in ("true", "false"):
value = value.lower() == "true"
# Handle numbers
elif value.isdigit():
value = int(value)
# Handle quoted strings
elif value.startswith('"') and value.endswith('"'):
value = value[1:-1]
elif value.startswith("'") and value.endswith("'"):
value = value[1:-1]
frontmatter[key] = value
return frontmatter, body
def _load_rule_file(self, file_path: Path) -> Optional[Rule]:
"""Load a single rule file with @import processing and truncation."""
try:
content = file_path.read_text(encoding="utf-8")
frontmatter, body = self._parse_frontmatter(content)
# Process @imports in the body
body = self._process_imports(body, file_path.parent)
# Truncate if exceeds limit
body = self._truncate_rule_content(body)
# Extract rule name from filename
name = file_path.stem
# Build Rule object
rule = Rule(
name=name,
content=body,
description=frontmatter.get("description", ""),
globs=frontmatter.get("globs", []),
activation=frontmatter.get("activation", "always"),
priority=frontmatter.get("priority", 0),
file_path=str(file_path)
)
self._log(f"Loaded rule '{name}' from {file_path}")
return rule
except Exception as e:
self._log(f"Error loading rule file {file_path}: {e}", logging.WARNING)
return None
def _discover_rules_in_dir(self, rules_dir: Path) -> List[Rule]:
"""Discover all rule files in a directory."""
rules = []
if not rules_dir.exists():
return rules
for ext in self.SUPPORTED_EXTENSIONS:
for file_path in rules_dir.glob(f"*{ext}"):
rule = self._load_rule_file(file_path)
if rule:
rules.append(rule)
return rules
def _load_root_instruction_file(self, file_path: Path) -> Optional[Rule]:
"""Load a root instruction file (CLAUDE.md, AGENTS.md, etc.) with @import processing."""
try:
if not file_path.exists():
return None
# Handle symlinks (like Claude Code supports)
if file_path.is_symlink():
file_path = file_path.resolve()
if not file_path.exists():
return None
content = file_path.read_text(encoding="utf-8")
frontmatter, body = self._parse_frontmatter(content)
# Process @imports in the body
body_content = body if body else content
body_content = self._process_imports(body_content, file_path.parent)
# Truncate if exceeds limit
body_content = self._truncate_rule_content(body_content)
# Use filename (without extension) as rule name
name = file_path.stem.lower().replace(".", "_")
# Determine description based on file type
descriptions = {
"claude": "Claude Code memory instructions",
"claude_local": "Claude Code local overrides",
"agents": "OpenAI Codex CLI instructions",
"gemini": "Gemini CLI memory instructions",
"cursorrules": "Cursor IDE rules (legacy)",
"windsurfrules": "Windsurf IDE rules (legacy)",
"praison": "PraisonAI native instructions",
"praison_local": "PraisonAI local overrides",
}
description = descriptions.get(name, f"Instructions from {file_path.name}")
# Local files get higher priority (override main files)
base_priority = 600 if "_local" in name else 500
rule = Rule(
name=name,
content=body_content,
description=frontmatter.get("description", description),
globs=frontmatter.get("globs", []),
activation=frontmatter.get("activation", "always"),
priority=frontmatter.get("priority", base_priority),
file_path=str(file_path)
)
self._log(f"Loaded root instruction file '{file_path.name}'")
return rule
except Exception as e:
self._log(f"Error loading root instruction file {file_path}: {e}", logging.WARNING)
return None
def _load_all_rules(self):
"""Load rules from all sources including git root and additional directories."""
self._rules = {}
# Find git root for monorepo support
git_root = self._find_git_root()
# 1. Load global rules (lowest priority)
global_rules = self._discover_rules_in_dir(self.global_rules_path)
for rule in global_rules:
rule.priority = rule.priority - 1000 # Lower priority for global
self._rules[f"global:{rule.name}"] = rule
# 2. Load root instruction files from git root (if different from workspace)
if git_root and git_root != self.workspace_path:
for filename in self.ROOT_INSTRUCTION_FILES:
file_path = git_root / filename
rule = self._load_root_instruction_file(file_path)
if rule:
rule.priority = rule.priority - 100 # Slightly lower than workspace
self._rules[f"gitroot:{rule.name}"] = rule
# 3. Load root instruction files from workspace (CLAUDE.md, AGENTS.md, etc.)
for filename in self.ROOT_INSTRUCTION_FILES:
file_path = self.workspace_path / filename
rule = self._load_root_instruction_file(file_path)
if rule:
self._rules[f"root:{rule.name}"] = rule
# 4. Load workspace rules from .praison/rules/
workspace_rules_dir = self.workspace_path / self.RULES_DIR_NAME.replace("/", os.sep)
workspace_rules = self._discover_rules_in_dir(workspace_rules_dir)
for rule in workspace_rules:
self._rules[f"workspace:{rule.name}"] = rule
# 5. Load additional rules directories (.claude/rules, .windsurf/rules, .cursor/rules)
for rules_dir_name in self.ADDITIONAL_RULES_DIRS:
# Check workspace
rules_dir = self.workspace_path / rules_dir_name.replace("/", os.sep)
additional_rules = self._discover_rules_in_dir(rules_dir)
for rule in additional_rules:
rule.priority = rule.priority + 50 # Slightly higher priority
dir_prefix = rules_dir_name.split("/")[0].strip(".")
self._rules[f"{dir_prefix}:{rule.name}"] = rule
# Check git root if different
if git_root and git_root != self.workspace_path:
rules_dir = git_root / rules_dir_name.replace("/", os.sep)
additional_rules = self._discover_rules_in_dir(rules_dir)
for rule in additional_rules:
dir_prefix = rules_dir_name.split("/")[0].strip(".")
self._rules[f"gitroot_{dir_prefix}:{rule.name}"] = rule
# 6. Load subdirectory rules (walk up from cwd to workspace/git root)
stop_at = git_root if git_root else self.workspace_path
current = Path.cwd()
while current != stop_at and current != current.parent:
# Check .praison/rules/
subdir_rules_dir = current / self.RULES_DIR_NAME.replace("/", os.sep)
subdir_rules = self._discover_rules_in_dir(subdir_rules_dir)
for rule in subdir_rules:
rule.priority = rule.priority + 100 # Higher priority for closer dirs
self._rules[f"subdir:{rule.name}"] = rule
# Check additional rules directories in subdirs
for rules_dir_name in self.ADDITIONAL_RULES_DIRS:
subdir_rules_dir = current / rules_dir_name.replace("/", os.sep)
subdir_rules = self._discover_rules_in_dir(subdir_rules_dir)
for rule in subdir_rules:
rule.priority = rule.priority + 150 # Even higher for subdir additional
dir_prefix = rules_dir_name.split("/")[0].strip(".")
self._rules[f"subdir_{dir_prefix}:{rule.name}"] = rule
current = current.parent
self._log(f"Loaded {len(self._rules)} rules total")
def reload(self):
"""Reload all rules from disk."""
self._load_all_rules()
def get_all_rules(self) -> List[Rule]:
"""Get all loaded rules, sorted by priority."""
rules = list(self._rules.values())
rules.sort(key=lambda r: r.priority, reverse=True)
return rules
def get_active_rules(self, file_path: Optional[str] = None) -> List[Rule]:
"""
Get rules that are currently active.
Args:
file_path: Optional file path to filter glob-based rules
Returns:
List of active rules sorted by priority
"""
active = []
for rule in self._rules.values():
if rule.activation == "always":
active.append(rule)
elif rule.activation == "glob" and file_path:
if rule.matches_file(file_path):
active.append(rule)
elif rule.activation == "ai_decision":
# AI will decide - include for context
active.append(rule)
# Skip "manual" - only activated via @mention
active.sort(key=lambda r: r.priority, reverse=True)
return active
def get_rules_for_file(self, file_path: str) -> List[Rule]:
"""Get rules that apply to a specific file."""
return self.get_active_rules(file_path=file_path)
def get_rule_by_name(self, name: str) -> Optional[Rule]:
"""Get a rule by name (for manual @mention invocation)."""
# Check all scopes
for scope in ["subdir", "workspace", "global"]:
key = f"{scope}:{name}"
if key in self._rules:
return self._rules[key]
# Also check without scope prefix
for rule in self._rules.values():
if rule.name == name:
return rule
return None
def get_manual_rules(self) -> List[Rule]:
"""Get all rules that require manual activation."""
return [r for r in self._rules.values() if r.activation == "manual"]
def build_rules_context(
self,
file_path: Optional[str] = None,
include_manual: Optional[List[str]] = None,
max_chars: int = 10000
) -> str:
"""
Build context string from active rules for LLM.
Args:
file_path: Optional file path for glob matching
include_manual: List of manual rule names to include
max_chars: Maximum characters in output
Returns:
Formatted rules context string
"""
parts = []
total_chars = 0
# Get active rules
active_rules = self.get_active_rules(file_path=file_path)
# Add manually invoked rules
if include_manual:
for name in include_manual:
rule = self.get_rule_by_name(name)
if rule and rule not in active_rules:
active_rules.append(rule)
# Sort by priority
active_rules.sort(key=lambda r: r.priority, reverse=True)
for rule in active_rules:
if total_chars >= max_chars:
break
rule_text = rule.content.strip()
if not rule_text:
continue
# Add rule with header
if rule.description:
header = f"## {rule.name}: {rule.description}"
else:
header = f"## {rule.name}"
section = f"{header}\n{rule_text}\n"
if total_chars + len(section) <= max_chars:
parts.append(section)
total_chars += len(section)
else:
# Truncate last rule
remaining = max_chars - total_chars
if remaining > 100:
parts.append(section[:remaining] + "\n... (truncated)")
break
return "\n".join(parts)
def create_rule(
self,
name: str,
content: str,
description: str = "",
globs: Optional[List[str]] = None,
activation: Literal["always", "glob", "manual", "ai_decision"] = "always",
priority: int = 0,
scope: Literal["global", "workspace"] = "workspace"
) -> Rule:
"""
Create a new rule file.
Args:
name: Rule name (used as filename)
content: Rule content (markdown)
description: Short description
globs: Glob patterns for activation
activation: Activation mode
priority: Priority (higher = applied first)
scope: Where to save (global or workspace)
Returns:
Created Rule object
"""
# Determine save path
if scope == "global":
rules_dir = self.global_rules_path
else:
rules_dir = self.workspace_path / self.RULES_DIR_NAME.replace("/", os.sep)
rules_dir.mkdir(parents=True, exist_ok=True)
file_path = rules_dir / f"{name}.md"
# Build frontmatter
frontmatter_lines = ["---"]
if description:
frontmatter_lines.append(f'description: "{description}"')
if globs:
globs_str = ", ".join(f'"{g}"' for g in globs)
frontmatter_lines.append(f"globs: [{globs_str}]")
frontmatter_lines.append(f"activation: {activation}")
if priority != 0:
frontmatter_lines.append(f"priority: {priority}")
frontmatter_lines.append("---")
frontmatter_lines.append("")
# Write file
full_content = "\n".join(frontmatter_lines) + content
file_path.write_text(full_content, encoding="utf-8")
# Create and register rule
rule = Rule(
name=name,
content=content,
description=description,
globs=globs or [],
activation=activation,
priority=priority,
file_path=str(file_path)
)
self._rules[f"{scope}:{name}"] = rule
self._log(f"Created rule '{name}' at {file_path}")
return rule
def delete_rule(self, name: str, scope: Optional[str] = None) -> bool:
"""
Delete a rule file.
Args:
name: Rule name
scope: Scope to delete from (global, workspace, or None for any)
Returns:
True if deleted, False if not found
"""
# Find the rule
rule = None
key = None
if scope:
key = f"{scope}:{name}"
rule = self._rules.get(key)
else:
for s in ["subdir", "workspace", "global"]:
key = f"{s}:{name}"
if key in self._rules:
rule = self._rules[key]
break
if not rule or not rule.file_path:
return False
# Delete file
try:
Path(rule.file_path).unlink()
del self._rules[key]
self._log(f"Deleted rule '{name}'")
return True
except Exception as e:
self._log(f"Error deleting rule '{name}': {e}", logging.ERROR)
return False
def evaluate_ai_decision(
self,
rule: Rule,
context: str,
llm_func: Optional[Callable[[str], str]] = None
) -> bool:
"""
Evaluate if an ai_decision rule should be applied.
Uses the rule's description to determine relevance to current context.
Args:
rule: Rule with activation="ai_decision"
context: Current conversation/task context
llm_func: Optional LLM function for evaluation (if None, returns True)
Returns:
True if rule should be applied, False otherwise
"""
if rule.activation != "ai_decision":
return rule.activation == "always"
if not llm_func:
# Without LLM, include ai_decision rules by default
return True
# Use LLM to decide
prompt = f"""Determine if the following rule should be applied to the current context.
Rule Name: {rule.name}
Rule Description: {rule.description}
Current Context:
{context[:1000]}
Should this rule be applied? Answer only "yes" or "no"."""
try:
response = llm_func(prompt).strip().lower()
return response.startswith("yes")
except Exception as e:
self._log(f"Error evaluating ai_decision for rule '{rule.name}': {e}", logging.WARNING)
return True # Default to including the rule
def get_stats(self) -> Dict[str, Any]:
"""Get rules statistics."""
rules = list(self._rules.values())
# Count rules by source
source_counts = {}
for key in self._rules.keys():
source = key.split(":")[0]
source_counts[source] = source_counts.get(source, 0) + 1
return {
"total_rules": len(rules),
"always_rules": len([r for r in rules if r.activation == "always"]),
"glob_rules": len([r for r in rules if r.activation == "glob"]),
"manual_rules": len([r for r in rules if r.activation == "manual"]),
"ai_decision_rules": len([r for r in rules if r.activation == "ai_decision"]),
"global_rules": source_counts.get("global", 0),
"root_rules": source_counts.get("root", 0),
"gitroot_rules": source_counts.get("gitroot", 0),
"workspace_rules": source_counts.get("workspace", 0),
"subdir_rules": source_counts.get("subdir", 0),
"claude_rules": source_counts.get("claude", 0),
"windsurf_rules": source_counts.get("windsurf", 0),
"cursor_rules": source_counts.get("cursor", 0),
"sources": source_counts
}
def create_rules_manager(
workspace_path: Optional[str] = None,
**kwargs
) -> RulesManager:
"""
Create a RulesManager instance.
Args:
workspace_path: Path to workspace
**kwargs: Additional configuration
Returns:
RulesManager instance
"""
return RulesManager(workspace_path=workspace_path, **kwargs)
@@ -0,0 +1,514 @@
"""
Workflows System for PraisonAI Agents.
Provides reusable multi-step workflows similar to Windsurf's Workflows,
allowing users to define and execute complex task sequences.
Features:
- YAML/Markdown workflow definitions
- Auto-discovery from .praison/workflows/
- Step-by-step execution with context passing
- Variable substitution
- Conditional steps
- Lazy loading for performance
Storage Structure:
.praison/workflows/
├── deploy.md # Deployment workflow
├── test.md # Testing workflow
└── review.md # Code review workflow
"""
import os
import re
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional, Callable, Literal
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class WorkflowStep:
"""A single step in a workflow."""
name: str
description: str = ""
action: str = "" # The action/prompt to execute
condition: Optional[str] = None # Optional condition for execution
on_error: Literal["stop", "continue", "retry"] = "stop"
max_retries: int = 1
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"action": self.action,
"condition": self.condition,
"on_error": self.on_error,
"max_retries": self.max_retries
}
@dataclass
class Workflow:
"""A complete workflow with multiple steps."""
name: str
description: str = ""
steps: List[WorkflowStep] = field(default_factory=list)
variables: Dict[str, Any] = field(default_factory=dict)
file_path: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"description": self.description,
"steps": [s.to_dict() for s in self.steps],
"variables": self.variables,
"file_path": self.file_path
}
class WorkflowManager:
"""
Manages workflow discovery, loading, and execution.
Workflows are defined in markdown files with YAML frontmatter
and step definitions.
Example workflow file (.praison/workflows/deploy.md):
```markdown
---
name: Deploy to Production
description: Deploy the application to production
variables:
environment: production
branch: main
---
## Step 1: Run Tests
Run all tests to ensure code quality.
```action
Run the test suite with pytest
```
## Step 2: Build
Build the application for production.
```action
Build the application with production settings
```
## Step 3: Deploy
Deploy to the production server.
```action
Deploy to {{environment}} from {{branch}}
```
```
Example usage:
```python
manager = WorkflowManager(workspace_path="/path/to/project")
# List available workflows
workflows = manager.list_workflows()
# Get a specific workflow
deploy = manager.get_workflow("deploy")
# Execute a workflow
results = manager.execute(
"deploy",
executor=lambda prompt: agent.chat(prompt),
variables={"branch": "release-1.0"}
)
```
"""
WORKFLOWS_DIR = ".praison/workflows"
SUPPORTED_EXTENSIONS = [".md", ".yaml", ".yml"]
def __init__(
self,
workspace_path: Optional[str] = None,
verbose: int = 0
):
"""
Initialize WorkflowManager.
Args:
workspace_path: Path to workspace/project root
verbose: Verbosity level
"""
self.workspace_path = Path(workspace_path) if workspace_path else Path.cwd()
self.verbose = verbose
self._workflows: Dict[str, Workflow] = {}
self._loaded = False
def _log(self, msg: str, level: int = logging.INFO):
"""Log message if verbose."""
if self.verbose >= 1:
logger.log(level, msg)
def _ensure_loaded(self):
"""Lazy load workflows on first access."""
if not self._loaded:
self._discover_workflows()
self._loaded = True
def _parse_frontmatter(self, content: str) -> tuple[Dict[str, Any], str]:
"""Parse YAML frontmatter from markdown content."""
frontmatter = {}
body = content
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
yaml_content = parts[1].strip()
body = parts[2].strip()
# Simple YAML parsing
for line in yaml_content.split("\n"):
line = line.strip()
if ":" in line and not line.startswith("#"):
key, value = line.split(":", 1)
key = key.strip()
value = value.strip()
# Handle basic types
if value.lower() in ("true", "false"):
value = value.lower() == "true"
elif value.isdigit():
value = int(value)
elif value.startswith('"') and value.endswith('"'):
value = value[1:-1]
frontmatter[key] = value
return frontmatter, body
def _parse_steps(self, body: str) -> List[WorkflowStep]:
"""Parse workflow steps from markdown body."""
steps = []
# Split by ## headers
step_pattern = re.compile(r'^##\s+(?:Step\s+\d+[:\s]*)?(.+)$', re.MULTILINE)
action_pattern = re.compile(r'```action\s*\n(.*?)\n```', re.DOTALL)
condition_pattern = re.compile(r'```condition\s*\n(.*?)\n```', re.DOTALL)
# Find all step headers
matches = list(step_pattern.finditer(body))
for i, match in enumerate(matches):
step_name = match.group(1).strip()
# Get content until next step or end
start = match.end()
end = matches[i + 1].start() if i + 1 < len(matches) else len(body)
step_content = body[start:end].strip()
# Extract action
action_match = action_pattern.search(step_content)
action = action_match.group(1).strip() if action_match else ""
# Extract condition
condition_match = condition_pattern.search(step_content)
condition = condition_match.group(1).strip() if condition_match else None
# Get description (text before action block)
description = step_content
if action_match:
description = step_content[:action_match.start()].strip()
# Clean up description (remove code blocks)
description = re.sub(r'```.*?```', '', description, flags=re.DOTALL).strip()
steps.append(WorkflowStep(
name=step_name,
description=description,
action=action,
condition=condition
))
return steps
def _load_workflow(self, file_path: Path) -> Optional[Workflow]:
"""Load a workflow from a file."""
try:
content = file_path.read_text(encoding="utf-8")
frontmatter, body = self._parse_frontmatter(content)
name = frontmatter.get("name", file_path.stem)
description = frontmatter.get("description", "")
variables = frontmatter.get("variables", {})
# Parse variables if it's a string (simple YAML)
if isinstance(variables, str):
variables = {}
steps = self._parse_steps(body)
workflow = Workflow(
name=name,
description=description,
steps=steps,
variables=variables,
file_path=str(file_path)
)
self._log(f"Loaded workflow '{name}' with {len(steps)} steps")
return workflow
except Exception as e:
self._log(f"Error loading workflow {file_path}: {e}", logging.WARNING)
return None
def _discover_workflows(self):
"""Discover all workflows in the workspace."""
self._workflows = {}
workflows_dir = self.workspace_path / self.WORKFLOWS_DIR.replace("/", os.sep)
if not workflows_dir.exists():
return
for ext in self.SUPPORTED_EXTENSIONS:
for file_path in workflows_dir.glob(f"*{ext}"):
workflow = self._load_workflow(file_path)
if workflow:
self._workflows[workflow.name.lower()] = workflow
self._log(f"Discovered {len(self._workflows)} workflows")
def reload(self):
"""Reload all workflows from disk."""
self._loaded = False
self._ensure_loaded()
def list_workflows(self) -> List[Workflow]:
"""List all available workflows."""
self._ensure_loaded()
return list(self._workflows.values())
def get_workflow(self, name: str) -> Optional[Workflow]:
"""Get a workflow by name."""
self._ensure_loaded()
return self._workflows.get(name.lower())
def _substitute_variables(
self,
text: str,
variables: Dict[str, Any]
) -> str:
"""Substitute {{variable}} placeholders in text."""
def replace(match):
var_name = match.group(1).strip()
return str(variables.get(var_name, match.group(0)))
return re.sub(r'\{\{(\w+)\}\}', replace, text)
def execute(
self,
workflow_name: str,
executor: Callable[[str], str],
variables: Optional[Dict[str, Any]] = None,
on_step: Optional[Callable[[WorkflowStep, int], None]] = None,
on_result: Optional[Callable[[WorkflowStep, str], None]] = None
) -> Dict[str, Any]:
"""
Execute a workflow.
Args:
workflow_name: Name of the workflow to execute
executor: Function to execute each step (e.g., agent.chat)
variables: Variables to substitute in steps
on_step: Callback before each step (step, index)
on_result: Callback after each step (step, result)
Returns:
Execution results with step outputs and status
"""
workflow = self.get_workflow(workflow_name)
if not workflow:
return {
"success": False,
"error": f"Workflow '{workflow_name}' not found",
"results": []
}
# Merge variables
all_variables = {**workflow.variables}
if variables:
all_variables.update(variables)
results = []
success = True
for i, step in enumerate(workflow.steps):
# Check condition
if step.condition:
condition = self._substitute_variables(step.condition, all_variables)
# Simple condition evaluation (could be enhanced)
if condition.lower() in ("false", "no", "skip", "0"):
results.append({
"step": step.name,
"status": "skipped",
"output": None
})
continue
# Callback before step
if on_step:
on_step(step, i)
# Substitute variables in action
action = self._substitute_variables(step.action, all_variables)
# Execute step
retries = 0
step_success = False
output = None
error = None
while retries <= step.max_retries and not step_success:
try:
output = executor(action)
step_success = True
except Exception as e:
error = str(e)
retries += 1
if retries <= step.max_retries:
self._log(f"Step '{step.name}' failed, retrying ({retries}/{step.max_retries})")
# Callback after step
if on_result and output:
on_result(step, output)
results.append({
"step": step.name,
"status": "success" if step_success else "failed",
"output": output,
"error": error
})
# Handle failure
if not step_success:
if step.on_error == "stop":
success = False
break
elif step.on_error == "continue":
continue
return {
"success": success,
"workflow": workflow.name,
"results": results
}
def create_workflow(
self,
name: str,
description: str = "",
steps: Optional[List[Dict[str, str]]] = None,
variables: Optional[Dict[str, Any]] = None
) -> Workflow:
"""
Create a new workflow file.
Args:
name: Workflow name
description: Workflow description
steps: List of step dicts with name, description, action
variables: Default variables
Returns:
Created Workflow object
"""
workflows_dir = self.workspace_path / self.WORKFLOWS_DIR.replace("/", os.sep)
workflows_dir.mkdir(parents=True, exist_ok=True)
file_path = workflows_dir / f"{name.lower().replace(' ', '_')}.md"
# Build content
lines = ["---"]
lines.append(f"name: {name}")
if description:
lines.append(f"description: {description}")
if variables:
lines.append("variables:")
for k, v in variables.items():
lines.append(f" {k}: {v}")
lines.append("---")
lines.append("")
# Add steps
workflow_steps = []
if steps:
for i, step in enumerate(steps, 1):
step_name = step.get("name", f"Step {i}")
step_desc = step.get("description", "")
step_action = step.get("action", "")
lines.append(f"## Step {i}: {step_name}")
if step_desc:
lines.append(step_desc)
lines.append("")
lines.append("```action")
lines.append(step_action)
lines.append("```")
lines.append("")
workflow_steps.append(WorkflowStep(
name=step_name,
description=step_desc,
action=step_action
))
# Write file
file_path.write_text("\n".join(lines), encoding="utf-8")
# Create and register workflow
workflow = Workflow(
name=name,
description=description,
steps=workflow_steps,
variables=variables or {},
file_path=str(file_path)
)
self._workflows[name.lower()] = workflow
self._log(f"Created workflow '{name}' at {file_path}")
return workflow
def get_stats(self) -> Dict[str, Any]:
"""Get workflow statistics."""
self._ensure_loaded()
total_steps = sum(len(w.steps) for w in self._workflows.values())
return {
"total_workflows": len(self._workflows),
"total_steps": total_steps,
"workflows": [w.name for w in self._workflows.values()]
}
def create_workflow_manager(
workspace_path: Optional[str] = None,
**kwargs
) -> WorkflowManager:
"""
Create a WorkflowManager instance.
Args:
workspace_path: Path to workspace
**kwargs: Additional configuration
Returns:
WorkflowManager instance
"""
return WorkflowManager(workspace_path=workspace_path, **kwargs)
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "0.0.178"
version = "0.0.180"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
requires-python = ">=3.10"
authors = [
@@ -651,3 +651,179 @@ class TestMemoryIntegrationWithOpenAI:
memory_path = f".praison/memory/{test_user_id}"
if os.path.exists(memory_path):
shutil.rmtree(memory_path)
class TestMultiAgentMemorySharing:
"""Tests for memory sharing between multiple agents."""
def test_agents_share_memory_with_same_user_id(self):
"""Test that agents with same user_id share memory."""
from praisonaiagents import Agent
with tempfile.TemporaryDirectory() as tmpdir:
user_id = "shared_user"
base_path = f"{tmpdir}/memory"
# Agent 1 stores memories
agent1 = Agent(
name="Agent1",
memory={"provider": "file", "user_id": user_id, "base_path": base_path},
verbose=False
)
agent1.store_memory("User prefers Python", memory_type="short_term")
agent1.store_memory("User name is Alice", memory_type="long_term", importance=0.9)
agent1._memory_instance.add_entity("Alice", "person", {"role": "developer"})
# Verify Agent1 stored memories
assert len(agent1._memory_instance.get_short_term()) == 1
assert len(agent1._memory_instance.get_long_term()) == 1
assert len(agent1._memory_instance.get_all_entities()) == 1
# Agent 2 with same user_id should load memories
agent2 = Agent(
name="Agent2",
memory={"provider": "file", "user_id": user_id, "base_path": base_path},
verbose=False
)
# Verify Agent2 loaded the same memories
assert len(agent2._memory_instance.get_short_term()) == 1
assert len(agent2._memory_instance.get_long_term()) == 1
assert len(agent2._memory_instance.get_all_entities()) == 1
# Verify content matches
assert agent2._memory_instance.get_short_term()[0].content == "User prefers Python"
assert agent2._memory_instance.get_long_term()[0].content == "User name is Alice"
assert agent2._memory_instance.get_entity("Alice") is not None
def test_agents_isolated_with_different_user_id(self):
"""Test that agents with different user_id have isolated memory."""
from praisonaiagents import Agent
with tempfile.TemporaryDirectory() as tmpdir:
base_path = f"{tmpdir}/memory"
# Agent 1 with user_id "user1"
agent1 = Agent(
name="Agent1",
memory={"provider": "file", "user_id": "user1", "base_path": base_path},
verbose=False
)
agent1.store_memory("User1 data", memory_type="long_term", importance=0.9)
# Agent 2 with different user_id "user2"
agent2 = Agent(
name="Agent2",
memory={"provider": "file", "user_id": "user2", "base_path": base_path},
verbose=False
)
# Agent2 should have no memories (different user_id)
assert len(agent2._memory_instance.get_long_term()) == 0
assert len(agent2._memory_instance.get_short_term()) == 0
# Verify paths are different
assert agent1._memory_instance.user_path != agent2._memory_instance.user_path
def test_memory_context_injected_into_system_prompt(self):
"""Test that memory context is injected into system prompt."""
from praisonaiagents import Agent
with tempfile.TemporaryDirectory() as tmpdir:
agent = Agent(
name="TestAgent",
memory={"provider": "file", "user_id": "test", "base_path": f"{tmpdir}/memory"},
verbose=False
)
# Store memories
agent.store_memory("User prefers dark mode", memory_type="short_term")
agent.store_memory("User name is John", memory_type="long_term", importance=0.9)
# Get memory context
context = agent.get_memory_context()
# Verify context contains the memories
assert "John" in context
assert "dark mode" in context
assert "Important Facts" in context or "Recent Context" in context
def test_memory_display_info(self):
"""Test that memory display info shows correct counts."""
from praisonaiagents import Agent
with tempfile.TemporaryDirectory() as tmpdir:
agent = Agent(
name="TestAgent",
memory={"provider": "file", "user_id": "test", "base_path": f"{tmpdir}/memory"},
verbose=True
)
# Store various memories
agent.store_memory("Short term 1", memory_type="short_term")
agent.store_memory("Short term 2", memory_type="short_term")
agent.store_memory("Long term 1", memory_type="long_term", importance=0.9)
agent._memory_instance.add_entity("Entity1", "person", {})
# Get stats
stats = agent._memory_instance.get_stats()
# Verify counts are as expected (at least the ones we added)
assert stats["short_term_count"] >= 2
assert stats["long_term_count"] >= 1
assert stats["entity_count"] >= 1
class TestMultiAgentMemoryIntegration:
"""Integration tests for multi-agent memory with OpenAI API."""
def test_multi_agent_memory_sharing_with_llm(self):
"""Test that multiple agents share memory and use it in LLM calls."""
import os
import shutil
# Skip if no API key
if not os.getenv("OPENAI_API_KEY"):
pytest.skip("OPENAI_API_KEY not set")
from praisonaiagents import Agent
test_user_id = "multi_agent_llm_test"
try:
# Agent 1 stores memories
agent1 = Agent(
name="Agent1",
memory=True,
user_id=test_user_id,
verbose=False
)
agent1.store_memory("User's favorite color is blue", memory_type="long_term", importance=0.9)
agent1.store_memory("User works as a teacher", memory_type="long_term", importance=0.85)
# Agent 2 with same user_id should recall memories
agent2 = Agent(
name="Agent2",
memory=True,
user_id=test_user_id,
verbose=False
)
# Verify memories loaded
stats = agent2._memory_instance.get_stats()
assert stats["long_term_count"] == 2
# Ask Agent2 about the user - should use loaded memories
result = agent2.start("What is my favorite color? Answer in one word.")
assert result is not None
assert "blue" in result.lower(), f"Expected 'blue' in response: {result}"
print(f"✅ Multi-agent memory sharing test passed! Response: {result}")
finally:
memory_path = f".praison/memory/{test_user_id}"
if os.path.exists(memory_path):
shutil.rmtree(memory_path)
@@ -0,0 +1,569 @@
"""
Tests for advanced memory features:
- Session Save/Resume
- Context Compression
- Checkpointing
- Slash Commands
- Rules Manager
"""
import os
import sys
import pytest
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from praisonaiagents.memory import FileMemory, RulesManager, Rule
class TestSessionManagement:
"""Tests for session save/resume functionality."""
@pytest.fixture
def memory(self, tmp_path):
"""Create a FileMemory instance with temp directory."""
return FileMemory(user_id="test_user", base_path=str(tmp_path / "memory"))
def test_save_session(self, memory):
"""Test saving a session."""
# Add some memories
memory.add_short_term("User prefers Python")
memory.add_short_term("Working on AI project")
memory.add_long_term("User name is John")
# Save session
path = memory.save_session("test_session")
assert Path(path).exists()
assert "test_session.json" in path
def test_save_session_with_conversation(self, memory):
"""Test saving session with conversation history."""
conversation = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"}
]
path = memory.save_session("conv_session", conversation_history=conversation)
assert Path(path).exists()
def test_resume_session(self, memory):
"""Test resuming a saved session."""
# Add memories and save
memory.add_short_term("Important context")
memory.save_session("resume_test")
# Clear and verify empty
memory.clear_short_term()
assert len(memory.get_short_term()) == 0
# Resume session
memory.resume_session("resume_test")
# Verify restored
assert len(memory.get_short_term()) > 0
assert "Important context" in [m.content for m in memory.get_short_term()]
def test_list_sessions(self, memory):
"""Test listing saved sessions."""
memory.save_session("session1")
memory.save_session("session2")
sessions = memory.list_sessions()
assert len(sessions) >= 2
names = [s["name"] for s in sessions]
assert "session1" in names
assert "session2" in names
def test_delete_session(self, memory):
"""Test deleting a session."""
memory.save_session("to_delete")
assert memory.delete_session("to_delete")
sessions = memory.list_sessions()
names = [s["name"] for s in sessions]
assert "to_delete" not in names
def test_resume_nonexistent_session(self, memory):
"""Test resuming a session that doesn't exist."""
with pytest.raises(FileNotFoundError):
memory.resume_session("nonexistent")
class TestContextCompression:
"""Tests for context compression functionality."""
@pytest.fixture
def memory(self, tmp_path):
"""Create a FileMemory instance with temp directory."""
return FileMemory(
user_id="test_user",
base_path=str(tmp_path / "memory"),
config={"short_term_limit": 20}
)
def test_compress_basic(self, memory):
"""Test basic compression without LLM."""
# Add many items
for i in range(15):
memory.add_short_term(f"Context item {i}")
# Compress
summary = memory.compress(max_items=5)
# Verify compression
assert len(memory.get_short_term()) <= 5
assert len(summary) > 0
def test_compress_with_llm_func(self, memory):
"""Test compression with custom LLM function."""
for i in range(15):
memory.add_short_term(f"Item {i}")
# Mock LLM function
def mock_llm(prompt):
return "Summary: Multiple items discussed."
summary = memory.compress(llm_func=mock_llm, max_items=5)
assert "Summary:" in summary
def test_auto_compress_below_threshold(self, memory):
"""Test auto-compress doesn't trigger below threshold."""
# Add few items (below 70% of 20 = 14)
for i in range(10):
memory.add_short_term(f"Item {i}")
result = memory.auto_compress_if_needed(threshold_percent=0.7)
assert result is None
def test_auto_compress_above_threshold(self, memory):
"""Test auto-compress triggers above threshold."""
# Add many items (above 70% of 20 = 14)
for i in range(16):
memory.add_short_term(f"Item {i}")
result = memory.auto_compress_if_needed(threshold_percent=0.7)
assert result is not None or len(memory.get_short_term()) < 16
class TestCheckpointing:
"""Tests for checkpointing functionality."""
@pytest.fixture
def memory(self, tmp_path):
"""Create a FileMemory instance with temp directory."""
return FileMemory(user_id="test_user", base_path=str(tmp_path / "memory"))
def test_create_checkpoint(self, memory):
"""Test creating a checkpoint."""
memory.add_short_term("Before checkpoint")
memory.add_long_term("Important fact")
checkpoint_id = memory.create_checkpoint("test_checkpoint")
assert checkpoint_id == "test_checkpoint"
checkpoints = memory.list_checkpoints()
assert any(c["id"] == "test_checkpoint" for c in checkpoints)
def test_create_checkpoint_auto_name(self, memory):
"""Test creating checkpoint with auto-generated name."""
checkpoint_id = memory.create_checkpoint()
assert checkpoint_id.startswith("checkpoint_")
def test_restore_checkpoint(self, memory):
"""Test restoring from checkpoint."""
# Add initial data
memory.add_short_term("Original data")
memory.create_checkpoint("restore_test")
# Modify data
memory.clear_short_term()
memory.add_short_term("New data")
# Restore
success = memory.restore_checkpoint("restore_test")
assert success
contents = [m.content for m in memory.get_short_term()]
assert "Original data" in contents
def test_checkpoint_with_files(self, memory, tmp_path):
"""Test checkpoint with file snapshots."""
# Create a test file
test_file = tmp_path / "test_file.txt"
test_file.write_text("Original content")
# Create checkpoint with file
checkpoint_id = memory.create_checkpoint(
"file_checkpoint",
include_files=[str(test_file)]
)
# Modify file
test_file.write_text("Modified content")
# Restore with files
memory.restore_checkpoint(checkpoint_id, restore_files=True)
assert test_file.read_text() == "Original content"
def test_list_checkpoints(self, memory):
"""Test listing checkpoints."""
memory.create_checkpoint("cp1")
memory.create_checkpoint("cp2")
checkpoints = memory.list_checkpoints()
assert len(checkpoints) >= 2
ids = [c["id"] for c in checkpoints]
assert "cp1" in ids
assert "cp2" in ids
def test_delete_checkpoint(self, memory):
"""Test deleting a checkpoint."""
memory.create_checkpoint("to_delete")
assert memory.delete_checkpoint("to_delete")
checkpoints = memory.list_checkpoints()
ids = [c["id"] for c in checkpoints]
assert "to_delete" not in ids
class TestSlashCommands:
"""Tests for slash command handling."""
@pytest.fixture
def memory(self, tmp_path):
"""Create a FileMemory instance with temp directory."""
return FileMemory(user_id="test_user", base_path=str(tmp_path / "memory"))
def test_command_show(self, memory):
"""Test /memory show command."""
memory.add_short_term("Test item")
result = memory.handle_command("/memory show")
assert result["action"] == "show"
assert "stats" in result
assert "short_term" in result
def test_command_add(self, memory):
"""Test /memory add command."""
result = memory.handle_command("/memory add User likes coffee")
assert result["action"] == "add"
assert result["content"] == "User likes coffee"
# Verify it was added
long_term = memory.get_long_term()
assert any("coffee" in m.content for m in long_term)
def test_command_clear(self, memory):
"""Test /memory clear command."""
memory.add_short_term("To be cleared")
result = memory.handle_command("/memory clear short")
assert result["action"] == "clear"
assert len(memory.get_short_term()) == 0
def test_command_search(self, memory):
"""Test /memory search command."""
memory.add_long_term("Python programming")
result = memory.handle_command("/memory search Python")
assert result["action"] == "search"
assert "results" in result
def test_command_save(self, memory):
"""Test /memory save command."""
result = memory.handle_command("/memory save my_session")
assert result["action"] == "save"
assert result["session"] == "my_session"
def test_command_sessions(self, memory):
"""Test /memory sessions command."""
memory.save_session("test")
result = memory.handle_command("/memory sessions")
assert result["action"] == "sessions"
assert "sessions" in result
def test_command_checkpoint(self, memory):
"""Test /memory checkpoint command."""
result = memory.handle_command("/memory checkpoint my_cp")
assert result["action"] == "checkpoint"
assert result["id"] == "my_cp"
def test_command_checkpoints(self, memory):
"""Test /memory checkpoints command."""
memory.create_checkpoint("test")
result = memory.handle_command("/memory checkpoints")
assert result["action"] == "checkpoints"
assert "checkpoints" in result
def test_command_help(self, memory):
"""Test /memory help command."""
result = memory.handle_command("/memory help")
assert result["action"] == "help"
assert "commands" in result
assert "/memory show" in result["commands"]
def test_command_invalid(self, memory):
"""Test invalid command."""
result = memory.handle_command("/memory invalid_action")
assert "error" in result
class TestRulesManager:
"""Tests for RulesManager functionality."""
@pytest.fixture
def rules_dir(self, tmp_path):
"""Create a temp rules directory."""
rules_path = tmp_path / ".praison" / "rules"
rules_path.mkdir(parents=True)
return rules_path
@pytest.fixture
def manager(self, tmp_path, rules_dir):
"""Create a RulesManager instance."""
return RulesManager(workspace_path=str(tmp_path))
def test_create_rule(self, manager, tmp_path):
"""Test creating a rule."""
rule = manager.create_rule(
name="python_style",
content="# Python Style\n- Use type hints\n- Follow PEP 8",
description="Python coding guidelines",
globs=["**/*.py"],
activation="glob"
)
assert rule.name == "python_style"
assert rule.activation == "glob"
assert "**/*.py" in rule.globs
# Verify file was created
rule_file = tmp_path / ".praison" / "rules" / "python_style.md"
assert rule_file.exists()
def test_load_rule_with_frontmatter(self, rules_dir, tmp_path):
"""Test loading a rule with YAML frontmatter."""
rule_content = """---
description: Test rule
globs: ["*.py", "*.pyx"]
activation: glob
priority: 10
---
# Test Rule
- Rule content here
"""
(rules_dir / "test_rule.md").write_text(rule_content)
manager = RulesManager(workspace_path=str(tmp_path))
rule = manager.get_rule_by_name("test_rule")
assert rule is not None
assert rule.description == "Test rule"
assert "*.py" in rule.globs
assert rule.activation == "glob"
assert rule.priority == 10
def test_get_active_rules_always(self, manager):
"""Test getting always-active rules."""
manager.create_rule(
name="always_rule",
content="Always active",
activation="always"
)
active = manager.get_active_rules()
assert any(r.name == "always_rule" for r in active)
def test_get_rules_for_file_glob(self, manager):
"""Test getting rules for a specific file."""
manager.create_rule(
name="py_rule",
content="Python rule",
globs=["**/*.py"],
activation="glob"
)
manager.create_rule(
name="js_rule",
content="JS rule",
globs=["**/*.js"],
activation="glob"
)
py_rules = manager.get_rules_for_file("src/main.py")
rule_names = [r.name for r in py_rules]
assert "py_rule" in rule_names
# js_rule should not match .py files
def test_get_manual_rules(self, manager):
"""Test getting manual-only rules."""
manager.create_rule(
name="manual_rule",
content="Manual only",
activation="manual"
)
manual = manager.get_manual_rules()
assert any(r.name == "manual_rule" for r in manual)
def test_build_rules_context(self, manager):
"""Test building context string from rules."""
manager.create_rule(
name="rule1",
content="Rule 1 content",
description="First rule",
activation="always"
)
manager.create_rule(
name="rule2",
content="Rule 2 content",
description="Second rule",
activation="always"
)
context = manager.build_rules_context()
assert "rule1" in context
assert "Rule 1 content" in context
def test_build_rules_context_with_manual(self, manager):
"""Test building context with manual rules included."""
manager.create_rule(
name="auto_rule",
content="Auto content",
activation="always"
)
manager.create_rule(
name="manual_rule",
content="Manual content",
activation="manual"
)
# Without manual
context1 = manager.build_rules_context()
assert "manual_rule" not in context1
# With manual included
context2 = manager.build_rules_context(include_manual=["manual_rule"])
assert "Manual content" in context2
def test_delete_rule(self, manager):
"""Test deleting a rule."""
manager.create_rule(name="to_delete", content="Delete me")
assert manager.delete_rule("to_delete")
assert manager.get_rule_by_name("to_delete") is None
def test_get_stats(self, manager):
"""Test getting rules statistics."""
manager.create_rule(name="r1", content="c1", activation="always")
manager.create_rule(name="r2", content="c2", activation="glob", globs=["*.py"])
manager.create_rule(name="r3", content="c3", activation="manual")
stats = manager.get_stats()
assert stats["total_rules"] >= 3
assert stats["always_rules"] >= 1
assert stats["glob_rules"] >= 1
assert stats["manual_rules"] >= 1
def test_reload_rules(self, manager, rules_dir):
"""Test reloading rules from disk."""
# Create rule via manager
manager.create_rule(name="initial", content="Initial")
# Create rule directly on disk
new_rule = """---
description: New rule
activation: always
---
New rule content
"""
(rules_dir / "new_rule.md").write_text(new_rule)
# Reload
manager.reload()
# Should find new rule
assert manager.get_rule_by_name("new_rule") is not None
class TestRuleMatching:
"""Tests for rule glob pattern matching."""
def test_rule_matches_exact_glob(self):
"""Test exact glob matching."""
rule = Rule(
name="test",
content="test",
globs=["*.py"],
activation="glob"
)
assert rule.matches_file("main.py")
assert not rule.matches_file("main.js")
def test_rule_matches_recursive_glob(self):
"""Test recursive glob matching."""
rule = Rule(
name="test",
content="test",
globs=["**/*.py"],
activation="glob"
)
assert rule.matches_file("src/main.py")
assert rule.matches_file("src/utils/helper.py")
def test_rule_always_matches(self):
"""Test always activation mode."""
rule = Rule(
name="test",
content="test",
activation="always"
)
assert rule.matches_file("anything.txt")
assert rule.matches_file("src/main.py")
def test_rule_manual_never_matches(self):
"""Test manual activation mode never auto-matches."""
rule = Rule(
name="test",
content="test",
activation="manual"
)
assert not rule.matches_file("anything.txt")
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -0,0 +1,570 @@
"""
Comprehensive tests for new memory features.
Tests:
- CLAUDE.local.md and local override files
- .claude/rules/, .windsurf/rules/, .cursor/rules/ discovery
- Git root discovery
- @Import syntax
- Rules character limit (12000)
- ai_decision activation mode
- Auto-generated memories
- Workflows
- Hooks
"""
import sys
import json
import tempfile
import unittest
from pathlib import Path
# Add parent to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
from praisonaiagents.memory import (
FileMemory,
RulesManager,
)
class TestLocalOverrideFiles(unittest.TestCase):
"""Test CLAUDE.local.md and PRAISON.local.md support."""
def test_claude_local_md_loaded(self):
"""Test that CLAUDE.local.md is discovered."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create CLAUDE.md
claude_md = Path(tmpdir) / "CLAUDE.md"
claude_md.write_text("# Main Claude Rules\n- Rule 1")
# Create CLAUDE.local.md (higher priority)
claude_local = Path(tmpdir) / "CLAUDE.local.md"
claude_local.write_text("# Local Overrides\n- Override 1")
manager = RulesManager(workspace_path=tmpdir)
self.assertIn("root:claude", manager._rules)
self.assertIn("root:claude_local", manager._rules)
# Local should have higher priority
claude_rule = manager._rules["root:claude"]
local_rule = manager._rules["root:claude_local"]
self.assertGreater(local_rule.priority, claude_rule.priority)
def test_praison_local_md_loaded(self):
"""Test that PRAISON.local.md is discovered."""
with tempfile.TemporaryDirectory() as tmpdir:
praison_local = Path(tmpdir) / "PRAISON.local.md"
praison_local.write_text("# Local Praison Rules")
manager = RulesManager(workspace_path=tmpdir)
self.assertIn("root:praison_local", manager._rules)
class TestAdditionalRulesDirectories(unittest.TestCase):
"""Test .claude/rules/, .windsurf/rules/, .cursor/rules/ discovery."""
def test_claude_rules_directory(self):
"""Test .claude/rules/ discovery."""
with tempfile.TemporaryDirectory() as tmpdir:
rules_dir = Path(tmpdir) / ".claude" / "rules"
rules_dir.mkdir(parents=True)
(rules_dir / "python.md").write_text("# Python Rules\n- Use type hints")
(rules_dir / "testing.md").write_text("# Testing Rules\n- Use pytest")
manager = RulesManager(workspace_path=tmpdir)
stats = manager.get_stats()
self.assertGreaterEqual(stats["total_rules"], 2)
self.assertIn("claude:python", manager._rules)
self.assertIn("claude:testing", manager._rules)
def test_windsurf_rules_directory(self):
"""Test .windsurf/rules/ discovery."""
with tempfile.TemporaryDirectory() as tmpdir:
rules_dir = Path(tmpdir) / ".windsurf" / "rules"
rules_dir.mkdir(parents=True)
(rules_dir / "style.md").write_text("# Style Rules")
manager = RulesManager(workspace_path=tmpdir)
self.assertIn("windsurf:style", manager._rules)
def test_cursor_rules_directory(self):
"""Test .cursor/rules/ discovery."""
with tempfile.TemporaryDirectory() as tmpdir:
rules_dir = Path(tmpdir) / ".cursor" / "rules"
rules_dir.mkdir(parents=True)
(rules_dir / "code.mdc").write_text("# Code Rules")
manager = RulesManager(workspace_path=tmpdir)
self.assertIn("cursor:code", manager._rules)
class TestGitRootDiscovery(unittest.TestCase):
"""Test git root discovery for monorepo support."""
def test_find_git_root(self):
"""Test _find_git_root method."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create fake git repo
git_dir = Path(tmpdir) / ".git"
git_dir.mkdir()
# Create subdirectory
subdir = Path(tmpdir) / "packages" / "frontend"
subdir.mkdir(parents=True)
manager = RulesManager(workspace_path=str(subdir))
git_root = manager._find_git_root()
# Resolve both paths to handle /private/var vs /var on macOS
self.assertEqual(git_root.resolve(), Path(tmpdir).resolve())
def test_rules_from_git_root(self):
"""Test that rules are loaded from git root."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create fake git repo
git_dir = Path(tmpdir) / ".git"
git_dir.mkdir()
# Create CLAUDE.md at git root
(Path(tmpdir) / "CLAUDE.md").write_text("# Root Rules")
# Create subdirectory
subdir = Path(tmpdir) / "packages" / "frontend"
subdir.mkdir(parents=True)
# Create CLAUDE.md in subdirectory
(subdir / "CLAUDE.md").write_text("# Subdir Rules")
manager = RulesManager(workspace_path=str(subdir))
# Should have both root and gitroot rules
self.assertIn("root:claude", manager._rules)
class TestImportSyntax(unittest.TestCase):
"""Test @import syntax for including other files."""
def test_basic_import(self):
"""Test basic @path/to/file import."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create file to import
docs_dir = Path(tmpdir) / "docs"
docs_dir.mkdir()
(docs_dir / "guidelines.md").write_text("# Guidelines\n- Be concise")
# Create CLAUDE.md with import
claude_md = Path(tmpdir) / "CLAUDE.md"
claude_md.write_text("# Main Rules\n@docs/guidelines.md\n- More rules")
manager = RulesManager(workspace_path=tmpdir)
rule = manager._rules.get("root:claude")
self.assertIsNotNone(rule)
self.assertIn("Be concise", rule.content)
def test_relative_import(self):
"""Test @./relative/path import."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create file to import
(Path(tmpdir) / "shared.md").write_text("# Shared Content")
# Create CLAUDE.md with relative import
claude_md = Path(tmpdir) / "CLAUDE.md"
claude_md.write_text("Include: @./shared.md")
manager = RulesManager(workspace_path=tmpdir)
rule = manager._rules.get("root:claude")
self.assertIn("Shared Content", rule.content)
def test_import_depth_limit(self):
"""Test that import depth is limited to prevent loops."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create circular imports
(Path(tmpdir) / "a.md").write_text("A imports @./b.md")
(Path(tmpdir) / "b.md").write_text("B imports @./c.md")
(Path(tmpdir) / "c.md").write_text("C imports @./a.md") # Circular
claude_md = Path(tmpdir) / "CLAUDE.md"
claude_md.write_text("Start: @./a.md")
# Should not hang or crash
manager = RulesManager(workspace_path=tmpdir)
rule = manager._rules.get("root:claude")
self.assertIsNotNone(rule)
def test_code_span_not_imported(self):
"""Test that @mentions in code spans are not imported."""
with tempfile.TemporaryDirectory() as tmpdir:
claude_md = Path(tmpdir) / "CLAUDE.md"
claude_md.write_text("Use `@org/package` for imports")
manager = RulesManager(workspace_path=tmpdir)
rule = manager._rules.get("root:claude")
# Should not try to import @org/package
self.assertIn("@org/package", rule.content)
class TestRulesCharacterLimit(unittest.TestCase):
"""Test 12000 character limit for rules."""
def test_truncation(self):
"""Test that long rules are truncated."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create very long rule
long_content = "x" * 15000
claude_md = Path(tmpdir) / "CLAUDE.md"
claude_md.write_text(long_content)
manager = RulesManager(workspace_path=tmpdir)
rule = manager._rules.get("root:claude")
self.assertLessEqual(len(rule.content), 12000)
self.assertIn("truncated", rule.content)
def test_short_rules_not_truncated(self):
"""Test that short rules are not truncated."""
with tempfile.TemporaryDirectory() as tmpdir:
short_content = "# Short Rule\n- Item 1"
claude_md = Path(tmpdir) / "CLAUDE.md"
claude_md.write_text(short_content)
manager = RulesManager(workspace_path=tmpdir)
rule = manager._rules.get("root:claude")
self.assertNotIn("truncated", rule.content)
class TestAiDecisionActivation(unittest.TestCase):
"""Test ai_decision activation mode."""
def test_ai_decision_rule_loaded(self):
"""Test that ai_decision rules are loaded."""
with tempfile.TemporaryDirectory() as tmpdir:
rules_dir = Path(tmpdir) / ".praison" / "rules"
rules_dir.mkdir(parents=True)
rule_content = """---
description: Security guidelines
activation: ai_decision
---
# Security Rules
- Validate inputs
"""
(rules_dir / "security.md").write_text(rule_content)
manager = RulesManager(workspace_path=tmpdir)
self.assertIn("workspace:security", manager._rules)
rule = manager._rules["workspace:security"]
self.assertEqual(rule.activation, "ai_decision")
def test_evaluate_ai_decision_without_llm(self):
"""Test ai_decision evaluation without LLM (defaults to True)."""
with tempfile.TemporaryDirectory() as tmpdir:
rules_dir = Path(tmpdir) / ".praison" / "rules"
rules_dir.mkdir(parents=True)
rule_content = """---
activation: ai_decision
description: Security rules
---
# Security
"""
(rules_dir / "security.md").write_text(rule_content)
manager = RulesManager(workspace_path=tmpdir)
rule = manager._rules["workspace:security"]
# Without LLM, should return True
result = manager.evaluate_ai_decision(rule, "some context")
self.assertTrue(result)
class TestAutoMemory(unittest.TestCase):
"""Test auto-generated memories."""
def test_extract_name(self):
"""Test name extraction from text."""
from praisonaiagents.memory import AutoMemoryExtractor
extractor = AutoMemoryExtractor()
memories = extractor.extract("My name is John and I work at Acme.")
names = [m for m in memories if m["type"] == "name"]
self.assertGreater(len(names), 0)
self.assertEqual(names[0]["content"], "John")
def test_extract_preference(self):
"""Test preference extraction."""
from praisonaiagents.memory import AutoMemoryExtractor
extractor = AutoMemoryExtractor()
memories = extractor.extract("I prefer Python for backend development.")
prefs = [m for m in memories if m["type"] == "preference"]
self.assertGreater(len(prefs), 0)
def test_extract_technology(self):
"""Test technology extraction."""
from praisonaiagents.memory import AutoMemoryExtractor
extractor = AutoMemoryExtractor()
memories = extractor.extract("I'm using Python and TypeScript.")
techs = [m for m in memories if m["type"] == "technology"]
self.assertGreater(len(techs), 0)
def test_should_remember(self):
"""Test quick filter for memorable content."""
from praisonaiagents.memory import AutoMemoryExtractor
extractor = AutoMemoryExtractor()
self.assertTrue(extractor.should_remember("My name is John"))
self.assertTrue(extractor.should_remember("I prefer dark mode"))
self.assertFalse(extractor.should_remember("Hello world"))
def test_auto_memory_wrapper(self):
"""Test AutoMemory wrapper with FileMemory."""
from praisonaiagents.memory import AutoMemory
with tempfile.TemporaryDirectory() as tmpdir:
memory = FileMemory(user_id="test", base_path=f"{tmpdir}/memory")
auto = AutoMemory(memory, enabled=True)
# Process interaction
memories = auto.process_interaction(
"My name is Alice and I prefer Python",
store=True
)
self.assertGreater(len(memories), 0)
# Check stored in base memory
stats = memory.get_stats()
self.assertGreater(stats["long_term_count"] + stats["entity_count"], 0)
class TestWorkflows(unittest.TestCase):
"""Test workflow system."""
def test_workflow_discovery(self):
"""Test workflow discovery from .praison/workflows/."""
from praisonaiagents.memory import WorkflowManager
with tempfile.TemporaryDirectory() as tmpdir:
workflows_dir = Path(tmpdir) / ".praison" / "workflows"
workflows_dir.mkdir(parents=True)
workflow_content = """---
name: Deploy
description: Deploy to production
---
## Step 1: Test
Run tests.
```action
Run pytest
```
## Step 2: Build
Build the app.
```action
Build application
```
"""
(workflows_dir / "deploy.md").write_text(workflow_content)
manager = WorkflowManager(workspace_path=tmpdir)
workflows = manager.list_workflows()
self.assertEqual(len(workflows), 1)
self.assertEqual(workflows[0].name, "Deploy")
self.assertEqual(len(workflows[0].steps), 2)
def test_workflow_execution(self):
"""Test workflow execution."""
from praisonaiagents.memory import WorkflowManager
with tempfile.TemporaryDirectory() as tmpdir:
workflows_dir = Path(tmpdir) / ".praison" / "workflows"
workflows_dir.mkdir(parents=True)
workflow_content = """---
name: Test
---
## Step 1: Echo
Echo environment.
```action
Echo staging
```
"""
(workflows_dir / "test.md").write_text(workflow_content)
manager = WorkflowManager(workspace_path=tmpdir)
# Mock executor
executed = []
def executor(prompt):
executed.append(prompt)
return f"Executed: {prompt}"
result = manager.execute("test", executor=executor)
self.assertTrue(result["success"])
self.assertEqual(len(executed), 1)
self.assertIn("staging", executed[0])
def test_create_workflow(self):
"""Test programmatic workflow creation."""
from praisonaiagents.memory import WorkflowManager
with tempfile.TemporaryDirectory() as tmpdir:
manager = WorkflowManager(workspace_path=tmpdir)
workflow = manager.create_workflow(
name="My Workflow",
description="Test workflow",
steps=[
{"name": "Step 1", "action": "Do something"},
{"name": "Step 2", "action": "Do something else"}
]
)
self.assertEqual(workflow.name, "My Workflow")
self.assertEqual(len(workflow.steps), 2)
# Check file was created
self.assertTrue(Path(workflow.file_path).exists())
class TestHooks(unittest.TestCase):
"""Test hooks system."""
def test_hooks_config_loading(self):
"""Test hooks configuration loading."""
from praisonaiagents.memory import HooksManager
with tempfile.TemporaryDirectory() as tmpdir:
config_dir = Path(tmpdir) / ".praison"
config_dir.mkdir()
config = {
"enabled": True,
"timeout": 30,
"hooks": {
"pre_write_code": "echo 'pre-write'",
"post_write_code": "echo 'post-write'"
}
}
(config_dir / "hooks.json").write_text(json.dumps(config))
manager = HooksManager(workspace_path=tmpdir)
stats = manager.get_stats()
self.assertTrue(stats["enabled"])
self.assertEqual(stats["script_hooks"], 2)
def test_callable_hook_registration(self):
"""Test Python callable hook registration."""
from praisonaiagents.memory import HooksManager
with tempfile.TemporaryDirectory() as tmpdir:
manager = HooksManager(workspace_path=tmpdir)
called = []
def my_hook(context):
called.append(context)
return {"modified": True}
manager.register("pre_write_code", my_hook)
result = manager.execute("pre_write_code", {"file": "test.py"})
self.assertTrue(result.success)
self.assertEqual(len(called), 1)
self.assertEqual(called[0]["file"], "test.py")
def test_has_hooks(self):
"""Test has_hooks method."""
from praisonaiagents.memory import HooksManager
with tempfile.TemporaryDirectory() as tmpdir:
manager = HooksManager(workspace_path=tmpdir)
self.assertFalse(manager.has_hooks("pre_write_code"))
manager.register("pre_write_code", lambda ctx: None)
self.assertTrue(manager.has_hooks("pre_write_code"))
def test_create_config(self):
"""Test hooks config creation."""
from praisonaiagents.memory import HooksManager
with tempfile.TemporaryDirectory() as tmpdir:
manager = HooksManager(workspace_path=tmpdir)
manager.create_config(
hooks={"pre_write_code": "echo 'test'"},
timeout=60
)
config_path = Path(tmpdir) / ".praison" / "hooks.json"
self.assertTrue(config_path.exists())
config = json.loads(config_path.read_text())
self.assertEqual(config["timeout"], 60)
class TestSymlinkSupport(unittest.TestCase):
"""Test symlink support for shared rules."""
def test_symlink_rule_file(self):
"""Test that symlinked rule files are loaded."""
with tempfile.TemporaryDirectory() as tmpdir:
# Create actual rule file
shared_dir = Path(tmpdir) / "shared"
shared_dir.mkdir()
(shared_dir / "common.md").write_text("# Common Rules")
# Create symlink
project_dir = Path(tmpdir) / "project"
project_dir.mkdir()
try:
(project_dir / "CLAUDE.md").symlink_to(shared_dir / "common.md")
except OSError:
# Skip on systems that don't support symlinks
self.skipTest("Symlinks not supported")
manager = RulesManager(workspace_path=str(project_dir))
# After symlink resolution, the name comes from the resolved file
# Check that some rule with Common Rules content exists
found = False
for rule in manager._rules.values():
if "Common Rules" in rule.content:
found = True
break
self.assertTrue(found, "Symlinked rule content not found")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,433 @@
"""
Integration tests for PraisonAI Memory System.
Tests the complete memory system with real API calls:
- FileMemory with session save/resume
- Context compression with LLM
- Rules management with CLAUDE.md, AGENTS.md, GEMINI.md
- Multi-agent workflows with shared memory
Requires: OPENAI_API_KEY environment variable
"""
import os
import sys
import tempfile
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from praisonaiagents import Agent
from praisonaiagents.memory import FileMemory, RulesManager
def test_file_memory_with_agent():
"""Test FileMemory integration with Agent."""
print("\n" + "="*60)
print("TEST: FileMemory with Agent")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
# Create memory instance
memory = FileMemory(user_id="test_user", base_path=f"{tmpdir}/memory")
# Add some context
memory.add_long_term("User prefers Python for backend development", importance=0.9)
memory.add_long_term("User's name is John", importance=0.95)
memory.add_short_term("Currently working on an AI project")
# Create agent with memory
agent = Agent(
name="Assistant",
instructions="You are a helpful assistant. Use the memory context to personalize responses.",
memory=memory,
verbose=True
)
# Test that memory context is included
response = agent.chat("What do you know about me?")
print(f"\nAgent Response: {response}")
# Verify memory was used
assert memory.get_stats()["long_term_count"] >= 2
print("✓ FileMemory integration test passed")
def test_session_save_resume():
"""Test session save and resume functionality."""
print("\n" + "="*60)
print("TEST: Session Save/Resume")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
# Create memory and add context
memory = FileMemory(user_id="session_test", base_path=f"{tmpdir}/memory")
memory.add_short_term("Important context from previous session")
memory.add_short_term("User asked about machine learning")
memory.add_long_term("User is a data scientist", importance=0.9)
# Save session with conversation history
conversation = [
{"role": "user", "content": "Tell me about ML"},
{"role": "assistant", "content": "Machine learning is..."}
]
session_path = memory.save_session("ml_session", conversation_history=conversation)
print(f"Session saved to: {session_path}")
# Clear memory
memory.clear_short_term()
assert len(memory.get_short_term()) == 0
# Resume session
session_data = memory.resume_session("ml_session")
print(f"Session resumed with {len(session_data.get('conversation_history', []))} messages")
# Verify restoration
assert len(memory.get_short_term()) > 0
assert len(session_data.get("conversation_history", [])) == 2
print("✓ Session save/resume test passed")
def test_context_compression_with_llm():
"""Test context compression with real LLM."""
print("\n" + "="*60)
print("TEST: Context Compression with LLM")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
memory = FileMemory(
user_id="compress_test",
base_path=f"{tmpdir}/memory",
config={"short_term_limit": 20}
)
# Add many items to trigger compression
for i in range(15):
memory.add_short_term(f"User discussed topic {i}: some detailed context about item {i}")
print(f"Short-term items before compression: {len(memory.get_short_term())}")
# Create LLM function for compression using Agent
def llm_summarize(prompt):
from praisonaiagents import Agent
summarizer = Agent(
name="Summarizer",
instructions="You summarize text concisely.",
verbose=False
)
return summarizer.chat(prompt)
# Compress with LLM
summary = memory.compress(llm_func=llm_summarize, max_items=5)
print(f"Compression summary: {summary[:100]}...")
print(f"Short-term items after compression: {len(memory.get_short_term())}")
# Verify compression
assert len(memory.get_short_term()) <= 5
assert len(summary) > 0
print("✓ Context compression test passed")
def test_checkpointing():
"""Test checkpointing with file snapshots."""
print("\n" + "="*60)
print("TEST: Checkpointing")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
memory = FileMemory(user_id="checkpoint_test", base_path=f"{tmpdir}/memory")
# Create a test file
test_file = Path(tmpdir) / "code.py"
test_file.write_text("def hello():\n print('Hello')")
# Add memory and create checkpoint
memory.add_long_term("Original implementation uses hello function")
checkpoint_id = memory.create_checkpoint("before_refactor", include_files=[str(test_file)])
print(f"Created checkpoint: {checkpoint_id}")
# Modify file and memory
test_file.write_text("def greet(name):\n print(f'Hello {name}')")
memory.clear_all()
memory.add_long_term("Refactored to use greet function")
print(f"File after modification: {test_file.read_text()[:30]}...")
# Restore checkpoint
memory.restore_checkpoint(checkpoint_id, restore_files=True)
print(f"File after restore: {test_file.read_text()[:30]}...")
# Verify restoration
assert "hello" in test_file.read_text()
assert memory.get_stats()["long_term_count"] > 0
print("✓ Checkpointing test passed")
def test_rules_manager_root_files():
"""Test RulesManager with CLAUDE.md, AGENTS.md, GEMINI.md files."""
print("\n" + "="*60)
print("TEST: RulesManager with Root Instruction Files")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
# Create root instruction files
claude_content = """# Claude Instructions
- Always be helpful and concise
- Use Python for code examples
- Follow best practices
"""
agents_content = """# Codex Instructions
- Run tests before committing
- Document all functions
"""
gemini_content = """# Gemini Instructions
- Prefer functional programming
- Use type hints
"""
praison_content = """---
description: PraisonAI project rules
activation: always
priority: 100
---
# PraisonAI Rules
- Use async when possible
- Follow PEP 8
"""
(Path(tmpdir) / "CLAUDE.md").write_text(claude_content)
(Path(tmpdir) / "AGENTS.md").write_text(agents_content)
(Path(tmpdir) / "GEMINI.md").write_text(gemini_content)
(Path(tmpdir) / "PRAISON.md").write_text(praison_content)
# Create RulesManager
rules_manager = RulesManager(workspace_path=tmpdir, verbose=1)
# Check stats
stats = rules_manager.get_stats()
print(f"Rules stats: {stats}")
# Verify root files were loaded
assert stats["root_rules"] >= 4, f"Expected at least 4 root rules, got {stats['root_rules']}"
# Get rules by name
claude_rule = rules_manager.get_rule_by_name("claude")
agents_rule = rules_manager.get_rule_by_name("agents")
gemini_rule = rules_manager.get_rule_by_name("gemini")
praison_rule = rules_manager.get_rule_by_name("praison")
assert claude_rule is not None, "CLAUDE.md not loaded"
assert agents_rule is not None, "AGENTS.md not loaded"
assert gemini_rule is not None, "GEMINI.md not loaded"
assert praison_rule is not None, "PRAISON.md not loaded"
print(f"Claude rule: {claude_rule.description}")
print(f"Agents rule: {agents_rule.description}")
print(f"Gemini rule: {gemini_rule.description}")
print(f"Praison rule: {praison_rule.description}")
# Build context
context = rules_manager.build_rules_context()
print(f"\nRules context ({len(context)} chars):")
print(context[:500] + "..." if len(context) > 500 else context)
assert "helpful" in context.lower() or "python" in context.lower()
print("✓ RulesManager root files test passed")
def test_multi_agent_with_shared_memory():
"""Test multiple agents sharing memory."""
print("\n" + "="*60)
print("TEST: Multi-Agent with Shared Memory")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
# Create shared memory
shared_memory = FileMemory(user_id="shared", base_path=f"{tmpdir}/memory")
shared_memory.add_long_term("Project uses Python 3.12", importance=0.9)
shared_memory.add_long_term("Database is PostgreSQL", importance=0.85)
# Create multiple agents with shared memory
researcher = Agent(
name="Researcher",
role="Research Analyst",
goal="Research and gather information",
backstory="Expert at finding and analyzing information",
memory=shared_memory,
verbose=True
)
writer = Agent(
name="Writer",
role="Technical Writer",
goal="Write clear documentation",
backstory="Expert at technical writing",
memory=shared_memory,
verbose=True
)
# Test researcher
print("\n--- Researcher Agent ---")
research_response = researcher.chat("What technology stack are we using?")
print(f"Researcher: {research_response[:200]}...")
# Add finding to memory
shared_memory.add_short_term("Researcher confirmed: Python 3.12 + PostgreSQL stack")
# Test writer (should see researcher's finding)
print("\n--- Writer Agent ---")
writer_response = writer.chat("What should I document about our stack?")
print(f"Writer: {writer_response[:200]}...")
# Verify shared memory
stats = shared_memory.get_stats()
print(f"\nShared memory stats: {stats}")
assert stats["short_term_count"] >= 1
assert stats["long_term_count"] >= 2
print("✓ Multi-agent shared memory test passed")
def test_slash_commands():
"""Test memory slash commands."""
print("\n" + "="*60)
print("TEST: Memory Slash Commands")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
memory = FileMemory(user_id="commands_test", base_path=f"{tmpdir}/memory")
# Test /memory add
result = memory.handle_command("/memory add User prefers dark mode")
print(f"/memory add: {result}")
assert result["action"] == "add"
# Test /memory show
result = memory.handle_command("/memory show")
print(f"/memory show: {result['stats']}")
assert result["action"] == "show"
assert result["stats"]["long_term_count"] >= 1
# Test /memory search
result = memory.handle_command("/memory search dark mode")
print(f"/memory search: {len(result.get('results', []))} results")
assert result["action"] == "search"
# Test /memory save
result = memory.handle_command("/memory save test_session")
print(f"/memory save: {result}")
assert result["action"] == "save"
# Test /memory sessions
result = memory.handle_command("/memory sessions")
print(f"/memory sessions: {len(result.get('sessions', []))} sessions")
assert result["action"] == "sessions"
# Test /memory checkpoint
result = memory.handle_command("/memory checkpoint test_cp")
print(f"/memory checkpoint: {result}")
assert result["action"] == "checkpoint"
# Test /memory checkpoints
result = memory.handle_command("/memory checkpoints")
print(f"/memory checkpoints: {len(result.get('checkpoints', []))} checkpoints")
assert result["action"] == "checkpoints"
# Test /memory help
result = memory.handle_command("/memory help")
print(f"/memory help: {len(result.get('commands', {}))} commands")
assert result["action"] == "help"
assert len(result["commands"]) >= 10
print("✓ Slash commands test passed")
def test_agent_with_rules():
"""Test Agent with RulesManager integration."""
print("\n" + "="*60)
print("TEST: Agent with Rules Integration")
print("="*60)
with tempfile.TemporaryDirectory() as tmpdir:
# Create PRAISON.md in workspace
praison_rules = """# Project Rules
- Always respond in a friendly manner
- Use bullet points for lists
- Keep responses concise
"""
(Path(tmpdir) / "PRAISON.md").write_text(praison_rules)
# Change to temp directory to test rules discovery
original_cwd = os.getcwd()
os.chdir(tmpdir)
try:
# Create agent (should auto-discover rules)
agent = Agent(
name="RulesAgent",
instructions="You are a helpful assistant.",
verbose=True
)
# Check if rules were loaded
if agent._rules_manager:
stats = agent._rules_manager.get_stats()
print(f"Rules discovered: {stats}")
# Get rules context
context = agent.get_rules_context()
print(f"Rules context: {context[:200]}..." if context else "No rules context")
# Test agent response
response = agent.chat("List 3 benefits of Python")
print(f"\nAgent response: {response[:300]}...")
print("✓ Agent with rules test passed")
finally:
os.chdir(original_cwd)
def run_all_tests():
"""Run all integration tests."""
print("\n" + "="*60)
print("PRAISONAI MEMORY SYSTEM - INTEGRATION TESTS")
print("="*60)
# Check for API key
if not os.environ.get("OPENAI_API_KEY"):
print("WARNING: OPENAI_API_KEY not set. Some tests may fail.")
tests = [
("FileMemory with Agent", test_file_memory_with_agent),
("Session Save/Resume", test_session_save_resume),
("Context Compression", test_context_compression_with_llm),
("Checkpointing", test_checkpointing),
("RulesManager Root Files", test_rules_manager_root_files),
("Multi-Agent Shared Memory", test_multi_agent_with_shared_memory),
("Slash Commands", test_slash_commands),
("Agent with Rules", test_agent_with_rules),
]
passed = 0
failed = 0
for name, test_func in tests:
try:
test_func()
passed += 1
except Exception as e:
print(f"\n✗ {name} FAILED: {e}")
import traceback
traceback.print_exc()
failed += 1
print("\n" + "="*60)
print(f"RESULTS: {passed} passed, {failed} failed")
print("="*60)
return failed == 0
if __name__ == "__main__":
success = run_all_tests()
sys.exit(0 if success else 1)
+1 -1
View File
@@ -2422,7 +2422,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.178"
version = "0.0.180"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
+123 -11
View File
@@ -178,10 +178,11 @@ result = agent.rewrite("What about cost?", chat_history=[...])
### 5. Agent Memory (Zero Dependencies)
Enable persistent memory for agents - works out of the box without any extra packages.
Enable persistent memory for agents - works out of the box without any extra packages. Features inspired by Claude, Gemini CLI, Codex CLI, Cursor, and Windsurf.
```python
from praisonaiagents import Agent
from praisonaiagents.memory import FileMemory
# Enable memory with a single parameter
agent = Agent(
@@ -191,17 +192,9 @@ agent = Agent(
user_id="user123" # Isolate memory per user
)
# Store memories programmatically
agent.store_memory("User prefers dark mode", memory_type="short_term")
agent.store_memory("User's name is John", memory_type="long_term", importance=0.9)
# Memory is automatically injected into conversations
result = agent.start("What do you know about me?")
# Agent will recall: "Your name is John and you prefer dark mode"
# Memory persists across sessions - create new agent with same user_id
agent2 = Agent(name="Assistant", memory=True, user_id="user123")
# agent2 automatically loads all previous memories!
result = agent.start("My name is John and I prefer Python")
# Agent will remember this for future conversations
```
**Memory Types:**
@@ -210,6 +203,28 @@ agent2 = Agent(name="Assistant", memory=True, user_id="user123")
- **Entity**: People, places, organizations with attributes
- **Episodic**: Date-based interaction history
**Advanced Features (like Gemini CLI):**
```python
from praisonaiagents.memory import FileMemory
memory = FileMemory(user_id="user123")
# Session Save/Resume
memory.save_session("project_session", conversation_history=[...])
memory.resume_session("project_session")
# Context Compression
memory.compress(llm_func=lambda p: agent.chat(p), max_items=10)
# Checkpointing
memory.create_checkpoint("before_refactor", include_files=["main.py"])
memory.restore_checkpoint("before_refactor", restore_files=True)
# Slash Commands
memory.handle_command("/memory show")
memory.handle_command("/memory save my_session")
```
**Storage Options:**
| Option | Dependencies | Description |
|--------|-------------|-------------|
@@ -218,6 +233,103 @@ agent2 = Agent(name="Assistant", memory=True, user_id="user123")
| `memory="sqlite"` | Built-in | SQLite with indexing |
| `memory="chromadb"` | chromadb | Vector/semantic search |
### 6. Rules & Instructions (like Cursor/Windsurf)
PraisonAI auto-discovers instruction files from your project root and git root:
| File | Description | Priority |
|------|-------------|----------|
| `PRAISON.md` | PraisonAI native instructions | High |
| `PRAISON.local.md` | Local overrides (gitignored) | Higher |
| `CLAUDE.md` | Claude Code memory file | High |
| `CLAUDE.local.md` | Local overrides (gitignored) | Higher |
| `AGENTS.md` | OpenAI Codex CLI instructions | High |
| `GEMINI.md` | Gemini CLI memory file | High |
| `.cursorrules` | Cursor IDE rules | High |
| `.windsurfrules` | Windsurf IDE rules | High |
| `.claude/rules/*.md` | Claude Code modular rules | Medium |
| `.windsurf/rules/*.md` | Windsurf modular rules | Medium |
| `.cursor/rules/*.mdc` | Cursor modular rules | Medium |
| `.praison/rules/*.md` | Workspace rules | Medium |
| `~/.praison/rules/*.md` | Global rules | Low |
```python
from praisonaiagents import Agent
# Agent auto-discovers CLAUDE.md, AGENTS.md, GEMINI.md, etc.
agent = Agent(name="Assistant", instructions="You are helpful.")
# Rules are injected into system prompt automatically
```
**@Import Syntax (like Claude Code):**
```markdown
# CLAUDE.md
See @README for project overview
See @docs/architecture.md for system design
@~/.praison/my-preferences.md
```
**Rule File Format (with YAML frontmatter):**
```markdown
---
description: Python coding guidelines
globs: ["**/*.py"]
activation: always # always, glob, manual, ai_decision
---
# Guidelines
- Use type hints
- Follow PEP 8
```
### 7. Auto-Generated Memories (like Windsurf Cascade)
```python
from praisonaiagents.memory import FileMemory, AutoMemory
memory = FileMemory(user_id="user123")
auto = AutoMemory(memory, enabled=True)
# Automatically extracts and stores memories from conversations
memories = auto.process_interaction(
"My name is John and I prefer Python for backend work"
)
# Extracts: name="John", preference="Python for backend"
```
### 8. Workflows (like Windsurf)
Create reusable multi-step workflows in `.praison/workflows/`:
```python
from praisonaiagents.memory import WorkflowManager
manager = WorkflowManager()
# Execute a workflow
result = manager.execute(
"deploy",
executor=lambda prompt: agent.chat(prompt),
variables={"environment": "production"}
)
```
### 9. Hooks (like Windsurf Cascade Hooks)
Configure in `.praison/hooks.json`:
```python
from praisonaiagents.memory import HooksManager
hooks = HooksManager()
# Register Python hooks
hooks.register("pre_write_code", lambda ctx: print(f"Writing {ctx['file']}"))
# Execute hooks
result = hooks.execute("pre_write_code", {"file": "main.py"})
```
## Using No Code
### Auto Mode:
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.3.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.3.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.4.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v2.3.4.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+1 -1
View File
@@ -57,7 +57,7 @@ class CloudDeployer:
file.write("FROM python:3.11-slim\n")
file.write("WORKDIR /app\n")
file.write("COPY . .\n")
file.write("RUN pip install flask praisonai==2.3.3 gunicorn markdown\n")
file.write("RUN pip install flask praisonai==2.3.4 gunicorn markdown\n")
file.write("EXPOSE 8080\n")
file.write('CMD ["gunicorn", "-b", "0.0.0.0:8080", "api:app"]\n')
+1 -1
View File
@@ -1 +1 @@
__version__ = "2.3.3"
__version__ = "2.3.4"
+1 -1
View File
@@ -12,7 +12,7 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=0.0.178",
"praisonaiagents>=0.0.180",
"python-dotenv>=0.19.0",
"instructor>=1.3.3",
"PyYAML>=6.0",
+4 -4
View File
@@ -4197,7 +4197,7 @@ requires-dist = [
{ name = "praisonai-tools", marker = "extra == 'autogen'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'autogen-v4'", specifier = ">=0.0.22" },
{ name = "praisonai-tools", marker = "extra == 'crewai'", specifier = ">=0.0.22" },
{ name = "praisonaiagents", specifier = ">=0.0.178" },
{ name = "praisonaiagents", specifier = ">=0.0.180" },
{ name = "pyautogen", marker = "extra == 'autogen'", specifier = "==0.2.29" },
{ name = "pydantic", marker = "extra == 'chat'", specifier = "<=2.10.1" },
{ name = "pydantic", marker = "extra == 'code'", specifier = "<=2.10.1" },
@@ -4252,7 +4252,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "0.0.178"
version = "0.0.180"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "aiohttp" },
@@ -4261,9 +4261,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "rich" },
]
sdist = { url = "https://files.pythonhosted.org/packages/db/40/9c7355184b107ab09db83525b6ac8633b27c2cd12104902c7f95cdef6e61/praisonaiagents-0.0.178.tar.gz", hash = "sha256:9427be0e26c08f9a6c5e356e52991ee43a531a8088bd45ecb0a27f4e4b602eaa", size = 352905, upload-time = "2025-12-15T14:52:03.258Z" }
sdist = { url = "https://files.pythonhosted.org/packages/a6/7e/8404716bff5d51dccada132bf2c9fb3986f858622f1f7651ff379d56a8fa/praisonaiagents-0.0.180.tar.gz", hash = "sha256:e668ac705e15b13478cd2b895c32cf603fbf49c380b8cfdb4949adefc8505014", size = 387604, upload-time = "2025-12-15T16:37:52.512Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5b/4f/46b7030a57852127a0f185b0b2a9340df277d3396dff7ad86f2f8e580122/praisonaiagents-0.0.178-py3-none-any.whl", hash = "sha256:91a032a0da8574662269b7266c41667ae905434388f1e9bc74265f18df5596a7", size = 354385, upload-time = "2025-12-15T14:52:01.915Z" },
{ url = "https://files.pythonhosted.org/packages/a0/ec/8e954809280beb9eee32c130fd32ee7f6c6e73f9d806228cb31b8bbc0752/praisonaiagents-0.0.180-py3-none-any.whl", hash = "sha256:c0d8dcde4deb4dd3029f1f1fe4b263a4e08783c803ad8253be1d11470a807157", size = 378348, upload-time = "2025-12-15T16:37:51.033Z" },
]
[[package]]