Something went wrong. Try again.
CLI tools and operational extensions for the void architecture on ATProtocol.
Something went wrong. Try again.
3.5 kB · 93 lines
Python
at master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293import osimport yamlimport sysimport refrom collections import Counter
def check_error_loops(): """Check for repeated error patterns in recent cron logs.""" LOG_FILE = 'cron_execution.log' CHECK_WINDOW = 5 # Last 5 cycles ERROR_THRESHOLD = 2 # Alert if same error appears 2+ times
if not os.path.exists(LOG_FILE): return False, []
try: with open(LOG_FILE, 'r') as f: lines = f.readlines()
# Find the last CHECK_WINDOW cycle markers cycle_starts = [] for i, line in enumerate(lines): if '--- Void Execution Loop Triggered' in line: cycle_starts.append(i)
# Get last CHECK_WINDOW cycles recent_cycles = cycle_starts[-CHECK_WINDOW:] if len(cycle_starts) >= CHECK_WINDOW else cycle_starts if not recent_cycles: return False, []
# Extract error patterns from recent cycles error_patterns = [] for i, start_idx in enumerate(recent_cycles): end_idx = recent_cycles[i+1] if i+1 < len(recent_cycles) else len(lines) cycle_content = ''.join(lines[start_idx:end_idx])
# Look for common error patterns if 'colon prefix' in cycle_content.lower() or ':cd:' in cycle_content or ':120000:' in cycle_content: error_patterns.append('colon_prefix_pollution') elif 'command not found' in cycle_content.lower(): error_patterns.append('command_not_found') elif 'error:' in cycle_content.lower(): error_patterns.append('generic_error')
# Count occurrences pattern_counts = Counter(error_patterns) repeated_patterns = [p for p, count in pattern_counts.items() if count >= ERROR_THRESHOLD]
return bool(repeated_patterns), repeated_patterns
except Exception as e: print(f"Warning: Error loop check failed: {e}") return False, []
def check_actionable_data(): has_actionable = False # Configuration Thresholds MIN_INBOX_THRESHOLD = 3 ADMIN_HANDLES = ['@cameron.stream', 'cameron.stream', '@cameron.pfiffer.org', 'cameron.pfiffer.org']
inbox_path = os.path.join(os.getcwd(), 'inbox.yaml') if os.path.exists(inbox_path): with open(inbox_path, 'r', encoding='utf-8') as f: try: inbox_data = yaml.safe_load(f) if inbox_data and isinstance(inbox_data, list): # Check for admin has_admin = any(item.get('author') in ADMIN_HANDLES for item in inbox_data) if has_admin: print("Trigger: Administrator notification found.") has_actionable = True elif len(inbox_data) >= MIN_INBOX_THRESHOLD: print(f"Trigger: Inbox threshold met ({len(inbox_data)} items).") has_actionable = True except yaml.YAMLError: pass
# Phase 2: Error loop detection has_error_loop, error_patterns = check_error_loops() if has_error_loop: print(f"ALERT: Detected error loop patterns: {', '.join(error_patterns)}") print("Recommendation: Manual intervention required.") # Still exit 0 to allow agent to attempt recovery, but alert is logged
if has_actionable: sys.exit(0) else: print("No actionable data found. Thresholds not met.") sys.exit(1)
if __name__ == '__main__': check_actionable_data()