from PyQt5.QtWidgets import * from PyQt5.QtCore import * from PyQt5.QtGui import * from services.system_monitor import get_system_snapshot from services.caretaker_service import caretaker from core.version import get_display_name, get_footer from core.event_bus import event_bus from core.service_manager import service_manager from core.task_manager import task_manager from core.agent_manager import agent_manager from core.shire_runtime import shire_runtime from framework.warden_theme import * from framework.warden_widgets import * from framework.warden_scroll import WardenTouchScrollArea import time from core.cooling_manager import cooling_manager class MissionStatusTile(WardenPanel): clicked = pyqtSignal(str) def __init__(self, key, title, state="online", icon="●", subtitle=""): super().__init__(state_colour(state)) self.key = key self.title_text = title self.subtitle_text = subtitle self.icon_text = icon self.state = state self._press_pos = None self.setCursor(Qt.PointingHandCursor) self.setAttribute(Qt.WA_AcceptTouchEvents, True) self.setMinimumHeight(68) layout = QVBoxLayout() layout.setContentsMargins(4, 3, 4, 3) layout.setSpacing(0) self.icon_label = QLabel(icon) self.icon_label.setAlignment(Qt.AlignCenter) self.icon_label.setAttribute(Qt.WA_TransparentForMouseEvents, True) self.icon_label.setMinimumHeight(36) self.icon_label.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Expanding) self.title_label = QLabel(title) self.title_label.setAlignment(Qt.AlignCenter) self.title_label.setWordWrap(True) self.title_label.setAttribute(Qt.WA_TransparentForMouseEvents, True) self.title_label.setMaximumHeight(15) self.subtitle_label = QLabel(subtitle) self.subtitle_label.setAlignment(Qt.AlignCenter) self.subtitle_label.setWordWrap(True) self.subtitle_label.setAttribute(Qt.WA_TransparentForMouseEvents, True) self.subtitle_label.setMaximumHeight(12) layout.addWidget(self.icon_label) layout.addWidget(self.title_label) layout.addWidget(self.subtitle_label) self.setLayout(layout) self.apply_state(state) def background_for_state(self, state): if state in ["offline", "error", "critical", "failed"]: return "rgba(65, 0, 0, 235)" if state in ["standby", "partial"]: return "rgba(34, 0, 58, 235)" if state in ["warning"]: return "rgba(70, 45, 0, 235)" if state in ["booting", "loading", "reading_heart", "reading_codex", "loading_memory", "checking_tasks", "checking_agents"]: return "rgba(0, 35, 65, 235)" return "rgba(0, 34, 14, 235)" def apply_state(self, state): self.state = state colour = state_colour(state) bg = self.background_for_state(state) self.normal_style = f""" QFrame {{ background-color: {bg}; border: 3px solid {colour}; border-radius: 12px; }} """ self.pressed_style = f""" QFrame {{ background-color: rgba(40, 0, 65, 245); border: 3px solid {PURPLE}; border-radius: 12px; }} """ self.setStyleSheet(self.normal_style) self.icon_label.setStyleSheet(f""" color:{colour}; font-size:40px; font-weight:bold; border:none; background:transparent; """) self.title_label.setStyleSheet(f""" color:{WHITE}; font-size:8px; font-weight:bold; border:none; background:transparent; """) self.subtitle_label.setStyleSheet(f""" color:{colour}; font-size:9px; font-weight:bold; border:none; background:transparent; """) def set_value(self, value, state=None): if state: self.apply_state(state) def mousePressEvent(self, event): self._press_pos = event.pos() self.setStyleSheet(self.pressed_style) super().mousePressEvent(event) def mouseReleaseEvent(self, event): self.setStyleSheet(self.normal_style) if self._press_pos is not None: dx = abs(event.pos().x() - self._press_pos.x()) dy = abs(event.pos().y() - self._press_pos.y()) if dx < 12 and dy < 12: self.clicked.emit(self.key) self._press_pos = None super().mouseReleaseEvent(event) class Dashboard(WardenBackground): def __init__(self, stack): super().__init__() self.stack = stack self.setAutoFillBackground(False) outer = QVBoxLayout() outer.setContentsMargins(8, 4, 8, 4) outer.setSpacing(4) self.scroll = WardenTouchScrollArea() holder = QWidget() holder.setStyleSheet("background: transparent;") holder.setMinimumHeight(720) main = QVBoxLayout() main.setContentsMargins(0, 0, 0, 0) main.setSpacing(8) title = QLabel("⚒ ARMOR FORGE CONTROL ⚒") title.setAlignment(Qt.AlignCenter) title.setStyleSheet(f""" color:{GREEN}; font-size:18px; font-weight:bold; background:transparent; """) title.setWordWrap(True) main.addWidget(title) self.version = QLabel(get_display_name() + " | CORE-01") self.version.setAlignment(Qt.AlignCenter) self.version.setStyleSheet(f""" color:{WHITE}; font-size:10px; font-weight:bold; background:transparent; """) main.addWidget(self.version) self.cards = {} status_grid = QGridLayout() status_grid.setSpacing(4) status_grid.setColumnStretch(0, 1) status_grid.setColumnStretch(1, 1) status_items = [ ("heart", "OATH", "ONLINE", "online", "♥", "HEART"), ("shire", "SHIRE", "BOOT", "booting", "◉", "AIOS"), ("sentinel", "WATCH", "ONLINE", "online", "♜", "SENTINEL"), ("memory", "MEMORY", "ONLINE", "online", "▤", "RECALL"), ("forge", "FORGE", "READY", "standby", "⚒", "BUILD"), ("caretaker", "CARE", "ONLINE", "online", "⚙", "SYSTEM"), ("agents", "CREW", "READY", "standby", "♟", "AGENTS"), ("atlas", "ATLAS", "READY", "standby", "◇", "MAP"), ] for i, item in enumerate(status_items): key, name, value, state, icon, sub = item card = MissionStatusTile(key, name, state, icon, sub) card.clicked.connect(self.handle_status_tile) self.cards[key] = card status_grid.addWidget(card, i // 2, i % 2) self.core = ForgeCore() self.core.setMinimumHeight(190) main.addWidget(self.core) main.addLayout(status_grid) self.smart_detail = QLabel() self.smart_detail.setAlignment(Qt.AlignLeft | Qt.AlignTop) self.smart_detail.setWordWrap(True) self.smart_detail.setStyleSheet(f""" color:{WHITE}; font-size:11px; font-weight:bold; background-color:{PANEL_BG}; border:2px solid {GREEN}; border-radius:8px; padding:8px; """) self.smart_detail.setMinimumHeight(110) self.smart_detail.setText( "SMART MISSION CONTROL\n" "Tap a service tile to inspect it or open its station." ) main.addWidget(self.smart_detail) self.quick_panel = WardenPanel(PURPLE) self.quick_panel.setMinimumHeight(96) self.quick_grid = QGridLayout() self.quick_grid.setContentsMargins(5, 5, 5, 5) self.quick_grid.setSpacing(5) self.quick_panel.setLayout(self.quick_grid) main.addWidget(self.quick_panel) self.health_summary = QLabel() self.health_summary.setAlignment(Qt.AlignLeft | Qt.AlignTop) self.health_summary.setWordWrap(True) self.health_summary.setStyleSheet(f""" color:{WHITE}; font-size:11px; font-weight:bold; background-color:{PANEL_BG}; border:2px solid {GREEN}; border-radius:8px; padding:8px; """) self.health_summary.setMinimumHeight(115) self.health_summary.setText( "HEALTH SUMMARY\n" "Awaiting first system scan..." ) main.addWidget(self.health_summary) system_box = QGridLayout() system_box.setSpacing(6) self.cpu = WardenStatusCard("CPU", "---", "online", "▣") self.temp = WardenStatusCard("TEMP", "---", "online", "♨") self.ram = WardenStatusCard("RAM", "---", "online", "▥") self.disk = WardenStatusCard("DISK", "---", "online", "▧") self.network = WardenStatusCard("NET", "---", "online", "◎") system_box.addWidget(self.cpu, 0, 0) system_box.addWidget(self.temp, 0, 1) system_box.addWidget(self.ram, 1, 0) system_box.addWidget(self.disk, 1, 1) system_box.addWidget(self.network, 2, 0, 1, 2) main.addLayout(system_box) self.log = QLabel() self.log.setAlignment(Qt.AlignLeft | Qt.AlignTop) self.log.setWordWrap(True) self.log.setStyleSheet(f""" color:{WHITE}; font-size:10px; font-weight:bold; background-color:{PANEL_BG}; border:2px solid {PURPLE}; border-radius:8px; padding:8px; """) self.log.setMinimumHeight(115) main.addWidget(self.log) self.objective = QLabel() self.objective.setAlignment(Qt.AlignCenter) self.objective.setWordWrap(True) self.objective.setStyleSheet(f""" color:{GREEN}; font-size:11px; font-weight:bold; background-color:{PANEL_BG}; border:2px solid {PURPLE}; border-radius:8px; padding:8px; """) self.objective.setMinimumHeight(95) main.addWidget(self.objective) hint = QLabel("DRAG UP / DOWN TO SCROLL ◀ ▶ CHANGE STATION") hint.setAlignment(Qt.AlignCenter) hint.setStyleSheet(f""" color:{PURPLE}; font-size:9px; font-weight:bold; background:transparent; """) main.addWidget(hint) footer = QLabel(get_footer()) footer.setAlignment(Qt.AlignCenter) footer.setWordWrap(True) footer.setStyleSheet(f""" color:{PURPLE}; font-size:8px; font-weight:bold; background:transparent; """) main.addWidget(footer) main.addStretch() holder.setLayout(main) self.scroll.setWidget(holder) outer.addWidget(self.scroll, 1) self.setLayout(outer) event_bus.publish("system", "boot", "Mission Control V2 Started", "online") service_manager.start_core_services() shire_runtime.boot() self.timer = QTimer() self.timer.timeout.connect(self.update_status) self.timer.start(1000) self.update_status() self.show_default_actions() def paintEvent(self, event): painter = QPainter(self) painter.setRenderHint(QPainter.Antialiasing) self.paint_warden_background(painter) super().paintEvent(event) def temp_state(self, temp): try: t = float(temp) if t >= 80: return "critical" if t >= 65: return "warning" return "online" except Exception: return "offline" def update_log(self): events = event_bus.get_events(6) lines = ["SYSTEM LOG"] for e in events: lines.append(f'{e["time"]} {e["message"]}') self.log.setText("\n".join(lines)) def set_smart_detail(self, title, lines): body = [title] for line in lines: body.append(f"● {line}") self.smart_detail.setText("\n".join(body)) try: QTimer.singleShot(80, lambda: self.scroll.ensureWidgetVisible(self.smart_detail, 0, 20)) except Exception: pass def clear_quick_actions(self): while self.quick_grid.count(): item = self.quick_grid.takeAt(0) widget = item.widget() if widget: widget.deleteLater() def add_quick_action(self, index, icon, title, subtitle, callback, state="online"): btn = WardenTouchButton(icon, title, subtitle, state) btn.setMinimumHeight(42) btn.clicked.connect(callback) row = index // 2 col = index % 2 self.quick_grid.addWidget(btn, row, col) def set_quick_actions(self, actions): self.clear_quick_actions() if not actions: actions = [ ("◉", "MISSION", "READY", self.show_default_actions, "online") ] for index, action in enumerate(actions[:6]): icon, title, subtitle, callback, state = action self.add_quick_action(index, icon, title, subtitle, callback, state) def open_station(self, index, label): self.set_smart_detail("OPENING STATION", [ f"Opening {label}.", "Use Mission Control or swipe to return." ]) self.stack.setCurrentIndex(index) def open_view_stls_launcher(self): self.set_smart_detail("VIEW STLS — READ ONLY", [ "Opening SHIRE AI Core.", "Use SHIRE command: VIEW STLS", "Or use SHIRE Action Menu: View STLs.", "Read-only folder/index viewer only.", "No Blender run, no export, no STL generation." ]) self.stack.setCurrentIndex(1) def show_default_actions(self): self.set_smart_detail("FORGE CONTROL", [ "Tap a rune tile to open its station.", "Quick actions are active.", "The forge is ready." ]) self.set_quick_actions([ ("◉", "SHIRE", "AI CORE", lambda: self.open_station(1, "SHIRE AI Core"), "online"), ("⚒", "FORGE", "STATION", lambda: self.open_station(2, "Forge Station"), "standby"), ("☑", "TASKS", "QUEUE", lambda: self.open_station(8, "Task Station"), "online"), ("♟", "AGENTS", "WORKERS", lambda: self.open_station(7, "Agent Manager"), "standby"), ("▧", "VIEW", "STLS", self.open_view_stls_launcher, "standby"), ]) def show_heart_oath(self): self.set_smart_detail("BLACKSMITH'S OATH", [ "Prime Directive: Protect Ray and his family.", "Help with legal, ethical, defensive work.", "Stay calm, clear, useful, and loyal to the build." ]) def refresh_caretaker(self): result = caretaker.check() snapshot = result.get("snapshot", {}) thresholds = result.get("thresholds", {}) cooling = cooling_manager.snapshot() self.set_smart_detail("CARETAKER REFRESH", [ f"Temp: {snapshot.get('temp', '---')}°C", f"Temp state: {result.get('temp_state', '---').upper()}", f"Disk: {snapshot.get('disk', '---')}%", f"Disk state: {result.get('disk_state', '---').upper()}", f"Temp warning starts at {thresholds.get('temp_warning_c', 65.0)}°C.", cooling.get("summary", "Cooling HAT: not detected") ]) def show_memory_detail(self): pending = task_manager.get_pending_count() total_tasks = len(getattr(task_manager, "tasks", [])) queued = 0 completed = 0 awaiting = 0 for task in getattr(task_manager, "tasks", []): status = task.get("status") if status == "queued": queued += 1 elif status in ["completed", "done"]: completed += 1 elif status == "awaiting_approval": awaiting += 1 self.set_smart_detail("MEMORY / TASKS", [ f"Memory service: {service_manager.get_state('memory').upper()}", f"Pending tasks: {pending}", f"Queued: {queued}", f"Awaiting approval: {awaiting}", f"Completed: {completed}", f"Total task records: {total_tasks}" ]) self.set_quick_actions([ ("☑", "OPEN", "TASKS", lambda: self.open_station(8, "Task Station"), "online"), ("♟", "OPEN", "AGENTS", lambda: self.open_station(7, "Agent Manager"), "standby"), ("◉", "SHIRE", "AI CORE", lambda: self.open_station(1, "SHIRE AI Core"), "online"), ("↩", "RESET", "PANEL", self.show_default_actions, "standby"), ]) def show_cooling_detail(self): data = cooling_manager.snapshot() detected = "YES" if data.get("detected") else "NO" temp = data.get("temp", "---") recommended = data.get("recommended_speed", 0) last_set = data.get("last_set_speed") state = data.get("state", "unknown").upper() if last_set is None: last_set_text = "Not set by ARMOR yet" elif last_set == 0: last_set_text = "OFF" else: last_set_text = f"{last_set}%" lines = [ f"HAT detected: {detected}", f"CPU temp: {temp}°C", f"Cooling state: {state}", f"Recommended fan: {recommended}%", f"Last ARMOR fan command: {last_set_text}", data.get("auto_status", "Auto cooling: unknown"), ] if not data.get("detected"): lines.append("Check I2C and confirm 0x0d appears.") self.set_smart_detail("COOLING HAT", lines) self.set_quick_actions([ ("◎", "AUTO", "ON", self.auto_cooling_on_quick, "online"), ("◌", "AUTO", "OFF", self.auto_cooling_off_quick, "standby"), ("▲", "FAN", "MAX", lambda: self.set_fan_quick(100), "critical"), ("○", "FAN", "OFF", lambda: self.set_fan_quick(0), "standby"), ]) def auto_cool_quick(self): result = cooling_manager.auto_cool_now() data = cooling_manager.snapshot() temp = result.get("temp", data.get("temp", "---")) speed = result.get("recommended_speed", data.get("recommended_speed", 0)) title = "AUTO COOLING SENT" if result.get("ok") else "AUTO COOLING FAILED" if speed == 0: speed_text = "OFF" else: speed_text = f"{speed}%" self.set_smart_detail(title, [ result.get("message", "No response."), f"CPU temp: {temp}°C", f"ARMOR recommended: {speed_text}", "Cooling curve is now more aggressive above 60°C." ]) self.set_quick_actions([ ("◎", "AUTO", "ON", self.auto_cooling_on_quick, "online"), ("◌", "AUTO", "OFF", self.auto_cooling_off_quick, "standby"), ("▲", "FAN", "MAX", lambda: self.set_fan_quick(100), "critical"), ("○", "FAN", "OFF", lambda: self.set_fan_quick(0), "standby"), ]) def auto_cooling_on_quick(self): result = cooling_manager.enable_auto_cooling() self.set_smart_detail("AUTO COOLING ON", [ result.get("message", "Auto cooling enabled."), "ARMOR will manage fan speed in the background.", "Use AUTO OFF to return to manual fan control." ]) self.show_cooling_detail() def auto_cooling_off_quick(self): result = cooling_manager.disable_auto_cooling(fan_off=False) self.set_smart_detail("AUTO COOLING OFF", [ result.get("message", "Auto cooling disabled."), "Manual fan controls remain available.", "Use FAN OFF if you want silence." ]) self.show_cooling_detail() def set_fan_quick(self, percent): result = cooling_manager.set_fan_percent(percent) data = cooling_manager.snapshot() temp = data.get("temp", "---") recommended = data.get("recommended_speed", 0) title = "FAN COMMAND SENT" if result.get("ok") else "FAN COMMAND FAILED" self.set_smart_detail(title, [ result.get("message", "No response."), f"CPU temp: {temp}°C", f"Recommended fan: {recommended}%", "Use FAN OFF if you want quiet mode." ]) self.set_quick_actions([ ("◎", "AUTO", "ON", self.auto_cooling_on_quick, "online"), ("◌", "AUTO", "OFF", self.auto_cooling_off_quick, "standby"), ("▲", "FAN", "MAX", lambda: self.set_fan_quick(100), "critical"), ("○", "FAN", "OFF", lambda: self.set_fan_quick(0), "standby"), ]) def start_night_forge_quick(self): try: shire_runtime.create_night_forge_task() agent_manager.start_night_forge_demo() self.set_smart_detail("NIGHT FORGE", [ "Night Forge task queued.", "Agent worker chain prepared.", "Check Task Station or Agent Manager." ]) except Exception as e: self.set_smart_detail("NIGHT FORGE ERROR", [ str(e) ]) self.set_quick_actions([ ("☑", "OPEN", "TASKS", lambda: self.open_station(8, "Task Station"), "online"), ("♟", "OPEN", "AGENTS", lambda: self.open_station(7, "Agent Manager"), "online"), ("⚒", "OPEN", "FORGE", lambda: self.open_station(2, "Forge Station"), "standby"), ("↩", "RESET", "PANEL", self.show_default_actions, "standby"), ]) def standby_agents_quick(self): agent_manager.standby_all() self.set_smart_detail("AGENTS STANDBY", [ "All agents returned to standby.", f"Agent status: {agent_manager.summary()}" ]) def handle_status_tile(self, key): direct_routes = { "heart": (1, "SHIRE AI Core"), "shire": (1, "SHIRE AI Core"), "sentinel": (10, "Sentinel Security"), "memory": (8, "Task Station"), "forge": (2, "Forge Station"), "caretaker": (6, "System Station"), "agents": (7, "Agent Manager"), "atlas": (4, "Atlas Station"), } if key in direct_routes: index, label = direct_routes[key] self.open_station(index, label) return if key == "shire": self.set_smart_detail("SHIRE AI CORE", [ "Command mode and AI services live there.", "Use SHIRE for direct commands, notes, tasks, and future voice work." ]) self.set_quick_actions([ ("◉", "OPEN", "SHIRE", lambda: self.open_station(1, "SHIRE AI Core"), "online"), ("☑", "OPEN", "TASKS", lambda: self.open_station(8, "Task Station"), "online"), ("⚙", "SYSTEM", "STATUS", lambda: self.open_station(6, "System Station"), "standby"), ("↩", "RESET", "PANEL", self.show_default_actions, "standby"), ]) return if key == "forge": self.set_smart_detail("FORGE STATION", [ "Engineering, design, and patch work live here.", "Night Forge can queue a demo worker chain.", f"Forge service: {service_manager.get_state('forge').upper()}" ]) self.set_quick_actions([ ("⚒", "OPEN", "FORGE", lambda: self.open_station(2, "Forge Station"), "online"), ("☾", "NIGHT", "FORGE", self.start_night_forge_quick, "standby"), ("♟", "OPEN", "AGENTS", lambda: self.open_station(7, "Agent Manager"), "standby"), ("☑", "OPEN", "TASKS", lambda: self.open_station(8, "Task Station"), "online"), ]) return if key == "agents": self.set_smart_detail("AGENT MANAGER", [ f"Agent status: {agent_manager.summary()}", "Agents can prepare worker chains and support Night Forge." ]) self.set_quick_actions([ ("♟", "OPEN", "AGENTS", lambda: self.open_station(7, "Agent Manager"), "online"), ("☾", "NIGHT", "FORGE", self.start_night_forge_quick, "standby"), ("◌", "STANDBY", "ALL", self.standby_agents_quick, "standby"), ("☑", "OPEN", "TASKS", lambda: self.open_station(8, "Task Station"), "online"), ]) return if key == "heart": self.set_smart_detail("HEART / OATH", [ f"Heart service: {service_manager.get_state('heart').upper()}", "Prime Directive: Protect Ray and his family.", "Blacksmith's Oath remains active." ]) self.set_quick_actions([ ("♡", "SHOW", "OATH", self.show_heart_oath, "online"), ("◉", "OPEN", "SHIRE", lambda: self.open_station(1, "SHIRE AI Core"), "online"), ("☑", "OPEN", "TASKS", lambda: self.open_station(8, "Task Station"), "online"), ("↩", "RESET", "PANEL", self.show_default_actions, "standby"), ]) return if key == "sentinel": self.set_smart_detail("SENTINEL WATCH", [ f"Sentinel service: {service_manager.get_state('sentinel').upper()}", "System watch active.", "Caretaker continues background health checks." ]) self.set_quick_actions([ ("⚙", "OPEN", "SYSTEM", lambda: self.open_station(6, "System Station"), "online"), ("⚙", "REFRESH", "CARE", self.refresh_caretaker, "online"), ("☑", "OPEN", "TASKS", lambda: self.open_station(8, "Task Station"), "online"), ("↩", "RESET", "PANEL", self.show_default_actions, "standby"), ]) return if key == "memory": self.show_memory_detail() return if key == "caretaker": self.set_smart_detail("CARETAKER SERVICE", [ f"Caretaker service: {service_manager.get_state('caretaker').upper()}", "Watching temperature, RAM, disk, and core health.", "Warnings will surface through Mission Control." ]) self.set_quick_actions([ ("❄", "COOLING", "HAT", self.show_cooling_detail, "online"), ("◎", "AUTO", "NOW", self.auto_cool_quick, "online"), ("▲", "FAN", "MAX", lambda: self.set_fan_quick(100), "critical"), ("○", "FAN", "OFF", lambda: self.set_fan_quick(0), "standby"), ]) return self.show_default_actions() def health_colour(self, state): if state == "critical": return RED if state == "watch": return AMBER return GREEN def update_health_summary(self, snapshot, caretaker_result): issues = [] watches = [] temp = snapshot.get("temp", "---") ram = snapshot.get("ram", "---") disk = snapshot.get("disk", "---") cpu = snapshot.get("cpu", "---") ip = snapshot.get("ip", "---") cooling = cooling_manager.snapshot() cooling_line = cooling.get("summary", "Cooling HAT: not detected") auto_cooling_line = cooling.get("auto_status", "Auto cooling: unknown") temp_state = caretaker_result.get("temp_state", "online") disk_state = caretaker_result.get("disk_state", "online") if temp_state == "critical": issues.append(f"CPU temperature critical: {temp}°C") elif temp_state == "warning": watches.append(f"CPU warm: {temp}°C") if disk_state == "critical": issues.append(f"Disk usage critical: {disk}%") elif disk_state == "warning": watches.append(f"Disk usage high: {disk}%") try: ram_float = float(ram) if ram_float >= 90: issues.append(f"RAM critical: {ram}%") elif ram_float >= 75: watches.append(f"RAM high: {ram}%") except Exception: watches.append("RAM reading unavailable") pending = task_manager.get_pending_count() if pending >= 10: watches.append(f"Large task queue: {pending} pending") elif pending > 0: watches.append(f"Tasks waiting: {pending}") active_agents = agent_manager.get_active_count() if active_agents > 0: watches.append(f"Agents active: {agent_manager.summary()}") service_keys = [ "heart", "shire", "sentinel", "memory", "forge", "caretaker", ] for service_key in service_keys: state = service_manager.get_state(service_key) if state in ["critical", "error", "offline", "failed"]: issues.append(f"{service_key.upper()} is {state.upper()}") elif state in ["warning", "partial"]: watches.append(f"{service_key.upper()} is {state.upper()}") if issues: overall = "CRITICAL" health_state = "critical" lines = issues[:4] elif watches: overall = "WATCH" health_state = "watch" lines = watches[:5] else: overall = "ALL CLEAR" health_state = "clear" lines = [ "Core systems stable.", "No critical warnings.", "Temperature, disk, and RAM are inside safe range." ] colour = self.health_colour(health_state) self.health_summary.setStyleSheet(f""" color:{WHITE}; font-size:11px; font-weight:bold; background-color:{PANEL_BG}; border:2px solid {colour}; border-radius:8px; padding:8px; """) body = [ f"HEALTH SUMMARY: {overall}", f"CPU: {cpu}% TEMP: {temp}°C", f"RAM: {ram}% DISK: {disk}%", f"IP: {ip}", cooling_line, auto_cooling_line, "" ] for line in lines: body.append(f"● {line}") self.health_summary.setText("\n".join(body)) def update_status(self): self.update() forge_state = service_manager.get_state("forge") shire_state = service_manager.get_state("shire") if shire_state in ["booting", "reading_heart", "reading_codex", "loading_memory", "checking_tasks", "checking_agents"]: self.core.set_state("loading") elif shire_state in ["thinking", "working", "generating"] or forge_state in ["working", "busy", "building"]: self.core.set_state("working") elif shire_state in ["error", "critical"] or forge_state in ["error", "critical", "offline"]: self.core.set_state("error") else: self.core.set_state(forge_state) self.core.update_pulse() s = get_system_snapshot() caretaker_result = caretaker.check() cooling_manager.auto_manage() self.update_health_summary(s, caretaker_result) self.cpu.set_value(f'{s["cpu"]}%', "online") self.temp.set_value(f'{s["temp"]}°C', self.temp_state(s["temp"])) self.ram.set_value(f'{s["ram"]}%', "online") self.disk.set_value(f'{s["disk"]}%', "online") self.network.set_value(s["ip"], "online") self.cards["heart"].set_value(service_manager.get_state("heart").upper(), service_manager.get_state("heart")) self.cards["shire"].set_value(service_manager.get_state("shire").upper(), service_manager.get_state("shire")) self.cards["sentinel"].set_value(service_manager.get_state("sentinel").upper(), service_manager.get_state("sentinel")) self.cards["memory"].set_value(service_manager.get_state("memory").upper(), service_manager.get_state("memory")) self.cards["forge"].set_value(service_manager.get_state("forge").upper(), service_manager.get_state("forge")) self.cards["caretaker"].set_value(service_manager.get_state("caretaker").upper(), service_manager.get_state("caretaker")) self.cards["agents"].set_value(agent_manager.summary(), agent_manager.get_state()) pending = task_manager.get_pending_count() self.objective.setText( f"CURRENT OBJECTIVE\n" f"ENG-0042\n" f"COOLING / FAN AWARENESS\n\n" f"TASKS: {pending}" ) self.update_log()