#!/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()