#!/usr/bin/env python3 """Swarm Tab — Native GTK3 view for Agent Teams. Reads directly from ~/.claude/teams/ and ~/.claude/tasks/ — no WebKit, no Flask server needed. Provides build_swarm_tab() and refresh_swarm() with the same API as the old WebKit-based version. """ import json import threading import time from datetime import datetime, timezone from pathlib import Path import gi gi.require_version("Gtk", "3.0") gi.require_version("WebKit2", "4.1") from gi.repository import Gtk, GLib, Gio, Pango, WebKit2 from theme import get_palette from monitor import ( get_anthropic_session_cost, get_provider_costs, get_active_sessions, get_sidecar_status, ) from swarm_visual import generate_comm_graph SWARM_HTML = Path.home() / ".agent" / "diagrams" / "agent-swarm-live.html" TEAMS_DIR = Path.home() / ".claude" / "teams" TASKS_DIR = Path.home() / ".claude" / "tasks" EVENTS_DIR = Path.home() / ".claude" / "events" # Events file offset tracking (like event_tab.py) _events_file_offset: int = 0 _events_current_day: str = "" # Gio.FileMonitor ref (prevent GC) _event_file_monitor: Gio.FileMonitor | None = None # Module-level refs for refresh _webview_ready: bool = False _team_flow: Gtk.FlowBox | None = None _task_list: Gtk.ListBox | None = None _msg_list: Gtk.ListBox | None = None _team_selector: Gtk.ComboBoxText | None = None _status_label: Gtk.Label | None = None _selected_team: str | None = None _refreshing: bool = False _stack: Gtk.Stack | None = None _webview: WebKit2.WebView | None = None # --------------------------------------------------------------------------- # Data layer # --------------------------------------------------------------------------- def _load_teams() -> list[dict]: """Load all team configs from ~/.claude/teams/*/config.json.""" teams = [] if not TEAMS_DIR.exists(): return teams for team_dir in sorted(TEAMS_DIR.iterdir()): if not team_dir.is_dir(): continue config = team_dir / "config.json" if not config.exists(): continue try: data = json.loads(config.read_text(encoding="utf-8")) name = data.get("name", team_dir.name) members = data.get("members", []) # Count tasks task_dir = TASKS_DIR / team_dir.name task_count = 0 active_tasks = 0 if task_dir.exists(): for tf in task_dir.glob("*.json"): try: td = json.loads(tf.read_text(encoding="utf-8")) if td.get("metadata", {}).get("_internal"): continue task_count += 1 if td.get("status") == "in_progress": mtime = tf.stat().st_mtime if (time.time() - mtime) < 7200: # 2h active_tasks += 1 except (json.JSONDecodeError, OSError): pass teams.append({ "name": name, "dir_name": team_dir.name, "members": len(members), "member_names": [m.get("name", "?") for m in members], "tasks": task_count, "active_tasks": active_tasks, "active": active_tasks > 0, "created": data.get("createdAt", 0), }) except (json.JSONDecodeError, OSError): continue return teams def _load_tasks(team_name: str) -> list[dict]: """Load tasks for a specific team.""" tasks = [] task_dir = TASKS_DIR / team_name if not task_dir.exists(): return tasks for tf in sorted(task_dir.glob("*.json"), key=lambda f: f.stem.zfill(10)): try: data = json.loads(tf.read_text(encoding="utf-8")) # Skip internal task-assignment tasks if data.get("metadata", {}).get("_internal"): continue tasks.append({ "id": data.get("id", tf.stem), "subject": data.get("subject", "(kein Titel)"), "description": data.get("description", "")[:120], "status": data.get("status", "pending"), "owner": data.get("owner", "(unassigned)"), }) except (json.JSONDecodeError, OSError): continue # Sort: in_progress first, then pending, then completed order = {"in_progress": 0, "pending": 1, "completed": 2} tasks.sort(key=lambda t: order.get(t["status"], 9)) return tasks def _collect_comm_data(team_name: str) -> dict: """Collect per-member status and new messages for live updates.""" team_dir = TEAMS_DIR / team_name config_path = team_dir / "config.json" if not config_path.exists(): return {"members": [], "tasks": []} config = json.loads(config_path.read_text(encoding="utf-8")) members = [] for m in config.get("members", []): name = m.get("name", "?") inbox = team_dir / "inboxes" / f"{name}.json" msg_count = 0 last_msg_time = "" if inbox.exists(): try: msgs = json.loads(inbox.read_text(encoding="utf-8")) if isinstance(msgs, list): # Filter internal messages real = [x for x in msgs if not any( k in x.get("text", "") for k in ['"type":"task_assignment"', '"type":"shutdown_request"', '"type":"idle_notification"'])] msg_count = len(real) if real: last_msg_time = real[-1].get("timestamp", "") except (json.JSONDecodeError, OSError): pass is_recent = False if last_msg_time: try: ts = datetime.fromisoformat(last_msg_time.replace("Z", "+00:00")) is_recent = (datetime.now(ts.tzinfo) - ts).total_seconds() < 120 except (ValueError, AttributeError): pass members.append({ "name": name, "status": "working" if is_recent else "idle", "msg_count": msg_count, "last_msg": last_msg_time, }) tasks = [] for t in _load_tasks(team_name): tasks.append({ "id": t["id"], "subject": t["subject"], "status": t["status"], "owner": t.get("owner", ""), }) # Collect new messages for the log all_msgs = [] inbox_dir = team_dir / "inboxes" if inbox_dir.exists(): for mf in inbox_dir.glob("*.json"): try: data = json.loads(mf.read_text(encoding="utf-8")) if not isinstance(data, list): continue recipient = mf.stem for msg in data: text = msg.get("text", "") if any(k in text for k in ['"type":"task_assignment"', '"type":"shutdown_request"', '"type":"idle_notification"']): continue ts_str = msg.get("timestamp", "") try: ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) time_fmt = ts.strftime("%H:%M") except (ValueError, AttributeError): time_fmt = "??:??" all_msgs.append({ "time": time_fmt, "sender": msg.get("from", "?"), "recipient": recipient, "text": text[:90].replace("\n", " "), "timestamp": ts_str, }) except (json.JSONDecodeError, OSError): continue all_msgs.sort(key=lambda m: m.get("timestamp", ""), reverse=True) return {"members": members, "tasks": tasks, "new_messages": all_msgs[:5], "events": []} def _load_messages(team_name: str, limit: int = 10) -> list[dict]: """Load recent messages from team inboxes.""" messages = [] inbox_dir = TEAMS_DIR / team_name / "inboxes" if not inbox_dir.exists(): return messages for mf in inbox_dir.glob("*.json"): try: data = json.loads(mf.read_text(encoding="utf-8")) if not isinstance(data, list): continue recipient = mf.stem for msg in data: # Skip internal task assignments and shutdown requests text = msg.get("text", "") if '"type":"task_assignment"' in text or '"type":"shutdown_request"' in text: continue ts_str = msg.get("timestamp", "") try: ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00")) time_fmt = ts.strftime("%H:%M") sort_key = ts.timestamp() except (ValueError, AttributeError): time_fmt = "??:??" sort_key = 0 sender = msg.get("from", "?") # Truncate text for preview preview = text[:100].replace("\n", " ") if len(text) > 100: preview += "..." messages.append({ "time": time_fmt, "sort_key": sort_key, "sender": sender, "recipient": recipient, "preview": preview, }) except (json.JSONDecodeError, OSError): continue messages.sort(key=lambda m: m["sort_key"], reverse=True) return messages[:limit] # --------------------------------------------------------------------------- # Event feed — reads hook events from ~/.claude/events/YYYY-MM-DD.jsonl # --------------------------------------------------------------------------- _INTERESTING_EVENTS = { "PostToolUse", "PreToolUse", "UserPromptSubmit", "SubagentStart", "SubagentStop", "SubagentComplete", } def _read_recent_events(limit: int = 20) -> list[dict]: """Read recent interesting events from today's event log. Uses file offset tracking to only read new data (like event_tab.py). """ global _events_file_offset, _events_current_day today = datetime.now().strftime("%Y-%m-%d") event_file = EVENTS_DIR / f"{today}.jsonl" if not event_file.exists(): return [] # Day rollover — seek near end instead of reading from byte 0 if today != _events_current_day: file_size = event_file.stat().st_size _events_file_offset = max(0, file_size - 50_000) _events_current_day = today new_lines = [] try: with open(event_file, "r", encoding="utf-8") as f: f.seek(_events_file_offset) new_lines = f.readlines() _events_file_offset = f.tell() except OSError: pass # Cap at last 100 lines to avoid processing stale history on day rollover if len(new_lines) > 100: new_lines = new_lines[-100:] events = [] for line in new_lines: line = line.strip() if not line: continue try: ev = json.loads(line) et = ev.get("event_type") or ev.get("hook_event_name", "") if et in _INTERESTING_EVENTS: events.append({ "type": et, "tool": ev.get("tool_name", ""), "time": ev.get("logged_iso", ""), "session": ev.get("session_id", "")[:8], "agent": ev.get("tool_input", {}).get("agentName", ""), }) except json.JSONDecodeError: pass # Return only the last `limit` events return events[-limit:] def _update_events_ui(events: list[dict]) -> bool: """Push events to the WebKit view on the GTK main thread. idle_add callback.""" if not (_stack and _stack.get_visible_child_name() == "visual" and _webview and _webview_ready): return False js = f"if(typeof updateEvents==='function')updateEvents({json.dumps(events, ensure_ascii=True)})" _webview.run_javascript(js, None, None, None) return False # one-shot: do not repeat def _read_events_async() -> None: """Read events in background thread, update UI via idle_add.""" def _worker(): events = _read_recent_events(20) if events: GLib.idle_add(_update_events_ui, events) thread = threading.Thread(target=_worker, daemon=True) thread.start() # --------------------------------------------------------------------------- # Live data collection for WebKit visual view # --------------------------------------------------------------------------- def _collect_live_data() -> dict: """Collect live data from monitor.py + local team/task files. Returns a dict matching the JSON structure that updateSwarm() expects. """ cost_data = get_anthropic_session_cost() provider_costs = get_provider_costs() active_sessions = get_active_sessions() sidecar = get_sidecar_status() teams = _load_teams() # Cost bars (percentage relative to reasonable daily max) OPUS_MAX = 5.0 SONNET_MAX = 10.0 HAIKU_MAX = 2.0 GEMINI_MAX = 1.0 models = cost_data.get("models", {}) opus_cost = models.get("opus", {}).get("cost_usd", 0.0) sonnet_cost = models.get("sonnet", {}).get("cost_usd", 0.0) haiku_cost = models.get("haiku", {}).get("cost_usd", 0.0) gemini_cost = provider_costs.get("gemini", 0.0) # Node statuses active_task_count = sum(t["active_tasks"] for t in teams) sidecar_firing = any( d.get("active") for d in sidecar.get("detectors", {}).values() ) # Build tasks list from all teams all_tasks = [] for team in teams: for task in _load_tasks(team["dir_name"]): task["team"] = team["dir_name"] all_tasks.append(task) return { "agents": { "opus": { "status": "working" if any( s.get("model", "").startswith("claude-opus") for s in active_sessions ) else "idle", "cost_usd": opus_cost, "cost_pct": min(100, round(opus_cost / OPUS_MAX * 100, 1)), }, "sonnet": { "status": "working" if any( s.get("model", "").startswith("claude-sonnet") for s in active_sessions ) else "idle", "cost_usd": sonnet_cost, "cost_pct": min(100, round(sonnet_cost / SONNET_MAX * 100, 1)), }, "haiku": { "status": "working" if any( s.get("model", "").startswith("claude-haiku") for s in active_sessions ) else "idle", "cost_usd": haiku_cost, "cost_pct": min(100, round(haiku_cost / HAIKU_MAX * 100, 1)), }, }, "gemini": { "status": "working" if gemini_cost > 0 else "idle", "cost_usd": gemini_cost, "cost_pct": min(100, round(gemini_cost / GEMINI_MAX * 100, 1)), }, "sidecar": { "status": "working" if sidecar_firing else "idle", "overall_severity": sidecar.get("overall_severity", "none"), "active_detectors": [ name for name, d in sidecar.get("detectors", {}).items() if d.get("active") ], "hook_events": sidecar.get("hook_events", 0), }, "teams": [ { "name": t["name"], "dir_name": t["dir_name"], "members": t["member_names"], "status": "working" if t["active_tasks"] > 0 else "idle", "tasks_total": t["tasks"], "tasks_active": t["active_tasks"], } for t in teams ], "tasks": [ { "id": task["id"], "subject": task["subject"], "status": task["status"], "owner": task.get("owner", ""), "team": task.get("team", ""), } for task in all_tasks ], "total_cost_usd": cost_data.get("cost_usd", 0.0), "active_agent_count": len(active_sessions), "events": [], } # --------------------------------------------------------------------------- # UI builders # --------------------------------------------------------------------------- def _build_team_card(team: dict) -> Gtk.Box: """Build a single team card widget.""" card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4) card.get_style_context().add_class("sc-card") card.set_size_request(180, -1) # Team name name_label = Gtk.Label(label=team["name"]) name_label.set_halign(Gtk.Align.START) name_label.get_style_context().add_class("session-project") card.pack_start(name_label, False, False, 0) # Stats line stats = f"{team['members']} Agents | {team['tasks']} Tasks" stats_label = Gtk.Label(label=stats) stats_label.set_halign(Gtk.Align.START) stats_label.get_style_context().add_class("session-meta") card.pack_start(stats_label, False, False, 0) # Active indicator if team["active"]: dot = Gtk.Label(label=f"\u25cf Aktiv ({team['active_tasks']} laufen)") dot.get_style_context().add_class("swarm-status-connected") else: dot = Gtk.Label(label="\u25cb Idle") dot.get_style_context().add_class("swarm-status-disconnected") dot.set_halign(Gtk.Align.START) card.pack_start(dot, False, False, 0) return card def _build_task_row(task: dict) -> Gtk.ListBoxRow: """Build a single task row.""" row = Gtk.ListBoxRow() row.get_style_context().add_class("swarm-task-row") hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) hbox.set_margin_top(4) hbox.set_margin_bottom(4) hbox.set_margin_start(8) hbox.set_margin_end(8) row.add(hbox) # Status icon status = task["status"] if status == "in_progress": icon_text = "\u25cf" badge_class = "swarm-badge-progress" elif status == "completed": icon_text = "\u2713" badge_class = "swarm-badge-completed" else: icon_text = "\u25cb" badge_class = "swarm-badge-pending" icon = Gtk.Label(label=icon_text) icon.get_style_context().add_class(badge_class) icon.set_size_request(20, -1) hbox.pack_start(icon, False, False, 0) # Task subject subject = Gtk.Label(label=task["subject"]) subject.set_halign(Gtk.Align.START) subject.set_ellipsize(Pango.EllipsizeMode.END) subject.set_max_width_chars(50) subject.set_hexpand(True) subject.get_style_context().add_class("session-preview") hbox.pack_start(subject, True, True, 0) # Status badge text status_text = status.upper().replace("_", " ") badge = Gtk.Label(label=status_text) badge.get_style_context().add_class(badge_class) badge.set_halign(Gtk.Align.END) hbox.pack_start(badge, False, False, 0) row.show_all() return row def _build_message_row(msg: dict) -> Gtk.ListBoxRow: """Build a single message row.""" row = Gtk.ListBoxRow() row.get_style_context().add_class("swarm-message") hbox = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) hbox.set_margin_top(2) hbox.set_margin_bottom(2) hbox.set_margin_start(8) hbox.set_margin_end(8) row.add(hbox) # Timestamp time_label = Gtk.Label(label=msg["time"]) time_label.get_style_context().add_class("log-time") time_label.set_size_request(40, -1) hbox.pack_start(time_label, False, False, 0) # Sender -> Recipient route = Gtk.Label(label=f"{msg['sender']} \u2192 {msg['recipient']}") route.get_style_context().add_class("session-meta") route.set_size_request(160, -1) route.set_halign(Gtk.Align.START) route.set_ellipsize(Pango.EllipsizeMode.END) hbox.pack_start(route, False, False, 0) # Preview preview = Gtk.Label(label=msg["preview"]) preview.set_halign(Gtk.Align.START) preview.set_ellipsize(Pango.EllipsizeMode.END) preview.set_hexpand(True) preview.get_style_context().add_class("session-preview") hbox.pack_start(preview, True, True, 0) row.show_all() return row # --------------------------------------------------------------------------- # Refresh logic # --------------------------------------------------------------------------- def _refresh_content() -> None: """Refresh all swarm content based on selected team.""" global _selected_team, _refreshing if _refreshing: return _refreshing = True try: _refresh_content_inner() finally: _refreshing = False def _refresh_content_inner() -> None: """Actual refresh logic — guarded by _refreshing flag.""" global _selected_team teams = _load_teams() # Update status if _status_label: total = len(teams) active = sum(1 for t in teams if t["active"]) _status_label.set_text(f"{total} Teams, {active} aktiv") # Update team cards if _team_flow: for child in _team_flow.get_children(): _team_flow.remove(child) for team in teams: card = _build_team_card(team) flow_child = Gtk.FlowBoxChild() flow_child.add(card) flow_child.show_all() _team_flow.add(flow_child) _team_flow.show_all() # Update selector if _team_selector: current = _team_selector.get_active_text() _team_selector.remove_all() _team_selector.append_text("Alle Teams") for team in teams: _team_selector.append_text(team["dir_name"]) # Restore selection if current: model = _team_selector.get_model() for i, row in enumerate(model): if row[0] == current: _team_selector.set_active(i) break else: _team_selector.set_active(0) else: _team_selector.set_active(0) active_text = _team_selector.get_active_text() if _team_selector else None if active_text and active_text != "Alle Teams": _selected_team = active_text else: _selected_team = None # Update tasks if _task_list: for child in _task_list.get_children(): _task_list.remove(child) if _selected_team: tasks = _load_tasks(_selected_team) else: # All teams tasks = [] for team in teams: for t in _load_tasks(team["dir_name"]): t["_team"] = team["dir_name"] tasks.append(t) order = {"in_progress": 0, "pending": 1, "completed": 2} tasks.sort(key=lambda t: order.get(t["status"], 9)) for task in tasks: _task_list.add(_build_task_row(task)) _task_list.show_all() # Update messages if _msg_list: for child in _msg_list.get_children(): _msg_list.remove(child) if _selected_team: messages = _load_messages(_selected_team) else: messages = [] for team in teams: messages.extend(_load_messages(team["dir_name"], limit=5)) messages.sort(key=lambda m: m["sort_key"], reverse=True) messages = messages[:10] for msg in messages: _msg_list.add(_build_message_row(msg)) _msg_list.show_all() # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def _update_visual_for_team() -> None: """Load comm graph HTML for the selected team, or fallback HTML for all.""" if not _stack or _stack.get_visible_child_name() != "visual" or not _webview: return active_text = _team_selector.get_active_text() if _team_selector else None if active_text and active_text != "Alle Teams": html = generate_comm_graph(active_text) _webview.load_html(html, None) elif SWARM_HTML.exists(): _webview.load_uri(SWARM_HTML.as_uri()) def _on_load_changed(webview, event): """Track when WebKit page has finished loading.""" global _webview_ready if event == WebKit2.LoadEvent.FINISHED: _webview_ready = True def _build_visual_view() -> Gtk.Widget: """Build the WebKit visual view loading agent-swarm-live.html.""" global _webview _webview = WebKit2.WebView() settings = _webview.get_settings() settings.set_enable_javascript(True) settings.set_enable_developer_extras(False) settings.set_hardware_acceleration_policy( WebKit2.HardwareAccelerationPolicy.NEVER ) _webview.connect("load-changed", _on_load_changed) if SWARM_HTML.exists(): _webview.load_uri(SWARM_HTML.as_uri()) else: p = get_palette() _webview.load_html( f'' f'
agent-swarm-live.html nicht gefunden
' f'{SWARM_HTML}
' '', None, ) return _webview def build_swarm_tab() -> Gtk.Box: """Build and return the Swarm tab widget.""" global _team_flow, _task_list, _msg_list, _team_selector, _status_label, _stack box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0) # --- Toolbar --- toolbar = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) toolbar.set_margin_start(8) toolbar.set_margin_end(8) toolbar.set_margin_top(6) toolbar.set_margin_bottom(4) toolbar.get_style_context().add_class("swarm-toolbar") title = Gtk.Label(label="Swarm") title.get_style_context().add_class("section-title") title_attrs = Pango.AttrList() title_attrs.insert(Pango.attr_weight_new(Pango.Weight.BOLD)) title_attrs.insert(Pango.attr_scale_new(1.1)) title.set_attributes(title_attrs) toolbar.pack_start(title, False, False, 0) # View toggle: Daten / Visual view_toggle = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0) view_toggle.get_style_context().add_class("linked") btn_data = Gtk.ToggleButton(label="Daten") btn_visual = Gtk.ToggleButton(label="Visual") btn_data.set_active(True) btn_data.get_style_context().add_class("swarm-btn") btn_visual.get_style_context().add_class("swarm-btn") def _on_toggle_data(btn): if btn.get_active(): btn_visual.set_active(False) if _stack: _stack.set_visible_child_name("data") elif not btn_visual.get_active(): btn.set_active(True) def _on_toggle_visual(btn): if btn.get_active(): btn_data.set_active(False) if _stack: _stack.set_visible_child_name("visual") _update_visual_for_team() elif not btn_data.get_active(): btn.set_active(True) btn_data.connect("toggled", _on_toggle_data) btn_visual.connect("toggled", _on_toggle_visual) view_toggle.pack_start(btn_data, False, False, 0) view_toggle.pack_start(btn_visual, False, False, 0) toolbar.pack_start(view_toggle, False, False, 4) # Team selector _team_selector = Gtk.ComboBoxText() _team_selector.append_text("Alle Teams") _team_selector.set_active(0) def _on_team_changed(_selector): _refresh_content() _update_visual_for_team() _team_selector.connect("changed", _on_team_changed) toolbar.pack_start(_team_selector, False, False, 4) # Spacer spacer = Gtk.Box() spacer.set_hexpand(True) toolbar.pack_start(spacer, True, True, 0) _status_label = Gtk.Label(label="") _status_label.get_style_context().add_class("session-stats") toolbar.pack_end(_status_label, False, False, 4) reload_btn = Gtk.Button(label="\u27f3 Refresh") reload_btn.get_style_context().add_class("swarm-btn") def _on_refresh(_): if _stack and _stack.get_visible_child_name() == "visual" and _webview: _webview.reload() else: _refresh_content() reload_btn.connect("clicked", _on_refresh) toolbar.pack_end(reload_btn, False, False, 0) box.pack_start(toolbar, False, False, 0) box.pack_start(Gtk.Separator(), False, False, 0) # --- Stack: Data view + Visual view --- _stack = Gtk.Stack() _stack.set_transition_type(Gtk.StackTransitionType.CROSSFADE) _stack.set_transition_duration(200) # Data view (native GTK) scrolled = Gtk.ScrolledWindow() scrolled.set_policy(Gtk.PolicyType.NEVER, Gtk.PolicyType.AUTOMATIC) content = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8) content.set_margin_top(8) content.set_margin_bottom(8) content.set_margin_start(8) content.set_margin_end(8) # --- Team overview (FlowBox) --- team_label = Gtk.Label(label="Teams") team_label.set_halign(Gtk.Align.START) team_label.get_style_context().add_class("section-title") content.pack_start(team_label, False, False, 0) _team_flow = Gtk.FlowBox() _team_flow.set_min_children_per_line(1) _team_flow.set_max_children_per_line(5) _team_flow.set_selection_mode(Gtk.SelectionMode.NONE) _team_flow.set_homogeneous(True) content.pack_start(_team_flow, False, False, 0) # --- Tasks --- task_label = Gtk.Label(label="Tasks") task_label.set_halign(Gtk.Align.START) task_label.set_margin_top(8) task_label.get_style_context().add_class("section-title") content.pack_start(task_label, False, False, 0) _task_list = Gtk.ListBox() _task_list.set_selection_mode(Gtk.SelectionMode.NONE) content.pack_start(_task_list, False, False, 0) # --- Messages --- msg_label = Gtk.Label(label="Messages (letzte 10)") msg_label.set_halign(Gtk.Align.START) msg_label.set_margin_top(8) msg_label.get_style_context().add_class("section-title") content.pack_start(msg_label, False, False, 0) _msg_list = Gtk.ListBox() _msg_list.set_selection_mode(Gtk.SelectionMode.NONE) content.pack_start(_msg_list, False, False, 0) scrolled.add(content) _stack.add_named(scrolled, "data") # Visual view (WebKit) _stack.add_named(_build_visual_view(), "visual") box.pack_start(_stack, True, True, 0) # Initial load + event file monitor GLib.idle_add(_refresh_content) GLib.idle_add(_setup_event_monitor) return box def _push_events_to_webview() -> None: """Read new events and push them to the WebKit visual view via JS.""" if not (_stack and _stack.get_visible_child_name() == "visual" and _webview and _webview_ready): return events = _read_recent_events(20) if not events: return js = f"if(typeof updateEvents==='function')updateEvents({json.dumps(events, ensure_ascii=True)})" _webview.run_javascript(js, None, None, None) def _on_events_file_changed(_monitor, file, _other, event_type) -> None: """Gio.FileMonitor callback — fires on every append to the JSONL file.""" if event_type not in (Gio.FileMonitorEvent.CHANGED, Gio.FileMonitorEvent.CHANGES_DONE_HINT): return # Schedule on GLib main loop (file monitor callback may be on any thread) GLib.idle_add(_push_events_to_webview) def _setup_event_monitor() -> None: """Set up Gio.FileMonitor on today's event JSONL file. Called once from build_swarm_tab(). Re-called on day rollover from refresh_swarm(). """ global _event_file_monitor, _events_current_day, _events_file_offset today = datetime.now().strftime("%Y-%m-%d") if _events_current_day == today and _event_file_monitor is not None: return # already monitoring today's file # Cancel previous monitor if _event_file_monitor is not None: _event_file_monitor.cancel() _event_file_monitor = None _events_current_day = today _events_file_offset = 0 event_file = EVENTS_DIR / f"{today}.jsonl" if not event_file.exists(): # File doesn't exist yet — check again on next timer tick return try: gfile = Gio.File.new_for_path(str(event_file)) _event_file_monitor = gfile.monitor_file(Gio.FileMonitorFlags.NONE, None) _event_file_monitor.connect("changed", _on_events_file_changed) except GLib.Error: pass def refresh_swarm() -> bool: """Called by GLib timer from panel.py every 30s. Handles team/task data refresh + day rollover for event monitor. Event updates are pushed instantly via Gio.FileMonitor (no polling). """ # Skip if swarm tab is not visible (MiniMax M1 fix) if _stack and not _stack.get_mapped(): return True # Re-check event monitor (handles day rollover + late file creation) _setup_event_monitor() try: _refresh_content() # Async event read — offloaded to background thread _read_events_async() # Also update visual view if visible and ready if ( _stack and _stack.get_visible_child_name() == "visual" and _webview and _webview_ready ): active_text = _team_selector.get_active_text() if _team_selector else None if active_text and active_text != "Alle Teams": # Comm graph: inject updateGraph() with live data data = _collect_comm_data(active_text) js = f"if(typeof updateGraph==='function')updateGraph({json.dumps(data, ensure_ascii=True)})" _webview.run_javascript(js, None, None, None) else: # Generic swarm view: inject updateSwarm() data = _collect_live_data() js = f"if(typeof updateSwarm==='function')updateSwarm({json.dumps(data, ensure_ascii=True)})" _webview.run_javascript(js, None, None, None) except Exception: pass return True # keep timer alive