Files
manual_slop/scripts/focused_entropy_audit.py
T
ed 2b5185a78f perf(entropy): Fix nested imports in hot paths
Hoisted imports from inside frequently-called functions to module level:

app_controller.py:
- Added traceback and inspect at module level
- Removed 3 nested traceback imports from exception handlers

gui_2.py:
- Added traceback at module level
- Removed nested traceback import from _gui_func exception handler
- Kept uvicorn lazy-loaded (only for --headless mode)

multi_agent_conductor.py:
- Removed unused 'import sys' from run()
- Removed redundant nested imports (already at module level)

Also adds audit scripts and entropy findings documentation.
2026-05-06 20:18:30 -04:00

200 lines
6.6 KiB
Python

#!/usr/bin/env python3
"""
Focused Entropy Audit for Manual Slop - Muratori Style
Focuses on ACTUAL issues, not style:
1. Duplicate logic (same thing done in multiple places)
2. State inconsistencies (parallel representations)
3. Logic errors / bugs
4. Performance concerns
"""
import os
import re
import ast
from pathlib import Path
from collections import defaultdict
from typing import List, Dict, Set, Tuple, Optional
def find_duplicate_logic_files():
"""Find files with similar patterns that might indicate duplicate logic."""
patterns = {
'calculate_track_progress': [],
'cascade_blocks': [],
'topological_sort': [],
'push_mma_state': [],
'active_tickets': [],
}
for f in Path('src').glob('*.py'):
content = f.read_text(encoding='utf-8', errors='ignore')
for pattern in patterns:
if re.search(pattern, content):
patterns[pattern].append(f.name)
return patterns
def check_ticket_state_management():
"""Check for state management issues in ticket handling."""
issues = []
# Check if there are parallel ticket representations
gui_2_content = Path('src/gui_2.py').read_text(encoding='utf-8', errors='ignore')
app_ctrl_content = Path('src/app_controller.py').read_text(encoding='utf-8', errors='ignore')
dag_content = Path('src/dag_engine.py').read_text(encoding='utf-8', errors='ignore')
# gui_2 uses dict-based tickets
if 'active_tickets' in gui_2_content:
if 'ticket["status"]' in gui_2_content or "t['status']" in gui_2_content:
issues.append(("gui_2.py", "Dict-based ticket access pattern found"))
# Check for blocking logic duplication
gui_blocking = len(re.findall(r'_cb_block_ticket|_cb_unblock_ticket', gui_2_content))
dag_blocking = len(re.findall(r'cascade_blocks', dag_content))
if gui_blocking > 0 and dag_blocking > 0:
issues.append(("architecture", "GUI has manual block/unblock that could conflict with DAG cascade_blocks"))
return issues
def check_import_issues():
"""Check for actual import problems - nested imports causing runtime issues."""
issues = []
for f in Path('src').glob('*.py'):
try:
content = f.read_text(encoding='utf-8', errors='ignore')
tree = ast.parse(content, filename=str(f))
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
for child in ast.walk(node):
if isinstance(child, (ast.Import, ast.ImportFrom)):
line = child.lineno or 0
# Check if this import is inside a HOT PATH function
if node.name in ['_process_pending_gui_tasks', '_gui_func', 'run', 'tick']:
issues.append((f.name, f"Nested import `{ast.unparse(child).strip()[:50]}` in hot path `{node.name}` line {line}"))
except:
pass
return issues
def check_potential_bugs():
"""Check for potential bugs - undefined variables, etc."""
bugs = []
# Check for == None vs is None patterns
for f in Path('src').glob('*.py'):
content = f.read_text(encoding='utf-8', errors='ignore')
lines = content.split('\n')
for i, line in enumerate(lines, 1):
# Skip comments
if line.strip().startswith('#'):
continue
# Check for mutable default arguments (common bug)
if re.search(r'def\s+\w+\([^)]*=\s*\[\s*\]', line):
bugs.append((f.name, i, "Mutable default argument", line.strip()[:60]))
if re.search(r'def\s+\w+\([^)]*=\s*\{\s*\}', line):
bugs.append((f.name, i, "Mutable default argument", line.strip()[:60]))
return bugs
def check_actual_duplicates():
"""Check for ACTUAL duplicate code - same logic copied."""
duplicates = []
seen_snippets = defaultdict(list)
# Look for duplicate patterns (3+ lines identical)
for f in sorted(Path('src').glob('*.py')):
try:
content = f.read_text(encoding='utf-8', errors='ignore')
lines = content.split('\n')
# Normalize and check consecutive duplicate lines
prev_normalized = None
dup_start = None
for i, line in enumerate(lines, 1):
if line.strip().startswith('#'):
prev_normalized = None
dup_start = None
continue
normalized = line.strip()
if not normalized:
prev_normalized = None
dup_start = None
continue
if normalized == prev_normalized:
if dup_start is None:
dup_start = i - 1
else:
if dup_start and i - dup_start >= 3:
# Found 3+ consecutive duplicate lines
duplicates.append((f.name, dup_start, i - 1, lines[dup_start-1].strip()[:60]))
dup_start = None
prev_normalized = normalized
except:
pass
return duplicates
def main():
print("=" * 70)
print("FOCUSED ENTROPY AUDIT - Muratori Style")
print("=" * 70)
print()
print("1. TICKET STATE MANAGEMENT ISSUES")
print("-" * 40)
issues = check_ticket_state_management()
if issues:
for issue in issues:
print(f" [{issue[0]}] {issue[1]}")
else:
print(" None found")
print()
print("2. NESTED IMPORTS IN HOT PATH FUNCTIONS")
print("-" * 40)
issues = check_import_issues()
if issues:
for fname, msg in issues[:10]:
print(f" [{fname}] {msg}")
else:
print(" None found")
print()
print("3. POTENTIAL BUGS (mutable defaults, etc)")
print("-" * 40)
issues = check_potential_bugs()
if issues:
for fname, line, bugtype, code in issues[:10]:
print(f" [{fname}:{line}] {bugtype}: {code}")
else:
print(" None found")
print()
print("4. ACTUAL DUPLICATE CODE (3+ consecutive lines)")
print("-" * 40)
duplicates = check_actual_duplicates()
if duplicates:
for fname, start, end, code in duplicates[:10]:
print(f" [{fname}:{start}-{end}] {code}")
else:
print(" None found")
print()
print("5. PATTERN USAGE ACROSS FILES")
print("-" * 40)
patterns = find_duplicate_logic_files()
for pattern, files in patterns.items():
if len(files) > 1:
print(f" {pattern}: {files}")
print()
if __name__ == "__main__":
main()