src/codex_dashboard/app.py

from __future__ import annotations

import fcntl
import logging
import os
import shutil
import subprocess
import threading
import time
from pathlib import Path

import gi

gi.require_version("Gtk", "3.0")
gi.require_version("Gdk", "3.0")
from gi.repository import Gdk, GLib, Gtk, Pango

try:
    gi.require_version("AyatanaAppIndicator3", "0.1")
    from gi.repository import AyatanaAppIndicator3
except (ImportError, ValueError):
    AyatanaAppIndicator3 = None

from .codex_data import (
    DashboardStats,
    ThreadSummary,
    format_time,
    format_tokens,
    list_logs_result,
    list_threads,
    list_threads_result,
    get_stats,
)
from .paths import CACHE_LOG, CONFIG_TOML, LOCK_PATH
from .processes import ProcessInfo, format_uptime, scan_processes, terminate_process


log = logging.getLogger("codex_dashboard")
_lock_file = None


CSS = b"""
window {
  background: #0f172a;
  color: #e2e8f0;
}
.topbar {
  background: #111827;
  border-bottom: 1px solid #1f2937;
  padding: 10px 12px;
}
.app-title {
  font-size: 18px;
  font-weight: 700;
  color: #f8fafc;
}
.muted {
  color: #94a3b8;
}
.panel {
  background: #0f172a;
}
.stat {
  background: #111827;
  border: 1px solid #1f2937;
  border-radius: 10px;
  padding: 12px;
}
.stat-value {
  font-size: 22px;
  font-weight: 700;
  color: #f8fafc;
}
.badge,
.badge-ok,
.badge-warn,
.badge-idle {
  border-radius: 999px;
  padding: 4px 10px;
  font-weight: 600;
}
.badge {
  background: #1e293b;
  color: #cbd5e1;
}
.badge-ok {
  background: #123a2b;
  color: #86efac;
}
.badge-warn {
  background: #3f2a12;
  color: #fcd34d;
}
.badge-idle {
  background: #1f2937;
  color: #cbd5e1;
}
.row {
  border: 1px solid transparent;
  border-bottom: 1px solid #1e293b;
  border-radius: 8px;
  padding: 10px 12px;
}
.row:hover {
  background: #111827;
  border-color: #334155;
}
button {
  border-radius: 8px;
}
entry {
  border-radius: 8px;
  background: #111827;
  color: #e2e8f0;
}
"""


def _setup_logging() -> None:
    CACHE_LOG.parent.mkdir(parents=True, exist_ok=True)
    logging.basicConfig(
        filename=str(CACHE_LOG),
        level=logging.INFO,
        format="%(asctime)s %(name)s %(levelname)s %(message)s",
    )


