diff --git a/.editorconfig b/.editorconfig index 56f90a9..572197b 100644 --- a/.editorconfig +++ b/.editorconfig @@ -2,7 +2,7 @@ root = true [*.py] indent_style = space -indent_size = 2 +indent_size = 1 [*.s] indent_style = tab diff --git a/coverage_report.txt b/coverage_report.txt deleted file mode 100644 index e85738c..0000000 Binary files a/coverage_report.txt and /dev/null differ diff --git a/scripts/mma_exec.py b/scripts/mma_exec.py index 62a2101..9c8cc1a 100644 --- a/scripts/mma_exec.py +++ b/scripts/mma_exec.py @@ -81,7 +81,7 @@ def get_role_documents(role: str) -> list[str]: return ['conductor/workflow.md'] return [] -def log_delegation(role, prompt, result=None): +def log_delegation(role, full_prompt, result=None, summary_prompt=None): os.makedirs('logs/agents', exist_ok=True) timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') log_file = f'logs/agents/mma_{role}_task_{timestamp}.log' @@ -91,7 +91,7 @@ def log_delegation(role, prompt, result=None): f.write(f"ROLE: {role}\n") f.write(f"TIMESTAMP: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write("--------------------------------------------------\n") - f.write(f"PROMPT:\n{prompt}\n") + f.write(f"FULL PROMPT:\n{full_prompt}\n") f.write("--------------------------------------------------\n") if result: f.write(f"RESULT:\n{result}\n") @@ -99,15 +99,13 @@ def log_delegation(role, prompt, result=None): # Also keep the master log os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True) + display_prompt = summary_prompt if summary_prompt else full_prompt with open(LOG_FILE, 'a', encoding='utf-8') as f: - f.write(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {role}: {prompt[:100]}... (Log: {log_file})\n") + f.write(f"[{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {role}: {display_prompt[:100]}... (Log: {log_file})\n") return log_file -def execute_agent(role: str, prompt: str, docs: list[str]) -> str: - model = get_model_for_role(role) - - # Advanced Context: Dependency skeletons for Tier 3 +def get_dependencies(filepath: str) -> list[str]: """Identify top-level module imports from a Python file.""" try: with open(filepath, 'r', encoding='utf-8') as f: @@ -200,12 +198,13 @@ def execute_agent(role: str, prompt: str, docs: list[str]) -> str: command_text += f"\n\nTASK: {prompt}\n\n" # Use subprocess with input to pipe the prompt via stdin, avoiding WinError 206. - # We use -p 'task' to ensure non-interactive (headless) mode and valid parsing. + # We use -p '-' to read from stdin if supported, otherwise just a placeholder that stdin appends to. ps_command = ( f"if (Test-Path 'C:\\projects\\misc\\setup_gemini.ps1') {{ . 'C:\\projects\\misc\\setup_gemini.ps1' }}; " - f"gemini -p 'task' --output-format json --model {model}" + f"gemini -p '---' --output-format json --model {model}" ) cmd = ['powershell.exe', '-NoProfile', '-Command', ps_command] + try: process = subprocess.run(cmd, input=command_text, capture_output=True, text=True, encoding='utf-8') @@ -215,7 +214,7 @@ def execute_agent(role: str, prompt: str, docs: list[str]) -> str: result = f"Error: {process.stderr}" # Log the attempt and result - log_file = log_delegation(role, command_text, result) + log_file = log_delegation(role, command_text, result, summary_prompt=prompt) print(f"Sub-agent log created: {log_file}") stdout = process.stdout diff --git a/tests/test_mma_exec.py b/tests/test_mma_exec.py index 03522a0..39ed108 100644 --- a/tests/test_mma_exec.py +++ b/tests/test_mma_exec.py @@ -20,11 +20,13 @@ def test_parser_invalid_role(): with pytest.raises(SystemExit): parser.parse_args(['--role', 'tier5', 'Some prompt']) -def test_parser_prompt_required(): - """Test that the prompt argument is mandatory.""" +def test_parser_prompt_optional(): + """Test that the prompt argument is optional if role is provided (or handled in main).""" parser = create_parser() - with pytest.raises(SystemExit): - parser.parse_args(['--role', 'tier3']) + # Prompt is now optional (nargs='?') + args = parser.parse_args(['--role', 'tier3']) + assert args.role == 'tier3' + assert args.prompt is None def test_parser_help(): """Test that the help flag works without raising errors (exits with 0).""" @@ -49,14 +51,13 @@ def test_get_model_for_role(): def test_execute_agent(): """ - Test that execute_agent calls subprocess.run with the correct gemini CLI arguments + Test that execute_agent calls subprocess.run with powershell and the correct gemini CLI arguments including the model specified for the role. """ role = "tier3-worker" prompt = "Write a unit test." docs = ["file1.py", "docs/spec.md"] - expected_gemini_arg = "Use the mma-tier3-worker skill. Write a unit test. @file1.py @docs/spec.md" expected_model = "gemini-2.5-flash-lite" mock_stdout = "Mocked AI Response" @@ -72,14 +73,16 @@ def test_execute_agent(): args, kwargs = mock_run.call_args cmd_list = args[0] - assert cmd_list[0] == "gemini" - assert cmd_list[1] == "-p" - assert cmd_list[2] == expected_gemini_arg - - # Verify the correct model is passed via --model flag - assert "--model" in cmd_list - model_idx = cmd_list.index("--model") - assert cmd_list[model_idx + 1] == expected_model + assert cmd_list[0] == "powershell.exe" + assert "-Command" in cmd_list + ps_cmd = cmd_list[cmd_list.index("-Command") + 1] + assert "gemini" in ps_cmd + assert f"--model {expected_model}" in ps_cmd + + # Verify input contains the prompt and system directive + input_text = kwargs.get("input") + assert "STRICT SYSTEM DIRECTIVE" in input_text + assert "TASK: Write a unit test." in input_text assert kwargs.get("capture_output") is True assert kwargs.get("text") is True @@ -95,13 +98,15 @@ def test_get_dependencies(tmp_path): ) filepath = tmp_path / "mock_script.py" filepath.write_text(content) - dependencies = get_dependencies(filepath) + dependencies = get_dependencies(str(filepath)) assert dependencies == ['os', 'sys', 'file_cache', 'mcp_client'] import re def test_execute_agent_logging(tmp_path): log_file = tmp_path / "mma_delegation.log" + # mma_exec now uses logs/agents/ for individual logs and logs/mma_delegation.log for master + # We will patch LOG_FILE to point to our temp location with patch("scripts.mma_exec.LOG_FILE", str(log_file)), \ patch("subprocess.run") as mock_run: mock_process = MagicMock() @@ -114,10 +119,9 @@ def test_execute_agent_logging(tmp_path): assert log_file.exists() log_content = log_file.read_text() assert test_role in log_content - assert test_prompt in log_content + assert test_prompt in log_content # Master log should now have the summary prompt 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" @@ -125,6 +129,8 @@ def test_execute_agent_tier3_injection(tmp_path): 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) + + # We need to ensure generate_skeleton is mockable or working old_cwd = os.getcwd() os.chdir(tmp_path) try: @@ -135,10 +141,10 @@ def test_execute_agent_tier3_injection(tmp_path): 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 + input_text = mock_run.call_args[1].get("input") + assert "DEPENDENCY SKELETON: dependency.py" in input_text + assert "def do_work():" in input_text + assert "Modify main.py" in input_text finally: os.chdir(old_cwd) +