From 221374eed617ea12def0dca0dd9957dc855254df Mon Sep 17 00:00:00 2001 From: Ed_ Date: Wed, 25 Feb 2026 20:21:12 -0500 Subject: [PATCH] feat(mma): Complete Phase 3 context features (injection, dependency mapping, logging) --- scripts/mma_exec.py | 81 +++++++++++++++++++++++++++------------- scripts/run_subagent.ps1 | 4 +- tests/test_mma_exec.py | 27 ++++++++++++++ 3 files changed, 85 insertions(+), 27 deletions(-) diff --git a/scripts/mma_exec.py b/scripts/mma_exec.py index ed93869..3fac2a0 100644 --- a/scripts/mma_exec.py +++ b/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 diff --git a/scripts/run_subagent.ps1 b/scripts/run_subagent.ps1 index fc292d5..303fc3f 100644 --- a/scripts/run_subagent.ps1 +++ b/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 diff --git a/tests/test_mma_exec.py b/tests/test_mma_exec.py index 93101f4..03522a0 100644 --- a/tests/test_mma_exec.py +++ b/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)