refactor(mma_exec): remove UNFETTERED_MODULES — all deps use generate_skeleton()
This commit is contained in:
@@ -7,7 +7,7 @@ Architecture reference: [docs/guide_architecture.md](../../../docs/guide_archite
|
|||||||
## Phase 1: Patch Context Amnesia Leak & Portability (Meta-Tooling)
|
## Phase 1: Patch Context Amnesia Leak & Portability (Meta-Tooling)
|
||||||
Focus: Stop `mma_exec.py` from injecting massive full-text dependencies and remove hardcoded external paths.
|
Focus: Stop `mma_exec.py` from injecting massive full-text dependencies and remove hardcoded external paths.
|
||||||
|
|
||||||
- [ ] Task 1.1: In `scripts/mma_exec.py`, completely remove the `UNFETTERED_MODULES` constant and its associated `if dep in UNFETTERED_MODULES:` check. Ensure all imported local dependencies strictly use `generate_skeleton()`.
|
- [~] Task 1.1: In `scripts/mma_exec.py`, completely remove the `UNFETTERED_MODULES` constant and its associated `if dep in UNFETTERED_MODULES:` check. Ensure all imported local dependencies strictly use `generate_skeleton()`.
|
||||||
- [ ] Task 1.2: In `scripts/mma_exec.py` and `scripts/claude_mma_exec.py`, remove the hardcoded reference to `C:\projects\misc\setup_*.ps1`. Rely on the active environment's PATH to resolve `gemini` and `claude`, or provide an `.env` configurable override.
|
- [ ] Task 1.2: In `scripts/mma_exec.py` and `scripts/claude_mma_exec.py`, remove the hardcoded reference to `C:\projects\misc\setup_*.ps1`. Rely on the active environment's PATH to resolve `gemini` and `claude`, or provide an `.env` configurable override.
|
||||||
|
|
||||||
## Phase 2: Complete MCP Tool Integration & Seal HITL Bypass (Application Core)
|
## Phase 2: Complete MCP Tool Integration & Seal HITL Bypass (Application Core)
|
||||||
|
|||||||
@@ -134,9 +134,6 @@ def execute_agent(role: str, prompt: str, docs: list[str], debug: bool = False,
|
|||||||
|
|
||||||
# Advanced Context: Dependency skeletons for Tier 3
|
# Advanced Context: Dependency skeletons for Tier 3
|
||||||
injected_context = ""
|
injected_context = ""
|
||||||
# Whitelist of modules that sub-agents have "unfettered" (full) access to.
|
|
||||||
# These will be provided in full if imported, instead of just skeletons.
|
|
||||||
UNFETTERED_MODULES: list[str] = ['mcp_client', 'project_manager', 'events', 'aggregate']
|
|
||||||
if role in ['tier3', 'tier3-worker']:
|
if role in ['tier3', 'tier3-worker']:
|
||||||
for doc in docs:
|
for doc in docs:
|
||||||
if doc.endswith('.py') and os.path.exists(doc):
|
if doc.endswith('.py') and os.path.exists(doc):
|
||||||
@@ -150,14 +147,9 @@ def execute_agent(role: str, prompt: str, docs: list[str], debug: bool = False,
|
|||||||
continue
|
continue
|
||||||
if os.path.exists(dep_file) and dep_file != doc:
|
if os.path.exists(dep_file) and dep_file != doc:
|
||||||
try:
|
try:
|
||||||
if dep in UNFETTERED_MODULES:
|
with open(dep_file, 'r', encoding='utf-8') as f:
|
||||||
with open(dep_file, 'r', encoding='utf-8') as f:
|
skeleton = generate_skeleton(f.read())
|
||||||
full_content = f.read()
|
injected_context += f"\n\nDEPENDENCY SKELETON: {dep_file}\n{skeleton}\n"
|
||||||
injected_context += f"\n\nFULL MODULE CONTEXT: {dep_file}\n{full_content}\n"
|
|
||||||
else:
|
|
||||||
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:
|
except Exception as e:
|
||||||
print(f"Error gathering context for {dep_file}: {e}")
|
print(f"Error gathering context for {dep_file}: {e}")
|
||||||
# Check for token-bloat safety: if injected_context is too large, truncate it
|
# Check for token-bloat safety: if injected_context is too large, truncate it
|
||||||
|
|||||||
71
tests/test_arch_boundary_phase1.py
Normal file
71
tests/test_arch_boundary_phase1.py
Normal file
@@ -0,0 +1,71 @@
|
|||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
import unittest.mock as mock
|
||||||
|
import importlib
|
||||||
|
import inspect
|
||||||
|
import tempfile
|
||||||
|
import shutil
|
||||||
|
|
||||||
|
# Ensure scripts directory is in sys.path
|
||||||
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'scripts')))
|
||||||
|
import mma_exec
|
||||||
|
|
||||||
|
class TestArchBoundaryPhase1(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
importlib.reload(mma_exec)
|
||||||
|
self.test_dir = tempfile.mkdtemp()
|
||||||
|
self.old_cwd = os.getcwd()
|
||||||
|
os.chdir(self.test_dir)
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
os.chdir(self.old_cwd)
|
||||||
|
shutil.rmtree(self.test_dir)
|
||||||
|
|
||||||
|
def test_unfettered_modules_constant_removed(self):
|
||||||
|
"""TEST 1: Check 'UNFETTERED_MODULES' string absent from execute_agent source."""
|
||||||
|
source = inspect.getsource(mma_exec.execute_agent)
|
||||||
|
self.assertNotIn('UNFETTERED_MODULES', source, "UNFETTERED_MODULES constant should be removed from execute_agent")
|
||||||
|
|
||||||
|
def test_full_module_context_never_injected(self):
|
||||||
|
"""TEST 2: Verify 'FULL MODULE CONTEXT' not in captured input for mcp_client."""
|
||||||
|
# Create a target file that imports mcp_client
|
||||||
|
target_py = os.path.join(self.test_dir, "target.py")
|
||||||
|
with open(target_py, "w") as f:
|
||||||
|
f.write("import mcp_client\n")
|
||||||
|
|
||||||
|
# Create mcp_client.py
|
||||||
|
mcp_client_py = os.path.join(self.test_dir, "mcp_client.py")
|
||||||
|
with open(mcp_client_py, "w") as f:
|
||||||
|
f.write("def dummy(): pass\n")
|
||||||
|
|
||||||
|
with mock.patch('subprocess.run') as mock_run:
|
||||||
|
mock_run.return_value = mock.Mock(stdout='{"response": "ok"}', returncode=0)
|
||||||
|
mma_exec.execute_agent('tier3-worker', 'test task', [target_py])
|
||||||
|
|
||||||
|
# Capture the input passed to subprocess.run
|
||||||
|
captured_input = mock_run.call_args[1].get('input', '')
|
||||||
|
self.assertNotIn('FULL MODULE CONTEXT: mcp_client.py', captured_input)
|
||||||
|
|
||||||
|
def test_skeleton_used_for_mcp_client(self):
|
||||||
|
"""TEST 3: Verify 'DEPENDENCY SKELETON' is used for mcp_client."""
|
||||||
|
# Create a target file that imports mcp_client
|
||||||
|
target_py = os.path.join(self.test_dir, "target.py")
|
||||||
|
with open(target_py, "w") as f:
|
||||||
|
f.write("import mcp_client\n")
|
||||||
|
|
||||||
|
# Create mcp_client.py
|
||||||
|
mcp_client_py = os.path.join(self.test_dir, "mcp_client.py")
|
||||||
|
with open(mcp_client_py, "w") as f:
|
||||||
|
f.write("def dummy(): pass\n")
|
||||||
|
|
||||||
|
with mock.patch('subprocess.run') as mock_run:
|
||||||
|
mock_run.return_value = mock.Mock(stdout='{"response": "ok"}', returncode=0)
|
||||||
|
mma_exec.execute_agent('tier3-worker', 'test task', [target_py])
|
||||||
|
|
||||||
|
# Capture the input passed to subprocess.run
|
||||||
|
captured_input = mock_run.call_args[1].get('input', '')
|
||||||
|
self.assertIn('DEPENDENCY SKELETON: mcp_client.py', captured_input)
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user