ed/manual_slop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(conductor): Apply review suggestions for track 'mma_utilization_refinement_20260226'

Browse Source
This commit is contained in:
Edward R. Gonzalez
2026-02-26 08:35:50 -05:00
parent 91693a5168
commit d343066435
4 changed files with 39 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.editorconfig
View File

@@ -2,7 +2,7 @@ root = true
[*.py]
indent_style = space
indent_size = 2
indent_size = 1
[*.s]
indent_style = tab

BIN
coverage_report.txt
View File

Binary file not shown.

19
scripts/mma_exec.py
View File

@@ -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,13 +198,14 @@ 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

48
tests/test_mma_exec.py
View File

@@ -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
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 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
# 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)