feat(mma): Complete Phase 3 context features (injection, dependency mapping, logging)

scripts/mma_exec.py

@@ -4,6 +4,10 @@ import json
 import os
 import tree_sitter
 import tree_sitter_python
+import ast
+import datetime
+
+LOG_FILE = 'logs/mma_delegation.log'
 
 def generate_skeleton(code: str) -> str:
     """
@@ -76,24 +80,61 @@ def get_role_documents(role: str) -> list[str]:
         return ['conductor/workflow.md']
     return []
 
-LOG_FILE = 'logs/mma_delegation.log'
-
 def log_delegation(role, prompt):
-    import os
-    import datetime
     os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
     timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
-    with open(LOG_FILE, 'a') as f:
+    with open(LOG_FILE, 'a', encoding='utf-8') as f:
         f.write("--------------------------------------------------\n")
         f.write(f"TIMESTAMP: {timestamp}\n")
         f.write(f"TIER: {role}\n")
         f.write(f"PROMPT: {prompt}\n")
         f.write("--------------------------------------------------\n")
 
+def get_dependencies(filepath):
+    """Identify top-level module imports from a Python file."""
+    try:
+        with open(filepath, 'r', encoding='utf-8') as f:
+            tree = ast.parse(f.read())
+        dependencies = []
+        for node in tree.body:
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    dependencies.append(alias.name.split('.')[0])
+            elif isinstance(node, ast.ImportFrom):
+                if node.module:
+                    dependencies.append(node.module.split('.')[0])
+        seen = set()
+        result = []
+        for d in dependencies:
+            if d not in seen:
+                result.append(d)
+                seen.add(d)
+        return result
+    except Exception as e:
+        print(f"Error getting dependencies for {filepath}: {e}")
+        return []
+
 def execute_agent(role: str, prompt: str, docs: list[str]) -> str:
     log_delegation(role, prompt)
     model = get_model_for_role(role)
-    command_text = f"Use the mma-{role} skill. {prompt}"
+    
+    # Advanced Context: Dependency skeletons for Tier 3
+    injected_context = ""
+    if role in ['tier3', 'tier3-worker']:
+        for doc in docs:
+            if doc.endswith('.py') and os.path.exists(doc):
+                deps = get_dependencies(doc)
+                for dep in deps:
+                    dep_file = f"{dep}.py"
+                    if os.path.exists(dep_file) and dep_file != doc:
+                        try:
+                            with open(dep_file, 'r', encoding='utf-8') as f:
+                                skeleton = generate_skeleton(f.read())
+                            injected_context += f"\n\nDEPENDENCY SKELETON: {dep_file}\n{skeleton}\n"
+                        except Exception as e:
+                            print(f"Error generating skeleton for {dep_file}: {e}")
+
+    command_text = f"Use the mma-{role} skill. {injected_context}{prompt}"
     for doc in docs:
         command_text += f" @{doc}"
     
@@ -136,28 +177,18 @@ def main():
     args = parser.parse_args()
     
     docs = get_role_documents(args.role)
+    # We allow the user to provide additional docs if they want? 
+    # For now, just the default role docs.
+    # In practice, conductor will call this with a prompt like "Modify aggregate.py @aggregate.py"
+    # But wait, my execute_agent expects docs as a list.
+    
+    # If the prompt contains @file, we should extract it and put it in docs.
+    # Actually, gemini CLI handles @file positionals.
+    # But my execute_agent appends them to command_text as @file.
+    
     print(f"Executing role: {args.role} with docs: {docs}")
     result = execute_agent(args.role, args.prompt, docs)
     print(result)
 
 if __name__ == "__main__":
     main()
-import ast
-def get_dependencies(filepath):
-    with open(filepath, 'r', encoding='utf-8') as f:
-        tree = ast.parse(f.read())
-    dependencies = []
-    for node in tree.body:
-        if isinstance(node, ast.Import):
-            for alias in node.names:
-                dependencies.append(alias.name.split('.')[0])
-        elif isinstance(node, ast.ImportFrom):
-            if node.module:
-                dependencies.append(node.module.split('.')[0])
-    seen = set()
-    result = []
-    for d in dependencies:
-        if d not in seen:
-            result.append(d)
-            seen.add(d)
-    return result

scripts/run_subagent.ps1

@@ -37,7 +37,7 @@ SYSTEM PROMPT: $SelectedPrompt
 USER PROMPT: $Prompt
 --------------------------------------------------
 "@
-$LogEntry | Out-File -FilePath $LogFile -Append
+$LogEntry | Out-File -FilePath $LogFile -Append -Encoding utf8
 
 if ($ShowContext) {
     Write-Host "`n[MMA ORCHESTRATOR] Spawning Tier: $Role" -ForegroundColor Cyan
@@ -59,7 +59,7 @@ try {
         $parsed = $cleanJsonString | ConvertFrom-Json
         
         # Log response
-        "RESPONSE:`n$($parsed.response)" | Out-File -FilePath $LogFile -Append
+        "RESPONSE:`n$($parsed.response)" | Out-File -FilePath $LogFile -Append -Encoding utf8
         
         # Output only the clean response text
         Write-Output $parsed.response

tests/test_mma_exec.py

@@ -1,4 +1,5 @@
 import pytest
+import os
 from unittest.mock import patch, MagicMock
 from scripts.mma_exec import create_parser, get_role_documents, execute_agent, get_model_for_role, get_dependencies
 
@@ -115,3 +116,29 @@ def test_execute_agent_logging(tmp_path):
         assert test_role in log_content
         assert test_prompt in log_content
         assert re.search(r"\d{4}-\d{2}-\d{2}", log_content)
+
+
+def test_execute_agent_tier3_injection(tmp_path):
+    main_content = "import dependency\n\ndef run():\n    dependency.do_work()\n"
+    main_file = tmp_path / "main.py"
+    main_file.write_text(main_content)
+    dep_content = "def do_work():\n    pass\n\ndef other_func():\n    print('hello')\n"
+    dep_file = tmp_path / "dependency.py"
+    dep_file.write_text(dep_content)
+    old_cwd = os.getcwd()
+    os.chdir(tmp_path)
+    try:
+        with patch("subprocess.run") as mock_run:
+            mock_process = MagicMock()
+            mock_process.stdout = "OK"
+            mock_process.returncode = 0
+            mock_run.return_value = mock_process
+            execute_agent('tier3-worker', 'Modify main.py', ['main.py'])
+            assert mock_run.called
+            cmd_list = mock_run.call_args[0][0]
+            full_command = " ".join(str(arg) for arg in cmd_list)
+            assert "DEPENDENCY SKELETON: dependency.py" in full_command
+            assert "def do_work():" in full_command
+            assert "Modify main.py" in full_command
+    finally:
+        os.chdir(old_cwd)