def _acquire_singleton_lock() -> None:
    global _lock_file
    fd = os.open(LOCK_PATH, os.O_CREAT | os.O_WRONLY, 0o644)
    _lock_file = os.fdopen(fd, "w")
    try:
        fcntl.flock(_lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
        _lock_file.truncate(0)
        _lock_file.write(str(os.getpid()))
        _lock_file.flush()
    except BlockingIOError:
        log.warning("CodexDashBoard already running")
        raise SystemExit(0)


def _label(text: str = "", *, xalign: float = 0.0, css: str | None = None) -> Gtk.Label:
    label = Gtk.Label(label=text, xalign=xalign)
    label.set_ellipsize(Pango.EllipsizeMode.END)
    if css:
        label.get_style_context().add_class(css)
    return label


def _button(icon_name: str, tooltip: str) -> Gtk.Button:
    button = Gtk.Button.new_from_icon_name(icon_name, Gtk.IconSize.BUTTON)
    button.set_tooltip_text(tooltip)
    return button


def _clear_list(list_box: Gtk.ListBox) -> None:
    for child in list_box.get_children():
        list_box.remove(child)


def _run_bg(work, done) -> None:
    def worker():
        try:
            result = work()
        except Exception as exc:  # pragma: no cover - UI guard
            log.exception("background task failed")
            result = exc
        GLib.idle_add(done, result)

    threading.Thread(target=worker, daemon=True).start()


def _open_path(path: str | Path) -> None:
    subprocess.Popen(["xdg-open", str(path)], start_new_session=True)


class CodexDashboard(Gtk.Window):
    def __init__(self) -> None:
        super().__init__(title="CodexDashBoard")
        self.set_default_size(940, 640)
        self.set_resizable(True)
        self.set_wmclass("codex-dashboard", "CodexDashBoard")

        self._tray_indicator = None
        self._session_query = ""
        self._log_query = ""
        self._include_archived = False

        self._stats_labels: dict[str, Gtk.Label] = {}
        self._refresh_in_flight: set[str] = set()
        self._overview_sessions = Gtk.ListBox()
        self._overview_processes = Gtk.ListBox()
        self._sessions_list = Gtk.ListBox()
        self._processes_list = Gtk.ListBox()
        self._logs_list = Gtk.ListBox()
        self._status = _label("Ready", css="muted")

        self._setup_css()
        self._build_ui()
        self._build_tray()
        self.connect("delete-event", self._on_delete)

        GLib.timeout_add_seconds(30, self._refresh_visible)
        self._refresh_visible()

    def _setup_css(self) -> None:
        provider = Gtk.CssProvider()
        provider.load_from_data(CSS)
        screen = Gdk.Screen.get_default()
        Gtk.StyleContext.add_provider_for_screen(
            screen,
            provider,
            Gtk.STYLE_PROVIDER_PRIORITY_APPLICATION,
        )

    def _build_ui(self) -> None:
        root = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
        self.add(root)

        topbar = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
        topbar.get_style_context().add_class("topbar")
        root.pack_start(topbar, False, False, 0)

        title = _label("CodexDashBoard", css="app-title")
        topbar.pack_start(title, False, False, 0)
        topbar.pack_start(self._status, True, True, 0)

        config_button = _button("emblem-system-symbolic", "Open Codex config")
        config_button.connect("clicked", lambda *_: _open_path(CONFIG_TOML))
        topbar.pack_end(config_button, False, False, 0)

        refresh_button = _button("view-refresh-symbolic", "Refresh")
        refresh_button.connect("clicked", lambda *_: self.refresh_all())
        topbar.pack_end(refresh_button, False, False, 0)

        body = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=0)
        root.pack_start(body, True, True, 0)

        self.stack = Gtk.Stack()
        self.stack.set_transition_type(Gtk.StackTransitionType.CROSSFADE)
        self.stack.set_transition_duration(120)
        self.stack.connect("notify::visible-child-name", lambda *_: self._refresh_visible())

        sidebar = Gtk.StackSidebar()
        sidebar.set_stack(self.stack)
        sidebar.set_size_request(150, -1)
        body.pack_start(sidebar, False, False, 0)
        body.pack_start(self.stack, True, True, 0)

        self.stack.add_titled(self._build_overview(), "overview", "Overview")
        self.stack.add_titled(self._build_sessions(), "sessions", "Sessions")
        self.stack.add_titled(self._build_processes(), "processes", "Processes")
        self.stack.add_titled(self._build_logs(), "logs", "Logs")

    def _build_overview(self) -> Gtk.Widget:
        scrolled = Gtk.ScrolledWindow()
        scrolled.get_style_context().add_class("panel")
        box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=14)
        box.set_margin_top(14)
        box.set_margin_bottom(14)
        box.set_margin_start(14)
        box.set_margin_end(14)
        scrolled.add(box)

        stats = Gtk.Grid(column_spacing=10, row_spacing=10)
        box.pack_start(stats, False, False, 0)
        for idx, (key, label_text) in enumerate(
            [
                ("threads", "Threads"),
                ("recent", "Updated Last Hour"),
                ("tokens", "Tokens"),
                ("processes", "Agent Processes"),
            ]
        ):
            card = self._stat_card(label_text)
            stats.attach(card, idx % 3, idx // 3, 1, 1)
            self._stats_labels[key] = card.value_label

        box.pack_start(_label("Letzte 10 Codex Sessions", css="app-title"), False, False, 0)
        box.pack_start(self._overview_sessions, False, False, 0)
        box.pack_start(_label("Active Processes", css="app-title"), False, False, 0)
        box.pack_start(self._overview_processes, False, False, 0)
        return scrolled

    def _stat_card(self, label_text: str) -> Gtk.Box:
        card = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
        card.get_style_context().add_class("stat")
        card.set_hexpand(True)
        value = _label("-", css="stat-value")
        label = _label(label_text, css="muted")
        card.pack_start(value, False, False, 0)
        card.pack_start(label, False, False, 0)
        card.value_label = value  # type: ignore[attr-defined]
        return card

    def _build_sessions(self) -> Gtk.Widget:
        box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
        box.set_margin_top(10)
        box.set_margin_bottom(10)
        box.set_margin_start(10)
        box.set_margin_end(10)

        controls = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
        search = Gtk.SearchEntry()
        search.set_placeholder_text("Search title, cwd, branch")
        search.connect("search-changed", self._on_session_search)
        controls.pack_start(search, True, True, 0)

        archived = Gtk.CheckButton(label="Archived")
        archived.connect("toggled", self._on_archived_toggled)
        controls.pack_end(archived, False, False, 0)
        box.pack_start(controls, False, False, 0)

        scrolled = Gtk.ScrolledWindow()
        scrolled.add(self._sessions_list)
        box.pack_start(scrolled, True, True, 0)
        return box

    def _build_processes(self) -> Gtk.Widget:
        box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
        box.set_margin_top(10)
        box.set_margin_bottom(10)
        box.set_margin_start(10)
        box.set_margin_end(10)
        hint = _label(
            "Codex-related processes. Stop actions ask first.",
            css="muted",
        )
        box.pack_start(hint, False, False, 0)
        scrolled = Gtk.ScrolledWindow()
        scrolled.add(self._processes_list)
        box.pack_start(scrolled, True, True, 0)
        return box

    def _build_logs(self) -> Gtk.Widget:
        box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
        box.set_margin_top(10)
        box.set_margin_bottom(10)
        box.set_margin_start(10)
        box.set_margin_end(10)
        search = Gtk.SearchEntry()
        search.set_placeholder_text("Search logs")
        search.connect("search-changed", self._on_log_search)
        box.pack_start(search, False, False, 0)
        scrolled = Gtk.ScrolledWindow()
        scrolled.add(self._logs_list)
        box.pack_start(scrolled, True, True, 0)
        return box

    def _build_tray(self) -> None:
        if AyatanaAppIndicator3 is None:
            return
        indicator = AyatanaAppIndicator3.Indicator.new(
            "codex-dashboard",
            "utilities-system-monitor",
            AyatanaAppIndicator3.IndicatorCategory.APPLICATION_STATUS,
        )
        indicator.set_status(AyatanaAppIndicator3.IndicatorStatus.ACTIVE)
        indicator.set_title("CodexDashBoard")

        menu = Gtk.Menu()
        show_item = Gtk.MenuItem(label="Show")
        show_item.connect("activate", lambda *_: self._show_window())
        menu.append(show_item)

        refresh_item = Gtk.MenuItem(label="Refresh")
        refresh_item.connect("activate", lambda *_: self.refresh_all())
        menu.append(refresh_item)

        menu.append(Gtk.SeparatorMenuItem())

        quit_item = Gtk.MenuItem(label="Quit")
        quit_item.connect("activate", lambda *_: Gtk.main_quit())
        menu.append(quit_item)

        menu.show_all()
        indicator.set_menu(menu)
        self._tray_indicator = indicator

    def _show_window(self) -> None:
        self.show_all()
        self.present()

    def _on_delete(self, *_args) -> bool:
        if self._tray_indicator is not None:
            self.hide()
            return True
        Gtk.main_quit()
        return False

    def _on_session_search(self, entry: Gtk.SearchEntry) -> None:
        self._session_query = entry.get_text()
        self.refresh_sessions()

    def _on_archived_toggled(self, button: Gtk.CheckButton) -> None:
        self._include_archived = button.get_active()
        self.refresh_sessions()

    def _on_log_search(self, entry: Gtk.SearchEntry) -> None:
        self._log_query = entry.get_text()
        self.refresh_logs()

    def _refresh_visible(self) -> bool:
        name = self.stack.get_visible_child_name()
        if name == "sessions":
            self.refresh_sessions()
        elif name == "processes":
            self.refresh_processes()
        elif name == "logs":
            self.refresh_logs()
        else:
            self.refresh_overview()
        return True

    def refresh_all(self) -> None:
        self._refresh_visible()

    def _run_refresh_once(self, key: str, work, done) -> None:
        if key in self._refresh_in_flight:
            self._set_status(f"{key.title()} refresh already running")
            return
        self._refresh_in_flight.add(key)

        def wrapped_done(result):
            try:
                return done(result)
            finally:
                self._refresh_in_flight.discard(key)

        _run_bg(work, wrapped_done)

    def refresh_overview(self) -> None:
        def work():
            return get_stats(), list_threads(limit=10), scan_processes()

        def done(result):
            if isinstance(result, Exception):
                self._set_status("Refresh failed")
                return False
            stats, threads, processes = result
            self._render_overview(stats, threads, processes)
            self._set_status("Overview refreshed")
            return False

        self._run_refresh_once("overview", work, done)

    def refresh_sessions(self) -> None:
        def work():
            return list_threads_result(
                query=self._session_query,
                include_archived=self._include_archived,
                limit=120,
            )

        def done(result):
            if isinstance(result, Exception):
                self._set_status("Session refresh failed")
                return False
            if result.error is not None:
                self._render_threads(self._sessions_list, result.data)
                self._set_status(f"Session database error: {result.error.message}")
                return False
            self._render_threads(self._sessions_list, result.data)
            self._set_status(f"{len(result.data)} sessions")
            return False

        self._run_refresh_once("sessions", work, done)

    def refresh_processes(self) -> None:
        def done(result):
            if isinstance(result, Exception):
                self._set_status("Process refresh failed")
                return False
            self._render_processes(self._processes_list, result, full=True)
            self._set_status(f"{len(result)} processes")
            return False

        self._run_refresh_once("processes", scan_processes, done)

    def refresh_logs(self) -> None:
        def work():
            return list_logs_result(query=self._log_query, limit=150)

        def done(result):
            if isinstance(result, Exception):
                self._set_status("Log refresh failed")
                return False
            if result.error is not None:
                self._render_logs(result.data)
                self._set_status(f"Log database error: {result.error.message}")
                return False
            self._render_logs(result.data)
            self._set_status(f"{len(result.data)} log rows")
            return False

        self._run_refresh_once("logs", work, done)

    def _render_overview(
        self,
        stats: DashboardStats,
        threads: list[ThreadSummary],
        processes: list[ProcessInfo],
    ) -> None:
        self._stats_labels["threads"].set_text(str(stats.thread_count))
        self._stats_labels["recent"].set_text(str(stats.recent_hour_count))
        self._stats_labels["tokens"].set_text(format_tokens(stats.total_tokens))
        self._stats_labels["processes"].set_text(str(len(processes)))
        active_cwds = {process.cwd for process in processes if process.cwd}
        self._render_threads(
            self._overview_sessions,
            threads,
            compact=True,
            active_cwds=active_cwds,
        )
        self._render_processes(self._overview_processes, processes[:6], full=False)

    def _render_threads(
        self,
        list_box: Gtk.ListBox,
        threads: list[ThreadSummary],
        *,
        compact: bool = False,
        active_cwds: set[str] | None = None,
    ) -> None:
        _clear_list(list_box)
        if not threads:
            list_box.add(self._empty_row("No Codex sessions found"))
            list_box.show_all()
            return
        for thread in threads:
            list_box.add(
                self._thread_row(thread, compact=compact, active_cwds=active_cwds or set())
            )
        list_box.show_all()

    def _thread_row(
        self,
        thread: ThreadSummary,
        *,
        compact: bool = False,
        active_cwds: set[str] | None = None,
    ) -> Gtk.ListBoxRow:
        row = Gtk.ListBoxRow()
        outer = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
        outer.get_style_context().add_class("row")
        row.add(outer)

        text_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=3)
        outer.pack_start(text_box, True, True, 0)

        title_prefix = "Thema: " if compact else ""
        title = _label(title_prefix + thread.title)
        title.set_tooltip_text(thread.id)
        text_box.pack_start(title, False, False, 0)

        state_text, state_class = self._thread_state(thread, active_cwds or set())
        meta_parts = [
            "Stand: " + state_text,
            "Update: " + format_time(thread.updated_ms),
            "Projekt: " + (Path(thread.cwd).name if thread.cwd else "no cwd"),
            thread.model or "model?",
        ]
        if thread.reasoning_effort:
            meta_parts.append(thread.reasoning_effort)
        if thread.git_branch:
            meta_parts.append(thread.git_branch)
        if not compact:
            meta_parts.append(format_tokens(thread.tokens_used) + " tokens")
        meta = _label("  |  ".join(meta_parts), css="muted")
        text_box.pack_start(meta, False, False, 0)

        badge = _label(state_text, xalign=0.5, css=state_class)
        outer.pack_end(badge, False, False, 0)

        if not compact:
            open_button = _button("folder-open-symbolic", "Open folder")
            open_button.connect("clicked", lambda *_: self._open_thread_folder(thread))
            outer.pack_end(open_button, False, False, 0)

            resume_button = _button("media-playback-start-symbolic", "Resume in terminal")
            resume_button.connect("clicked", lambda *_: self._resume_thread(thread))
            outer.pack_end(resume_button, False, False, 0)

        return row

    def _thread_state(self, thread: ThreadSummary, active_cwds: set[str]) -> tuple[str, str]:
        if thread.archived:
            return "Archived", "badge"
        if thread.cwd and thread.cwd in active_cwds:
            return "Running", "badge-ok"
        age_ms = int(time.time() * 1000) - thread.updated_ms
        if age_ms <= 10 * 60 * 1000:
            return "Active", "badge-ok"
        if age_ms <= 60 * 60 * 1000:
            return "Recent", "badge-warn"
        return "Idle", "badge-idle"

    def _render_processes(
        self,
        list_box: Gtk.ListBox,
        processes: list[ProcessInfo],
        *,
        full: bool,
    ) -> None:
        _clear_list(list_box)
        if not processes:
            list_box.add(self._empty_row("No Codex-related processes found"))
            list_box.show_all()
            return
        for process in processes:
            list_box.add(self._process_row(process, full=full))
        list_box.show_all()

    def _process_row(self, process: ProcessInfo, *, full: bool) -> Gtk.ListBoxRow:
        row = Gtk.ListBoxRow()
        outer = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8)
        outer.get_style_context().add_class("row")
        row.add(outer)

        text_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=3)
        outer.pack_start(text_box, True, True, 0)

        title = _label(f"{process.name}  pid {process.pid}")
        title.set_tooltip_text(process.command)
        text_box.pack_start(title, False, False, 0)

        cwd = Path(process.cwd).name if process.cwd else "-"
        meta = (
            f"{process.ram_mb:.1f} MB  |  {process.cpu:.1f}% CPU  |  "
            f"{format_uptime(process.uptime_seconds)}  |  {cwd}"
        )
        text_box.pack_start(_label(meta, css="muted"), False, False, 0)

        if process.stale:
            badge = _label("Stale", xalign=0.5, css="badge-warn")
            outer.pack_end(badge, False, False, 0)

        if full:
            stop_button = _button("process-stop-symbolic", "Stop process")
            stop_button.connect("clicked", lambda *_: self._confirm_stop(process))
            outer.pack_end(stop_button, False, False, 0)

        return row

    def _render_logs(self, logs) -> None:
        _clear_list(self._logs_list)
        if not logs:
            self._logs_list.add(self._empty_row("No Codex logs found"))
            self._logs_list.show_all()
            return
        for entry in logs:
            row = Gtk.ListBoxRow()
            outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=3)
            outer.get_style_context().add_class("row")
            row.add(outer)

            header = _label(
                f"{format_time(entry.ts_ms)}  |  {entry.level}  |  {entry.target}"
            )
            outer.pack_start(header, False, False, 0)
            if entry.message:
                outer.pack_start(_label(entry.message, css="muted"), False, False, 0)
            elif entry.thread_id:
                outer.pack_start(_label(entry.thread_id, css="muted"), False, False, 0)
            self._logs_list.add(row)
        self._logs_list.show_all()

    def _empty_row(self, text: str) -> Gtk.ListBoxRow:
        row = Gtk.ListBoxRow()
        label = _label(text, css="muted")
        label.set_margin_top(12)
        label.set_margin_bottom(12)
        label.set_margin_start(12)
        label.set_margin_end(12)
        row.add(label)
        return row

    def _resume_thread(self, thread: ThreadSummary) -> None:
        codex = shutil.which("codex")
        kitty = shutil.which("kitty")
        cwd = thread.cwd if thread.cwd and Path(thread.cwd).is_dir() else str(Path.home())
        if codex is None:
            self._set_status("codex command not found")
            return
        if kitty:
            cmd = [kitty, "-e", codex, "resume", thread.id]
        else:
            cmd = [codex, "resume", thread.id]
        try:
            subprocess.Popen(cmd, cwd=cwd, start_new_session=True)
            self._set_status("Resume launched")
        except OSError:
            log.exception("resume failed")
            self._set_status("Resume failed")

    def _open_thread_folder(self, thread: ThreadSummary) -> None:
        path = Path(thread.cwd)
        if path.is_dir():
            _open_path(path)
            self._set_status("Folder opened")
        else:
            self._set_status("Folder missing")

    def _confirm_stop(self, process: ProcessInfo) -> None:
        dialog = Gtk.MessageDialog(
            transient_for=self,
            flags=0,
            message_type=Gtk.MessageType.WARNING,
            buttons=Gtk.ButtonsType.CANCEL,
            text=f"Stop process {process.pid}?",
        )
        dialog.format_secondary_text(process.command)
        dialog.add_button("Stop", Gtk.ResponseType.OK)
        response = dialog.run()
        dialog.destroy()
        if response != Gtk.ResponseType.OK:
            return
        try:
            terminate_process(process.pid)
            self._set_status(f"Sent SIGTERM to {process.pid}")
            self.refresh_processes()
        except OSError:
            log.exception("terminate failed")
            self._set_status("Stop failed")

    def _set_status(self, text: str) -> None:
        self._status.set_text(text)


def main() -> int:
    _setup_logging()
    _acquire_singleton_lock()
    win = CodexDashboard()
    win.connect("destroy", Gtk.main_quit)
    win.show_all()
    Gtk.main()
    return 0