diff --git a/aworld-cli/src/aworld_cli/console.py b/aworld-cli/src/aworld_cli/console.py index 2c8373f6b..563728266 100644 --- a/aworld-cli/src/aworld_cli/console.py +++ b/aworld-cli/src/aworld_cli/console.py @@ -905,6 +905,7 @@ async def run_chat_session(self, agent_name: str, executor: Callable[[str], Any] f"Type '/agents' to list all available agents.\n" f"Type '/cost' for current session, '/cost -all' for global history.\n" f"Type '/compact' to run context compression.\n" + f"Type '/memory' to edit project context, '/memory view' to view, '/memory status' for status.\n" f"Use @filename to include images or text files (e.g., @photo.jpg or @document.txt)." ) @@ -920,6 +921,7 @@ async def run_chat_session(self, agent_name: str, executor: Callable[[str], Any] slash_cmds = [ "/agents", "/skills", "/new", "/restore", "/latest", "/exit", "/quit", "/switch", "/cost", "/cost -all", "/compact", + "/memory", "/memory view", "/memory reload", "/memory status", ] switch_with_agents = [f"/switch {n}" for n in agent_names] if agent_names else [] all_words = slash_cmds + switch_with_agents + ["exit", "quit"] @@ -935,6 +937,10 @@ async def run_chat_session(self, agent_name: str, executor: Callable[[str], Any] "/cost": "View query history (current session)", "/cost -all": "View global history (all sessions)", "/compact": "Run context compression", + "/memory": "Edit AWORLD.md project context", + "/memory view": "View current memory content", + "/memory reload": "Reload memory from file", + "/memory status": "Show memory system status", "exit": "Exit chat", "quit": "Exit chat", } @@ -1166,6 +1172,142 @@ async def run_chat_session(self, agent_name: str, executor: Callable[[str], Any] traceback.print_exc() continue + # Handle memory command + memory_input = user_input.strip().lower() + if memory_input.startswith(("/memory", "memory")): + try: + parts = user_input.split(maxsplit=1) + subcommand = parts[1] if len(parts) > 1 else "" + + # Import required modules + import os + from pathlib import Path + import subprocess + + # Find AWORLD.md file + def find_aworld_file(): + """Find AWORLD.md in standard locations""" + working_dir = Path.cwd() + search_paths = [ + Path.home() / '.aworld' / 'AWORLD.md', + working_dir / '.aworld' / 'AWORLD.md', + working_dir / 'AWORLD.md', + ] + for path in search_paths: + if path.exists(): + return path + return None + + def get_editor(): + """Get editor from environment variables""" + return os.environ.get('VISUAL') or os.environ.get('EDITOR') or 'nano' + + if subcommand == "view": + # View current memory content + aworld_file = find_aworld_file() + if not aworld_file: + self.console.print("[yellow]No AWORLD.md file found.[/yellow]") + self.console.print("[dim]Create one with: /memory[/dim]") + else: + self.console.print(f"[dim]Reading from: {aworld_file}[/dim]\n") + content = aworld_file.read_text(encoding='utf-8') + # Display in a panel + from rich.panel import Panel + from rich.syntax import Syntax + syntax = Syntax(content, "markdown", theme="monokai", line_numbers=False) + self.console.print(Panel(syntax, title="AWORLD.md", border_style="cyan")) + + elif subcommand == "reload": + # Reload memory from file + self.console.print("[dim]Memory reload functionality requires agent restart.[/dim]") + self.console.print("[dim]The AWORLD.md file will be automatically loaded on next agent start.[/dim]") + + elif subcommand == "status": + # Show memory system status + aworld_file = find_aworld_file() + from rich.table import Table + table = Table(title="Memory System Status", box=box.ROUNDED) + table.add_column("Property", style="cyan") + table.add_column("Value", style="green") + + if aworld_file: + table.add_row("AWORLD.md Location", str(aworld_file)) + table.add_row("File Size", f"{aworld_file.stat().st_size} bytes") + from datetime import datetime + mtime = datetime.fromtimestamp(aworld_file.stat().st_mtime) + table.add_row("Last Modified", mtime.strftime("%Y-%m-%d %H:%M:%S")) + table.add_row("Status", "✅ Active") + else: + table.add_row("AWORLD.md Location", "Not found") + table.add_row("Status", "❌ Not configured") + + table.add_row("Feature", "AWORLDFileNeuron") + table.add_row("Auto-load", "Enabled") + self.console.print(table) + + else: + # Edit AWORLD.md (default action) + aworld_file = find_aworld_file() + + if not aworld_file: + # Create new file in user directory (DEFAULT) + default_location = Path.home() / '.aworld' / 'AWORLD.md' + self.console.print(f"[yellow]No AWORLD.md found. Creating new file at:[/yellow]") + self.console.print(f"[cyan]{default_location}[/cyan]") + self.console.print(f"[dim](Default: ~/.aworld/AWORLD.md)[/dim]\n") + + # Create directory if needed + default_location.parent.mkdir(parents=True, exist_ok=True) + + # Create template + template = """# Project Context + +## Project Overview +Describe your project here. + +## Important Guidelines +- Add your project-specific guidelines +- Include coding standards +- Document important conventions + +## Technical Details +- List key technologies +- Document architecture decisions +- Note important file locations + +## Custom Instructions +Add any custom instructions for AI agents working on this project. + +--- +*This file is automatically loaded by AWorld agents.* +""" + default_location.write_text(template, encoding='utf-8') + aworld_file = default_location + + # Open in editor + editor = get_editor() + self.console.print(f"[dim]Opening {aworld_file} in {editor}...[/dim]") + + try: + # Open editor and wait for it to close + result = subprocess.run([editor, str(aworld_file)]) + if result.returncode == 0: + self.console.print("[green]✅ AWORLD.md saved successfully.[/green]") + self.console.print("[dim]Changes will take effect on next agent start.[/dim]") + else: + self.console.print(f"[yellow]Editor exited with code {result.returncode}[/yellow]") + except FileNotFoundError: + self.console.print(f"[red]Editor '{editor}' not found.[/red]") + self.console.print("[dim]Set EDITOR or VISUAL environment variable to your preferred editor.[/dim]") + except Exception as e: + self.console.print(f"[red]Error opening editor: {e}[/red]") + + except Exception as e: + self.console.print(f"[red]Error handling memory command: {e}[/red]") + import traceback + traceback.print_exc() + continue + # Handle agents command if user_input.lower() in ("/agents", "agents"): try: diff --git a/aworld/core/context/amni/config.py b/aworld/core/context/amni/config.py index 05d835d54..1108a5307 100644 --- a/aworld/core/context/amni/config.py +++ b/aworld/core/context/amni/config.py @@ -86,6 +86,10 @@ class AgentContextConfig(BaseConfig): enable_system_prompt_augment: bool = Field(default=False, description="enable_system_prompt_augment") neuron_names: Optional[list[str]] = Field(default_factory=list) neuron_config: Optional[Dict[str, NeuronStrategyConfig]] = Field(default_factory=list) + + # AWORLD.md File Support + enable_aworld_file: bool = Field(default=True, description="Enable AWORLD.md file loading for project-specific context") + aworld_file_path: Optional[str] = Field(default=None, description="Custom path to AWORLD.md file (optional override)") # Context Reduce - Purge history_rounds: int = Field(default=100, diff --git a/aworld/core/context/amni/processor/op/system_prompt_augment_op.py b/aworld/core/context/amni/processor/op/system_prompt_augment_op.py index b6ce689fb..d041f1c48 100644 --- a/aworld/core/context/amni/processor/op/system_prompt_augment_op.py +++ b/aworld/core/context/amni/processor/op/system_prompt_augment_op.py @@ -99,6 +99,12 @@ async def _process_neurons(self, context: ApplicationContext, event: SystemPromp from aworld.core.context.amni.prompt.neurons.skill_neuron import SKILL_NEURON_NAME if SKILL_NEURON_NAME not in neuron_names: neuron_names.append(SKILL_NEURON_NAME) + + # Enable AWORLD.md File Feature + if agent_context_config.enable_aworld_file: + from aworld.core.context.amni.prompt.neurons.aworld_file_neuron import AWORLD_FILE_NEURON_NAME + if AWORLD_FILE_NEURON_NAME not in neuron_names: + neuron_names.insert(0, AWORLD_FILE_NEURON_NAME) # High priority - insert at beginning # Enable Planing Feature if agent_context_config.automated_cognitive_ingestion: diff --git a/aworld/core/context/amni/prompt/neurons/__init__.py b/aworld/core/context/amni/prompt/neurons/__init__.py index 626d9bcc7..83d5c123f 100644 --- a/aworld/core/context/amni/prompt/neurons/__init__.py +++ b/aworld/core/context/amni/prompt/neurons/__init__.py @@ -18,6 +18,7 @@ class Neurons: WORKSPACE = "workspace" GRAPH = "graph" ENTITY = "entity" + AWORLD_FILE = "aworld_file" class Neuron(ABC): """ diff --git a/aworld/core/context/amni/prompt/neurons/aworld_file_neuron.py b/aworld/core/context/amni/prompt/neurons/aworld_file_neuron.py new file mode 100644 index 000000000..fd7f140e7 --- /dev/null +++ b/aworld/core/context/amni/prompt/neurons/aworld_file_neuron.py @@ -0,0 +1,239 @@ +import os +import re +from pathlib import Path +from typing import List, Optional +from ... import ApplicationContext +from . import Neuron +from .neuron_factory import neuron_factory +from aworld.logs.util import logger + +AWORLD_FILE_NEURON_NAME = "aworld_file" + + +@neuron_factory.register( + name=AWORLD_FILE_NEURON_NAME, + desc="Neuron for loading AWORLD.md file content", + prio=50 # Higher priority than basic (100), lower than task (0) +) +class AWORLDFileNeuron(Neuron): + """ + Neuron that loads and processes AWORLD.md files + + Features: + - Searches for AWORLD.md in multiple locations + - Supports @import syntax for including other files + - Caches content to avoid repeated file I/O + - Handles circular imports gracefully + """ + + IMPORT_PATTERN = re.compile(r'^@(.+\.md)\s*$', re.MULTILINE) + + def __init__(self): + super().__init__() + self._content_cache: Optional[str] = None + self._last_modified: Optional[float] = None + self._file_path: Optional[Path] = None + + def _find_aworld_file(self, context: ApplicationContext) -> Optional[Path]: + """ + Find AWORLD.md file in standard locations + + Search order (priority): + 1. ~/.aworld/AWORLD.md (user-level, global) - DEFAULT and HIGHEST PRIORITY + 2. .aworld/AWORLD.md (project-specific, if exists) + 3. AWORLD.md (project root, if exists) + + Note: User-level config (~/.aworld/AWORLD.md) is the DEFAULT location. + Project-level configs are OPTIONAL overrides. + """ + # Get working directory from context + working_dir = getattr(context, 'working_directory', os.getcwd()) + + search_paths = [ + Path.home() / '.aworld' / 'AWORLD.md', # User-level (DEFAULT) + Path(working_dir) / '.aworld' / 'AWORLD.md', # Project-specific (optional) + Path(working_dir) / 'AWORLD.md', # Project root (optional) + ] + + for path in search_paths: + if path.exists() and path.is_file(): + logger.info(f"Found AWORLD.md at: {path}") + return path + + logger.debug("No AWORLD.md file found") + return None + + def _resolve_import_path(self, import_path: str, base_path: Path) -> Optional[Path]: + """ + Resolve import path relative to base file + + Args: + import_path: Path from @import statement + base_path: Path of file containing the import + + Returns: + Resolved absolute path or None if not found + """ + import_path = import_path.strip() + + # Handle absolute paths + if import_path.startswith('/'): + resolved = Path(import_path) + else: + # Relative to base file's directory + resolved = (base_path.parent / import_path).resolve() + + if resolved.exists() and resolved.is_file(): + return resolved + + logger.warning(f"Import file not found: {import_path} (resolved to {resolved})") + return None + + def _load_file_with_imports( + self, + file_path: Path, + visited: Optional[set] = None + ) -> str: + """ + Load file content and recursively process @imports + + Args: + file_path: Path to file to load + visited: Set of already visited files (for circular import detection) + + Returns: + Processed content with all imports resolved + """ + if visited is None: + visited = set() + + # Circular import detection + file_path_str = str(file_path.resolve()) + if file_path_str in visited: + logger.warning(f"Circular import detected: {file_path}") + return f"\n\n" + + visited.add(file_path_str) + + try: + content = file_path.read_text(encoding='utf-8') + except Exception as e: + logger.error(f"Error reading file {file_path}: {e}") + return f"\n\n" + + # Process imports + def replace_import(match): + import_path = match.group(1) + resolved_path = self._resolve_import_path(import_path, file_path) + + if resolved_path: + imported_content = self._load_file_with_imports(resolved_path, visited.copy()) + return f"\n\n{imported_content}\n\n" + else: + return f"\n\n" + + processed_content = self.IMPORT_PATTERN.sub(replace_import, content) + return processed_content + + def _should_reload(self, file_path: Path) -> bool: + """ + Check if file should be reloaded based on modification time + """ + if self._content_cache is None: + return True + + if self._file_path != file_path: + return True + + try: + current_mtime = file_path.stat().st_mtime + if self._last_modified is None or current_mtime > self._last_modified: + return True + except Exception as e: + logger.warning(f"Error checking file modification time: {e}") + return True + + return False + + async def format_items( + self, + context: ApplicationContext, + namespace: str = None, + **kwargs + ) -> List[str]: + """ + Load and format AWORLD.md content + + Returns: + List containing the processed content + """ + items = [] + + try: + # Find AWORLD.md file + file_path = self._find_aworld_file(context) + + if not file_path: + return items + + # Check if reload is needed + if self._should_reload(file_path): + logger.info(f"Loading AWORLD.md from: {file_path}") + + # Load content with imports + content = self._load_file_with_imports(file_path) + + # Update cache + self._content_cache = content + self._file_path = file_path + self._last_modified = file_path.stat().st_mtime + + # Return cached content + if self._content_cache: + items.append(self._content_cache) + + except Exception as e: + logger.error(f"Error processing AWORLD.md: {e}") + items.append(f"") + + return items + + async def format( + self, + context: ApplicationContext, + items: List[str] = None, + namespace: str = None, + **kwargs + ) -> str: + """ + Format AWORLD.md content for injection into system prompt + """ + if not items: + items = await self.format_items(context, namespace, **kwargs) + + if not items: + return "" + + # Combine all items with proper formatting + content = "\n\n".join(items) + + # Wrap in a clear section + formatted = f""" +## Project Context (from AWORLD.md) + +{content} + +--- +""" + return formatted + + async def desc( + self, + context: ApplicationContext, + namespace: str = None, + **kwargs + ) -> str: + """ + Return description of this neuron + """ + return "Project-specific context loaded from AWORLD.md file" diff --git a/examples/cast/test_aworld_md_integration.py b/examples/cast/test_aworld_md_integration.py new file mode 100644 index 000000000..c95ca58f8 --- /dev/null +++ b/examples/cast/test_aworld_md_integration.py @@ -0,0 +1,340 @@ +""" +Integration test for AWORLD.md functionality +Tests the complete flow of AWORLD.md loading and integration with the memory system +""" +import asyncio +import os +import sys +import tempfile +from pathlib import Path + +# Add parent directory to path +sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + +from aworld.core.context.amni import ApplicationContext, TaskInput, AmniConfigFactory +from aworld.core.context.amni.config import AmniConfigLevel +from aworld.core.context.amni.prompt.neurons.aworld_file_neuron import AWORLDFileNeuron + + +async def test_basic_loading(): + """Test 1: Basic AWORLD.md loading""" + print("\n" + "="*60) + print("TEST 1: Basic AWORLD.md Loading") + print("="*60) + + with tempfile.TemporaryDirectory() as tmpdir: + # Create AWORLD.md + aworld_file = Path(tmpdir) / '.aworld' / 'AWORLD.md' + aworld_file.parent.mkdir(parents=True, exist_ok=True) + aworld_file.write_text(""" +# Test Project Context + +## Project Overview +This is a test project for AWORLD.md functionality. + +## Guidelines +- Use Python 3.10+ +- Follow PEP 8 +- Write comprehensive tests +""") + + # Create context + task_input = TaskInput( + user_id="test_user", + session_id="test_session", + task_id="test_task", + task_content="test", + origin_user_input="test" + ) + + context_config = AmniConfigFactory.create(AmniConfigLevel.PILOT) + context = await ApplicationContext.from_input(task_input, context_config=context_config) + context.working_directory = tmpdir + + # Create neuron and load content + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + # Verify + assert len(items) == 1, f"Expected 1 item, got {len(items)}" + assert "Test Project Context" in items[0], "Content not loaded correctly" + assert "Python 3.10+" in items[0], "Guidelines not loaded" + + print("✅ Basic loading works") + print(f" Loaded {len(items[0])} characters") + print(f" Content preview: {items[0][:100]}...") + + +async def test_import_functionality(): + """Test 2: @import syntax""" + print("\n" + "="*60) + print("TEST 2: @import Functionality") + print("="*60) + + with tempfile.TemporaryDirectory() as tmpdir: + # Create main file + aworld_file = Path(tmpdir) / '.aworld' / 'AWORLD.md' + aworld_file.parent.mkdir(parents=True, exist_ok=True) + aworld_file.write_text(""" +# Main Context + +@guidelines.md +@architecture.md +""") + + # Create imported files + (Path(tmpdir) / '.aworld' / 'guidelines.md').write_text(""" +## Coding Guidelines +- Use type hints +- Write tests +""") + + (Path(tmpdir) / '.aworld' / 'architecture.md').write_text(""" +## Architecture +- Follow MVC pattern +- Use dependency injection +""") + + # Create context + task_input = TaskInput( + user_id="test_user", + session_id="test_session", + task_id="test_task", + task_content="test", + origin_user_input="test" + ) + + context_config = AmniConfigFactory.create(AmniConfigLevel.PILOT) + context = await ApplicationContext.from_input(task_input, context_config=context_config) + context.working_directory = tmpdir + + # Load content + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + # Verify + content = items[0] + assert "Main Context" in content, "Main content not found" + assert "Coding Guidelines" in content, "Imported guidelines not found" + assert "Use type hints" in content, "Guidelines details not found" + assert "Architecture" in content, "Imported architecture not found" + assert "MVC pattern" in content, "Architecture details not found" + + print("✅ Import functionality works") + print(" ✓ Main content loaded") + print(" ✓ guidelines.md imported") + print(" ✓ architecture.md imported") + + +async def test_circular_import_detection(): + """Test 3: Circular import detection""" + print("\n" + "="*60) + print("TEST 3: Circular Import Detection") + print("="*60) + + with tempfile.TemporaryDirectory() as tmpdir: + # Create circular imports + file_a = Path(tmpdir) / '.aworld' / 'a.md' + file_b = Path(tmpdir) / '.aworld' / 'b.md' + file_a.parent.mkdir(parents=True, exist_ok=True) + + file_a.write_text("# File A\n@b.md") + file_b.write_text("# File B\n@a.md") + + aworld_file = Path(tmpdir) / '.aworld' / 'AWORLD.md' + aworld_file.write_text("@a.md") + + # Create context + task_input = TaskInput( + user_id="test_user", + session_id="test_session", + task_id="test_task", + task_content="test", + origin_user_input="test" + ) + + context_config = AmniConfigFactory.create(AmniConfigLevel.PILOT) + context = await ApplicationContext.from_input(task_input, context_config=context_config) + context.working_directory = tmpdir + + # Load content + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + # Verify + assert len(items) == 1, "Should handle circular imports gracefully" + assert "Circular import" in items[0], "Circular import not detected" + + print("✅ Circular import detection works") + print(" ✓ Detected circular reference") + print(" ✓ Handled gracefully") + + +async def test_caching(): + """Test 4: Content caching""" + print("\n" + "="*60) + print("TEST 4: Content Caching") + print("="*60) + + with tempfile.TemporaryDirectory() as tmpdir: + aworld_file = Path(tmpdir) / '.aworld' / 'AWORLD.md' + aworld_file.parent.mkdir(parents=True, exist_ok=True) + aworld_file.write_text("# Version 1") + + # Create context + task_input = TaskInput( + user_id="test_user", + session_id="test_session", + task_id="test_task", + task_content="test", + origin_user_input="test" + ) + + context_config = AmniConfigFactory.create(AmniConfigLevel.PILOT) + context = await ApplicationContext.from_input(task_input, context_config=context_config) + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + + # First load + items1 = await neuron.format_items(context) + assert "Version 1" in items1[0], "First load failed" + + # Second load (should use cache) + items2 = await neuron.format_items(context) + assert items1 == items2, "Cache not working" + + # Modify file + import time + time.sleep(0.1) # Ensure mtime changes + aworld_file.write_text("# Version 2") + + # Third load (should reload) + items3 = await neuron.format_items(context) + assert "Version 2" in items3[0], "Reload failed" + assert items3 != items1, "Should have reloaded" + + print("✅ Caching works correctly") + print(" ✓ First load successful") + print(" ✓ Cache used on second load") + print(" ✓ Reloaded after file modification") + + +async def test_formatted_output(): + """Test 5: Formatted output""" + print("\n" + "="*60) + print("TEST 5: Formatted Output") + print("="*60) + + with tempfile.TemporaryDirectory() as tmpdir: + aworld_file = Path(tmpdir) / '.aworld' / 'AWORLD.md' + aworld_file.parent.mkdir(parents=True, exist_ok=True) + aworld_file.write_text("# Test Content") + + # Create context + task_input = TaskInput( + user_id="test_user", + session_id="test_session", + task_id="test_task", + task_content="test", + origin_user_input="test" + ) + + context_config = AmniConfigFactory.create(AmniConfigLevel.PILOT) + context = await ApplicationContext.from_input(task_input, context_config=context_config) + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + formatted = await neuron.format(context) + + # Verify + assert "Project Context (from AWORLD.md)" in formatted, "Header not found" + assert "# Test Content" in formatted, "Content not found" + assert "---" in formatted, "Footer not found" + + print("✅ Formatted output works") + print(" ✓ Header present") + print(" ✓ Content formatted") + print(" ✓ Footer present") + + +async def test_no_file(): + """Test 6: Behavior when no file exists""" + print("\n" + "="*60) + print("TEST 6: No File Handling") + print("="*60) + + with tempfile.TemporaryDirectory() as tmpdir: + # Create context without AWORLD.md + task_input = TaskInput( + user_id="test_user", + session_id="test_session", + task_id="test_task", + task_content="test", + origin_user_input="test" + ) + + context_config = AmniConfigFactory.create(AmniConfigLevel.PILOT) + context = await ApplicationContext.from_input(task_input, context_config=context_config) + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + # Verify + assert len(items) == 0, "Should return empty list when no file" + + formatted = await neuron.format(context) + assert formatted == "", "Should return empty string when no file" + + print("✅ No file handling works") + print(" ✓ Returns empty list") + print(" ✓ Returns empty string for format") + + +async def main(): + """Run all tests""" + print("\n" + "="*60) + print("AWORLD.md Integration Test Suite") + print("="*60) + + tests = [ + test_basic_loading, + test_import_functionality, + test_circular_import_detection, + test_caching, + test_formatted_output, + test_no_file, + ] + + passed = 0 + failed = 0 + + for test in tests: + try: + await test() + passed += 1 + except Exception as e: + failed += 1 + print(f"\n❌ Test failed: {e}") + import traceback + traceback.print_exc() + + print("\n" + "="*60) + print("Test Summary") + print("="*60) + print(f"Total: {len(tests)}") + print(f"✅ Passed: {passed}") + print(f"❌ Failed: {failed}") + + if failed == 0: + print("\n🎉 All tests passed!") + return 0 + else: + print(f"\n⚠️ {failed} test(s) failed") + return 1 + + +if __name__ == "__main__": + exit_code = asyncio.run(main()) + sys.exit(exit_code) diff --git a/tests/core/context/amni/prompt/neurons/test_aworld_file_neuron.py b/tests/core/context/amni/prompt/neurons/test_aworld_file_neuron.py new file mode 100644 index 000000000..1fbc9d9cf --- /dev/null +++ b/tests/core/context/amni/prompt/neurons/test_aworld_file_neuron.py @@ -0,0 +1,242 @@ +import pytest +import tempfile +import time +from pathlib import Path +from aworld.core.context.amni.prompt.neurons.aworld_file_neuron import AWORLDFileNeuron +from aworld.core.context import ApplicationContext + + +@pytest.fixture +def temp_aworld_file(): + """Create temporary AWORLD.md file""" + with tempfile.TemporaryDirectory() as tmpdir: + aworld_file = Path(tmpdir) / 'AWORLD.md' + aworld_file.write_text(""" +# Test Project Context + +This is a test project. + +## Guidelines +- Use Python 3.10+ +- Follow PEP 8 +""") + yield tmpdir, aworld_file + + +@pytest.fixture +def temp_aworld_file_with_imports(): + """Create AWORLD.md with imports""" + with tempfile.TemporaryDirectory() as tmpdir: + # Create main file + aworld_file = Path(tmpdir) / 'AWORLD.md' + aworld_file.write_text(""" +# Main Context + +@guidelines.md +@architecture.md +""") + + # Create imported files + (Path(tmpdir) / 'guidelines.md').write_text(""" +## Coding Guidelines +- Use type hints +- Write tests +""") + + (Path(tmpdir) / 'architecture.md').write_text(""" +## Architecture +- Follow MVC pattern +- Use dependency injection +""") + + yield tmpdir, aworld_file + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_basic(temp_aworld_file): + """Test basic AWORLD.md loading""" + tmpdir, aworld_file = temp_aworld_file + + # Create context with working directory + context = ApplicationContext() + context.working_directory = tmpdir + + # Create neuron + neuron = AWORLDFileNeuron() + + # Load content + items = await neuron.format_items(context) + + assert len(items) == 1 + assert "Test Project Context" in items[0] + assert "Python 3.10+" in items[0] + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_with_imports(temp_aworld_file_with_imports): + """Test AWORLD.md with @imports""" + tmpdir, aworld_file = temp_aworld_file_with_imports + + context = ApplicationContext() + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + assert len(items) == 1 + content = items[0] + + # Check main content + assert "Main Context" in content + + # Check imported content + assert "Coding Guidelines" in content + assert "Use type hints" in content + assert "Architecture" in content + assert "MVC pattern" in content + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_circular_import(): + """Test circular import detection""" + with tempfile.TemporaryDirectory() as tmpdir: + # Create circular imports + file_a = Path(tmpdir) / 'a.md' + file_b = Path(tmpdir) / 'b.md' + + file_a.write_text("# File A\n@b.md") + file_b.write_text("# File B\n@a.md") + + aworld_file = Path(tmpdir) / 'AWORLD.md' + aworld_file.write_text("@a.md") + + context = ApplicationContext() + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + # Should handle gracefully + assert len(items) == 1 + assert "Circular import" in items[0] + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_caching(): + """Test content caching""" + with tempfile.TemporaryDirectory() as tmpdir: + aworld_file = Path(tmpdir) / 'AWORLD.md' + aworld_file.write_text("# Version 1") + + context = ApplicationContext() + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + + # First load + items1 = await neuron.format_items(context) + assert "Version 1" in items1[0] + + # Second load (should use cache) + items2 = await neuron.format_items(context) + assert items1 == items2 + + # Modify file + time.sleep(0.1) # Ensure mtime changes + aworld_file.write_text("# Version 2") + + # Third load (should reload) + items3 = await neuron.format_items(context) + assert "Version 2" in items3[0] + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_format(): + """Test formatted output""" + with tempfile.TemporaryDirectory() as tmpdir: + aworld_file = Path(tmpdir) / 'AWORLD.md' + aworld_file.write_text("# Test") + + context = ApplicationContext() + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + formatted = await neuron.format(context) + + assert "Project Context (from AWORLD.md)" in formatted + assert "# Test" in formatted + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_no_file(): + """Test behavior when no AWORLD.md file exists""" + with tempfile.TemporaryDirectory() as tmpdir: + context = ApplicationContext() + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + # Should return empty list + assert len(items) == 0 + + # Format should return empty string + formatted = await neuron.format(context) + assert formatted == "" + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_desc(): + """Test neuron description""" + context = ApplicationContext() + neuron = AWORLDFileNeuron() + + desc = await neuron.desc(context) + assert desc == "Project-specific context loaded from AWORLD.md file" + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_import_not_found(): + """Test behavior when imported file doesn't exist""" + with tempfile.TemporaryDirectory() as tmpdir: + aworld_file = Path(tmpdir) / 'AWORLD.md' + aworld_file.write_text("# Main\n@nonexistent.md") + + context = ApplicationContext() + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + assert len(items) == 1 + assert "Import not found" in items[0] + + +@pytest.mark.asyncio +async def test_aworld_file_neuron_nested_imports(): + """Test nested imports (import chain)""" + with tempfile.TemporaryDirectory() as tmpdir: + # Create import chain: AWORLD.md -> level1.md -> level2.md + aworld_file = Path(tmpdir) / 'AWORLD.md' + aworld_file.write_text("# Root\n@level1.md") + + level1 = Path(tmpdir) / 'level1.md' + level1.write_text("# Level 1\n@level2.md") + + level2 = Path(tmpdir) / 'level2.md' + level2.write_text("# Level 2\nDeep content") + + context = ApplicationContext() + context.working_directory = tmpdir + + neuron = AWORLDFileNeuron() + items = await neuron.format_items(context) + + assert len(items) == 1 + content = items[0] + + # All levels should be present + assert "Root" in content + assert "Level 1" in content + assert "Level 2" in content + assert "Deep content" in content