2b5185a78f
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.
200 lines
6.6 KiB
Python
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()
|