"""Echo Core (eco) service + session + doctor endpoints.""" import json import os import shutil import subprocess from datetime import datetime from pathlib import Path from urllib.parse import parse_qs, urlparse import constants class EcoHandlers: """Mixin for /api/eco/* endpoints.""" # ── /api/eco/status ───────────────────────────────────────── def handle_eco_status(self): """Get status of echo-core services + active sessions.""" try: services = [] for svc in constants.ECO_SERVICES: info = {'name': svc, 'active': False, 'pid': None, 'uptime': None, 'memory': None} result = subprocess.run( ['systemctl', '--user', 'is-active', svc], capture_output=True, text=True, timeout=5, ) info['active'] = result.stdout.strip() == 'active' if info['active']: result = subprocess.run( ['systemctl', '--user', 'show', '-p', 'MainPID', '--value', svc], capture_output=True, text=True, timeout=5, ) pid = result.stdout.strip() if pid and pid != '0': info['pid'] = int(pid) try: r = subprocess.run( ['systemctl', '--user', 'show', '-p', 'ActiveEnterTimestamp', '--value', svc], capture_output=True, text=True, timeout=5, ) ts = r.stdout.strip() if ts: start = datetime.strptime(ts, '%a %Y-%m-%d %H:%M:%S %Z') info['uptime'] = int((datetime.utcnow() - start).total_seconds()) except Exception: pass try: for line in Path(f'/proc/{pid}/status').read_text().splitlines(): if line.startswith('VmRSS:'): info['memory'] = line.split(':')[1].strip() break except Exception: pass services.append(info) self.send_json({'services': services}) except Exception as e: self.send_json({'error': str(e)}, 500) # ── sessions ──────────────────────────────────────────────── def _eco_channel_map(self): """Build channel_id -> {name, platform, is_group} from config.json.""" config_file = constants.ECHO_CORE_DIR / 'config.json' m = {} try: cfg = json.loads(config_file.read_text()) for name, ch in cfg.get('channels', {}).items(): m[str(ch['id'])] = {'name': name, 'platform': 'discord'} for name, ch in cfg.get('telegram_channels', {}).items(): m[str(ch['id'])] = {'name': name, 'platform': 'telegram'} for name, ch in cfg.get('whatsapp_channels', {}).items(): m[str(ch['id'])] = {'name': name, 'platform': 'whatsapp', 'is_group': True} for admin_id in cfg.get('bot', {}).get('admins', []): m.setdefault(str(admin_id), {'name': 'TG DM', 'platform': 'telegram'}) wa_owner = cfg.get('whatsapp', {}).get('owner', '') if wa_owner: m.setdefault(f'wa-{wa_owner}', {'name': 'WA Owner', 'platform': 'whatsapp'}) except Exception: pass return m def _eco_enrich_sessions(self): """Return enriched sessions list sorted by last_message_at desc.""" raw = {} if constants.ECHO_SESSIONS_FILE.exists(): try: raw = json.loads(constants.ECHO_SESSIONS_FILE.read_text()) except Exception: pass cmap = self._eco_channel_map() sessions = [] if isinstance(raw, dict): for ch_id, sdata in raw.items(): if 'MagicMock' in ch_id: continue entry = dict(sdata) if isinstance(sdata, dict) else {} entry['channel_id'] = ch_id if ch_id in cmap: entry['platform'] = cmap[ch_id]['platform'] entry['channel_name'] = cmap[ch_id]['name'] entry['is_group'] = cmap[ch_id].get('is_group', False) elif ch_id.startswith('wa-') or '@g.us' in ch_id or '@s.whatsapp.net' in ch_id: entry['platform'] = 'whatsapp' entry['is_group'] = '@g.us' in ch_id entry['channel_name'] = ('WA Grup' if entry['is_group'] else 'WA DM') elif ch_id.isdigit() and len(ch_id) >= 17: entry['platform'] = 'discord' entry['channel_name'] = 'Discord #' + ch_id[-6:] elif ch_id.isdigit(): entry['platform'] = 'telegram' entry['channel_name'] = 'TG ' + ch_id else: entry['platform'] = 'unknown' entry['channel_name'] = ch_id[:20] sessions.append(entry) sessions.sort(key=lambda s: s.get('last_message_at', ''), reverse=True) return sessions def handle_eco_sessions(self): """Return enriched sessions list.""" try: self.send_json({'sessions': self._eco_enrich_sessions()}) except Exception as e: self.send_json({'error': str(e)}, 500) def handle_eco_session_content(self): """Return conversation messages from a Claude session transcript.""" try: params = parse_qs(urlparse(self.path).query) session_id = params.get('id', [''])[0] if not session_id or '/' in session_id or '..' in session_id: self.send_json({'error': 'Invalid session id'}, 400) return transcript = Path.home() / '.claude' / 'projects' / '-home-moltbot-echo-core' / f'{session_id}.jsonl' if not transcript.exists(): self.send_json({'messages': [], 'error': 'Transcript not found'}) return messages = [] for line in transcript.read_text().splitlines(): try: d = json.loads(line) except Exception: continue t = d.get('type', '') if t == 'user': msg = d.get('message', {}) content = msg.get('content', '') if isinstance(content, str): text = content.replace('[EXTERNAL CONTENT]\n', '').replace('\n[END EXTERNAL CONTENT]', '').strip() if text: messages.append({'role': 'user', 'text': text[:20000]}) elif t == 'assistant': msg = d.get('message', {}) content = msg.get('content', '') if isinstance(content, list): parts = [block['text'] for block in content if block.get('type') == 'text'] text = '\n'.join(parts).strip() if text: messages.append({'role': 'assistant', 'text': text[:20000]}) elif isinstance(content, str) and content.strip(): messages.append({'role': 'assistant', 'text': content[:20000]}) self.send_json({'messages': messages}) except Exception as e: self.send_json({'error': str(e)}, 500) def handle_eco_sessions_clear(self): """Clear active sessions (all or specific channel).""" try: data = self._read_post_json() channel = data.get('channel', None) if not constants.ECHO_SESSIONS_FILE.exists(): self.send_json({'success': True, 'message': 'No sessions file'}) return if channel: sessions = json.loads(constants.ECHO_SESSIONS_FILE.read_text()) if isinstance(sessions, list): sessions = [s for s in sessions if s.get('channel') != channel] elif isinstance(sessions, dict): sessions.pop(channel, None) constants.ECHO_SESSIONS_FILE.write_text(json.dumps(sessions, indent=2)) self.send_json({'success': True, 'message': f'Cleared session: {channel}'}) else: if isinstance(json.loads(constants.ECHO_SESSIONS_FILE.read_text()), list): constants.ECHO_SESSIONS_FILE.write_text('[]') else: constants.ECHO_SESSIONS_FILE.write_text('{}') self.send_json({'success': True, 'message': 'All sessions cleared'}) except Exception as e: self.send_json({'success': False, 'error': str(e)}, 500) # ── logs + doctor ─────────────────────────────────────────── def handle_eco_logs(self): """Return last N lines from echo-core.log.""" try: params = parse_qs(urlparse(self.path).query) lines = min(int(params.get('lines', ['100'])[0]), 500) if not constants.ECHO_LOG_FILE.exists(): self.send_json({'lines': ['(log file not found)']}) return result = subprocess.run( ['tail', '-n', str(lines), str(constants.ECHO_LOG_FILE)], capture_output=True, text=True, timeout=10, ) self.send_json({'lines': result.stdout.splitlines()}) except Exception as e: self.send_json({'error': str(e)}, 500) def handle_eco_doctor(self): """Run health checks on the echo-core ecosystem.""" checks = [] # 1. Services for svc in constants.ECO_SERVICES: try: r = subprocess.run( ['systemctl', '--user', 'is-active', svc], capture_output=True, text=True, timeout=5, ) active = r.stdout.strip() == 'active' checks.append({ 'name': f'Service: {svc}', 'pass': active, 'detail': 'active' if active else r.stdout.strip(), }) except Exception as e: checks.append({'name': f'Service: {svc}', 'pass': False, 'detail': str(e)}) # 2. Disk space try: st = shutil.disk_usage('/') pct_free = (st.free / st.total) * 100 checks.append({ 'name': 'Disk space', 'pass': pct_free > 5, 'detail': f'{pct_free:.1f}% free ({st.free // (1024**3)} GB)', }) except Exception as e: checks.append({'name': 'Disk space', 'pass': False, 'detail': str(e)}) # 3. Log file try: if constants.ECHO_LOG_FILE.exists(): size_mb = constants.ECHO_LOG_FILE.stat().st_size / (1024 * 1024) checks.append({ 'name': 'Log file', 'pass': size_mb < 100, 'detail': f'{size_mb:.1f} MB', }) else: checks.append({'name': 'Log file', 'pass': False, 'detail': 'Not found'}) except Exception as e: checks.append({'name': 'Log file', 'pass': False, 'detail': str(e)}) # 4. Sessions file try: if constants.ECHO_SESSIONS_FILE.exists(): data = json.loads(constants.ECHO_SESSIONS_FILE.read_text()) count = len(data) if isinstance(data, list) else len(data.keys()) if isinstance(data, dict) else 0 checks.append({'name': 'Sessions file', 'pass': True, 'detail': f'{count} active'}) else: checks.append({'name': 'Sessions file', 'pass': False, 'detail': 'Not found'}) except Exception as e: checks.append({'name': 'Sessions file', 'pass': False, 'detail': str(e)}) # 5. Config config_file = constants.ECHO_CORE_DIR / 'config.json' try: if config_file.exists(): json.loads(config_file.read_text()) checks.append({'name': 'Config', 'pass': True, 'detail': 'Valid JSON'}) else: checks.append({'name': 'Config', 'pass': False, 'detail': 'Not found'}) except Exception as e: checks.append({'name': 'Config', 'pass': False, 'detail': str(e)}) # 6. WhatsApp bridge log wa_log = constants.ECHO_CORE_DIR / 'logs' / 'whatsapp-bridge.log' try: if wa_log.exists(): r = subprocess.run(['tail', '-1', str(wa_log)], capture_output=True, text=True, timeout=5) last = r.stdout.strip() has_error = 'error' in last.lower() or 'fatal' in last.lower() checks.append({ 'name': 'WhatsApp bridge log', 'pass': not has_error, 'detail': last[:80] if last else 'Empty', }) else: checks.append({'name': 'WhatsApp bridge log', 'pass': False, 'detail': 'Not found'}) except Exception as e: checks.append({'name': 'WhatsApp bridge log', 'pass': False, 'detail': str(e)}) # 7. Claude CLI try: r = subprocess.run(['which', 'claude'], capture_output=True, text=True, timeout=5) found = r.returncode == 0 checks.append({ 'name': 'Claude CLI', 'pass': found, 'detail': r.stdout.strip() if found else 'Not in PATH', }) except Exception as e: checks.append({'name': 'Claude CLI', 'pass': False, 'detail': str(e)}) self.send_json({'checks': checks}) # ── service control ───────────────────────────────────────── def handle_eco_restart(self): """Restart an echo-core service (not the taskboard itself).""" try: data = self._read_post_json() svc = data.get('service', '') if svc not in constants.ECO_SERVICES: self.send_json({'success': False, 'error': f'Unknown service: {svc}'}, 400) return if svc == 'echo-taskboard': self.send_json({'success': False, 'error': 'Cannot restart taskboard from itself'}, 400) return result = subprocess.run( ['systemctl', '--user', 'restart', svc], capture_output=True, text=True, timeout=30, ) if result.returncode == 0: self.send_json({'success': True, 'message': f'{svc} restarted'}) else: self.send_json({'success': False, 'error': result.stderr.strip()}, 500) except Exception as e: self.send_json({'success': False, 'error': str(e)}, 500) def handle_eco_stop(self): """Stop an echo-core service (not the taskboard itself).""" try: data = self._read_post_json() svc = data.get('service', '') if svc not in constants.ECO_SERVICES: self.send_json({'success': False, 'error': f'Unknown service: {svc}'}, 400) return if svc == 'echo-taskboard': self.send_json({'success': False, 'error': 'Cannot stop taskboard from itself'}, 400) return result = subprocess.run( ['systemctl', '--user', 'stop', svc], capture_output=True, text=True, timeout=30, ) if result.returncode == 0: self.send_json({'success': True, 'message': f'{svc} stopped'}) else: self.send_json({'success': False, 'error': result.stderr.strip()}, 500) except Exception as e: self.send_json({'success': False, 'error': str(e)}, 500) def handle_eco_restart_taskboard(self): """Restart the taskboard itself. Sends response then exits; systemd restarts.""" import threading self.send_json({'success': True, 'message': 'Restarting taskboard in 1s...'}) def _exit(): import time time.sleep(1) os._exit(0) threading.Thread(target=_exit, daemon=True).start()