from __future__ import annotations import logging import os import queue import threading from dataclasses import dataclass from typing import Any @dataclass(frozen=True) class CompletionOverlayPayload: job_id: str objective: str return_message: str steps: int elapsed_seconds: float class DesktopOverlayManager: def __init__(self, logger: logging.Logger | None = None, *, auto_dismiss_seconds: float = 10.0) -> None: self.logger = logger or logging.getLogger("screenjob.overlay") self._queue: queue.Queue[CompletionOverlayPayload] = queue.Queue() self._thread: threading.Thread | None = None self._lock = threading.Lock() self._ready = threading.Event() self._disabled = False self._warned = False self._auto_dismiss_ms = max(0, int(round(float(auto_dismiss_seconds) * 1000))) def show_completion( self, *, job_id: str, objective: str, return_message: str, steps: int, elapsed_seconds: float, ) -> None: if os.name != "nt": self._disable_once("Desktop completion HUD is only enabled on Windows.") return if not self._ensure_thread(): return self._queue.put( CompletionOverlayPayload( job_id=job_id, objective=objective, return_message=return_message, steps=max(0, int(steps)), elapsed_seconds=max(0.0, float(elapsed_seconds)), ) ) def _ensure_thread(self) -> bool: with self._lock: if self._disabled: return False if self._thread is None or not self._thread.is_alive(): self._ready.clear() self._thread = threading.Thread(target=self._ui_main, name="screenjob-overlay", daemon=True) self._thread.start() self._ready.wait(timeout=2.0) return not self._disabled def _disable_once(self, reason: str) -> None: with self._lock: self._disabled = True already_warned = self._warned self._warned = True self._ready.set() if not already_warned: self.logger.warning("%s Overlay notifications disabled.", reason) def _format_elapsed(self, elapsed_seconds: float) -> str: total_seconds = max(0, int(round(elapsed_seconds))) minutes, seconds = divmod(total_seconds, 60) hours, minutes = divmod(minutes, 60) if hours: return f"{hours}h {minutes}m {seconds}s" if minutes: return f"{minutes}m {seconds}s" return f"{seconds}s" def _shorten(self, text: str, limit: int) -> str: raw = " ".join(str(text or "").split()) if len(raw) <= limit: return raw return raw[: max(0, limit - 1)].rstrip() + "..." def _ui_main(self) -> None: try: import tkinter as tk except Exception as exc: # noqa: BLE001 self._disable_once(f"tkinter is unavailable ({type(exc).__name__}: {exc}).") return try: root = tk.Tk() root.withdraw() root.update_idletasks() except Exception as exc: # noqa: BLE001 self._disable_once(f"Desktop overlay could not initialize ({type(exc).__name__}: {exc}).") return cards: list[dict[str, Any]] = [] self._ready.set() def reposition() -> None: screen_width = root.winfo_screenwidth() top = 24 for entry in cards: window = entry["window"] if not bool(window.winfo_exists()): continue window.update_idletasks() width = max(320, int(window.winfo_width() or 360)) height = max(120, int(window.winfo_height() or 160)) left = max(12, screen_width - width - 24) window.geometry(f"{width}x{height}+{left}+{top}") top += height + 16 def dismiss(window: Any) -> None: for index, entry in enumerate(list(cards)): if entry["window"] is window: after_id = entry.get("after_id") if after_id is not None: try: window.after_cancel(after_id) except Exception: # noqa: BLE001 pass cards.pop(index) break try: if bool(window.winfo_exists()): window.destroy() except Exception: # noqa: BLE001 pass if cards: reposition() def add_card(payload: CompletionOverlayPayload) -> None: card = tk.Toplevel(root) card.withdraw() card.overrideredirect(True) card.attributes("-topmost", True) card.configure(bg="#0f172a") frame = tk.Frame(card, bg="#0f172a", highlightthickness=1, highlightbackground="#22c55e", bd=0) frame.pack(fill="both", expand=True) close_button = tk.Button( frame, text="×", command=lambda win=card: dismiss(win), bg="#0f172a", fg="#cbd5e1", activebackground="#111827", activeforeground="#ffffff", relief="flat", borderwidth=0, font=("Segoe UI", 14, "bold"), padx=6, pady=0, ) close_button.place(relx=1.0, x=-8, y=6, anchor="ne") header = tk.Label( frame, text="Completed", bg="#0f172a", fg="#86efac", font=("Segoe UI", 10, "bold"), anchor="w", ) header.pack(fill="x", padx=14, pady=(12, 2)) title = tk.Label( frame, text=self._shorten(payload.objective, 72) or "Job complete", bg="#0f172a", fg="#f8fafc", font=("Segoe UI", 11, "bold"), justify="left", wraplength=320, anchor="w", ) title.pack(fill="x", padx=14) job_row = tk.Label( frame, text=f"Job {payload.job_id}", bg="#0f172a", fg="#94a3b8", font=("Segoe UI", 9), justify="left", anchor="w", ) job_row.pack(fill="x", padx=14, pady=(2, 8)) message = tk.Label( frame, text=self._shorten(payload.return_message, 180) or "Task completed.", bg="#0f172a", fg="#e2e8f0", font=("Segoe UI", 9), justify="left", wraplength=320, anchor="w", ) message.pack(fill="x", padx=14) footer = tk.Label( frame, text=f"{payload.steps} step(s) | {self._format_elapsed(payload.elapsed_seconds)}", bg="#0f172a", fg="#94a3b8", font=("Segoe UI", 9), justify="left", anchor="w", ) footer.pack(fill="x", padx=14, pady=(10, 12)) after_id = None if self._auto_dismiss_ms > 0: after_id = card.after(self._auto_dismiss_ms, lambda win=card: dismiss(win)) cards.insert(0, {"window": card, "after_id": after_id}) while len(cards) > 3: stale = cards.pop() try: stale_after_id = stale.get("after_id") if stale_after_id is not None: stale["window"].after_cancel(stale_after_id) stale["window"].destroy() except Exception: # noqa: BLE001 pass card.update_idletasks() reposition() card.deiconify() def pump_queue() -> None: try: while True: add_card(self._queue.get_nowait()) except queue.Empty: pass try: root.after(120, pump_queue) except Exception: # noqa: BLE001 self._disable_once("Desktop overlay event loop stopped unexpectedly.") pump_queue() try: root.mainloop() except Exception as exc: # noqa: BLE001 self._disable_once(f"Desktop overlay main loop failed ({type(exc).__name__}: {exc}).") _overlay_singleton: DesktopOverlayManager | None = None _overlay_lock = threading.Lock() def get_desktop_overlay_manager(logger: logging.Logger | None = None) -> DesktopOverlayManager: global _overlay_singleton with _overlay_lock: if _overlay_singleton is None: _overlay_singleton = DesktopOverlayManager(logger=logger) elif logger is not None: _overlay_singleton.logger = logger return _overlay_singleton