""" Win32 window-layer engine for the Togen Windows Display Agent (DVI-1230). Implements the four layer modes the server can assign to a display: top -> HWND_TOPMOST, reasserted periodically (default). bottom -> HWND_BOTTOM, reasserted periodically + on foreground-change events. windowed -> normal movable/resizable window at the given geometry. wallpaper -> WorkerW re-parent trick (Progman 0x052C message), sits behind desktop icons; EXPERIMENTAL. Any failure falls back to `bottom`. All real Win32 calls are isolated behind small, injectable functions so the policy logic (monitor resolution, wallpaper fallback state machine, layer sanitization) is unit-testable on any platform without a live Windows desktop or a running window. `ctypes.windll`/`ctypes.WinDLL` are only ever touched inside functions that run on Windows, never at import time, so this module imports cleanly on macOS/Linux dev machines too (see tests/test_winlayer.py). """ from __future__ import annotations import sys import time import logging IS_WINDOWS = sys.platform == "win32" log = logging.getLogger("display_agent.winlayer") VALID_LAYERS = ("top", "bottom", "windowed", "wallpaper") DEFAULT_LAYER = "top" # How often the periodic reassert thread re-applies top/bottom z-order. REASSERT_INTERVAL_S = 2.0 # Win32 constants (kept local — no dependency on pywin32). _HWND_TOP = 0 _HWND_BOTTOM = 1 _HWND_TOPMOST = -1 _HWND_NOTOPMOST = -2 _SWP_NOSIZE = 0x0001 _SWP_NOMOVE = 0x0002 _SWP_NOACTIVATE = 0x0010 _SWP_SHOWWINDOW = 0x0040 _GW_HWNDNEXT = 2 _SMTO_NORMAL = 0x0000 _WM_SPAWN_WORKERW = 0x052C _WS_CAPTION = 0x00C00000 _WS_THICKFRAME = 0x00040000 _WS_POPUP = 0x80000000 _GWL_STYLE = -16 _ES_CONTINUOUS = 0x80000000 _ES_SYSTEM_REQUIRED = 0x00000001 _ES_DISPLAY_REQUIRED = 0x00000002 # --------------------------------------------------------------------------- # Pure policy helpers (platform-independent, unit tested directly) # --------------------------------------------------------------------------- def sanitize_layer(value) -> str: """Coerce a server-supplied layer string to a known value, defaulting to `top` for anything unrecognized (defence in depth — the server also sanitizes, but a stale/edited config file should never crash the agent).""" v = (value or "").strip().lower() if isinstance(value, str) else "" return v if v in VALID_LAYERS else DEFAULT_LAYER def resolve_monitor(monitors: list, spec) -> "dict | None": """Pick a monitor record from an already-enumerated list. monitors: [{"index": <1-based int>, "rect": (left, top, right, bottom), "primary": bool}, ...] in stable enumeration order. spec: "primary" (or blank/None) or a 1-based index (int or numeric str). Falls back to the primary monitor, then the first monitor, then None (no display attached — callers should skip layer application). """ if not monitors: return None if spec and str(spec).strip().lower() != "primary": try: idx = int(str(spec).strip()) except (TypeError, ValueError): idx = None if idx is not None: for m in monitors: if m.get("index") == idx: return m for m in monitors: if m.get("primary"): return m return monitors[0] def sanitize_windowed_geometry(geom, monitor_rect) -> dict: """Clamp a windowed-mode geometry dict to sane bounds within the monitor. geom: {"x","y","width","height"} (any subset; missing keys default to a centered ~60% window). monitor_rect: (left, top, right, bottom). """ geom = geom if isinstance(geom, dict) else {} left, top, right, bottom = monitor_rect mon_w, mon_h = max(1, right - left), max(1, bottom - top) default_w, default_h = int(mon_w * 0.6), int(mon_h * 0.6) def _coerce(key, default, lo, hi): try: v = int(geom.get(key, default)) except (TypeError, ValueError): v = default return max(lo, min(hi, v)) width = _coerce("width", default_w, 200, mon_w) height = _coerce("height", default_h, 150, mon_h) x = _coerce("x", left + (mon_w - width) // 2, left - width + 50, right - 50) y = _coerce("y", top + (mon_h - height) // 2, top - height + 50, bottom - 50) return {"x": x, "y": y, "width": width, "height": height} # --------------------------------------------------------------------------- # Wallpaper (WorkerW) re-parent state machine — Win32 calls injected so the # fallback logic is testable without a real desktop. # --------------------------------------------------------------------------- class WallpaperReparenter: """Implements the Progman/WorkerW re-parent trick used by Wallpaper Engine/Lively. Every Win32 primitive is an injected callable; production code wires these to real user32 calls (see `_real_wallpaper_reparenter`), tests wire them to fakes to exercise the fallback paths deterministically. """ def __init__(self, find_progman, send_progman_spawn, enum_top_windows, has_shelldll_defview_child, get_next_sibling, set_parent, find_workerw_child=None): self._find_progman = find_progman self._send_progman_spawn = send_progman_spawn self._enum_top_windows = enum_top_windows self._has_shelldll_defview_child = has_shelldll_defview_child self._get_next_sibling = get_next_sibling self._set_parent = set_parent self._find_workerw_child = find_workerw_child def reparent(self, hwnd) -> "tuple[bool, str]": progman = self._find_progman() if not progman: return False, "Progman window not found" self._send_progman_spawn(progman) # Win10 shape: the spawned WorkerW is the next SIBLING of whichever # top-level window hosts SHELLDLL_DefView. workerw = None for top in self._enum_top_windows(): if self._has_shelldll_defview_child(top): candidate = self._get_next_sibling(top) if candidate: workerw = candidate break # Win11 22H2+ shape: SHELLDLL_DefView stays under Progman and the # spawned WorkerW appears as a CHILD of Progman instead (found live # on Win11 in the DVI-1227 P3 e2e). if not workerw and self._find_workerw_child is not None: workerw = self._find_workerw_child(progman) if not workerw: return False, "no WorkerW window appeared after the Progman spawn message" if not self._set_parent(hwnd, workerw): return False, "SetParent onto WorkerW failed" return True, "ok" # --------------------------------------------------------------------------- # Real Win32 bindings (Windows only; lazily bound so import stays portable) # --------------------------------------------------------------------------- def _user32(): import ctypes return ctypes.WinDLL("user32", use_last_error=True) def _kernel32(): import ctypes return ctypes.WinDLL("kernel32", use_last_error=True) def _real_enumerate_monitors() -> list: """Enumerate physical monitors via EnumDisplayMonitors. Returns [{"index": 1-based, "rect": (l,t,r,b), "primary": bool}, ...].""" import ctypes from ctypes import wintypes user32 = _user32() results = [] MONITORENUMPROC = ctypes.WINFUNCTYPE( ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p, ctypes.POINTER(wintypes.RECT), ctypes.c_double, ) class MONITORINFO(ctypes.Structure): _fields_ = [ ("cbSize", wintypes.DWORD), ("rcMonitor", wintypes.RECT), ("rcWork", wintypes.RECT), ("dwFlags", wintypes.DWORD), ] MONITORINFOF_PRIMARY = 0x1 def _cb(hmonitor, hdc, rect_ptr, data): info = MONITORINFO() info.cbSize = ctypes.sizeof(MONITORINFO) if user32.GetMonitorInfoW(hmonitor, ctypes.byref(info)): r = info.rcMonitor results.append({ "index": len(results) + 1, "rect": (r.left, r.top, r.right, r.bottom), "primary": bool(info.dwFlags & MONITORINFOF_PRIMARY), }) return 1 proc = MONITORENUMPROC(_cb) user32.EnumDisplayMonitors(None, None, proc, 0) return results def _real_wallpaper_reparenter() -> WallpaperReparenter: import ctypes from ctypes import wintypes user32 = _user32() def find_progman(): return user32.FindWindowW("Progman", None) def send_progman_spawn(progman_hwnd): result = wintypes.DWORD() user32.SendMessageTimeoutW( progman_hwnd, _WM_SPAWN_WORKERW, 0, 0, _SMTO_NORMAL, 1000, ctypes.byref(result), ) def enum_top_windows(): handles = [] WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p) def _cb(hwnd, _lparam): handles.append(hwnd) return 1 user32.EnumWindows(WNDENUMPROC(_cb), 0) return handles def has_shelldll_defview_child(hwnd): return bool(user32.FindWindowExW(hwnd, None, "SHELLDLL_DefView", None)) def get_next_sibling(hwnd): # The WorkerW that should host wallpaper windows is the sibling # immediately after the top-level window owning SHELLDLL_DefView. return user32.FindWindowExW(None, hwnd, "WorkerW", None) def set_parent(hwnd, new_parent): return bool(user32.SetParent(hwnd, new_parent)) def find_workerw_child(progman_hwnd): # Win11 22H2+: the spawned WorkerW is a child of Progman. return user32.FindWindowExW(progman_hwnd, None, "WorkerW", None) return WallpaperReparenter( find_progman, send_progman_spawn, enum_top_windows, has_shelldll_defview_child, get_next_sibling, set_parent, find_workerw_child=find_workerw_child, ) # --------------------------------------------------------------------------- # LayerEngine — public API used by display_agent.py # --------------------------------------------------------------------------- class LayerEngine: """Applies a server-assigned {layer, monitor, windowed} config to a frameless window HWND. Owns the periodic reassert thread for top/bottom and the wallpaper-fallback bookkeeping. `enumerate_monitors`/`wallpaper_reparenter`/`set_window_pos`/`set_parent`/ `get_window_style`/`set_window_style` are injectable for testing; they default to real user32 calls on Windows. """ def __init__(self, enumerate_monitors=None, wallpaper_reparenter=None, win32_ops=None): self._enumerate_monitors = enumerate_monitors or ( _real_enumerate_monitors if IS_WINDOWS else (lambda: []) ) self._wallpaper_reparenter = wallpaper_reparenter self._win32_ops = win32_ops or (_RealWin32Ops() if IS_WINDOWS else None) self.effective_layer = None self._reassert_stop = None self._reassert_thread = None # True while the window is re-parented under WorkerW (wallpaper # mode) — leaving wallpaper MUST SetParent back to the desktop or # the window stays an invisible WorkerW child and z-order/topmost # changes silently do nothing (found live in DVI-1227 P3 e2e). self._wallpapered = False # -- monitor / geometry ------------------------------------------------- def monitors(self) -> list: try: return self._enumerate_monitors() except Exception: log.exception("enumerate_monitors failed") return [] def target_rect(self, monitor_spec) -> "tuple[int, int, int, int] | None": mon = resolve_monitor(self.monitors(), monitor_spec) return mon["rect"] if mon else None # -- layer application ---------------------------------------------------- def apply(self, hwnd, cfg: dict) -> str: """Apply {layer, monitor, windowed} to hwnd. Returns the EFFECTIVE layer actually applied (may differ from the requested one when wallpaper mode falls back to bottom).""" self.stop_reassert() layer = sanitize_layer((cfg or {}).get("layer")) monitor_spec = (cfg or {}).get("monitor") or "primary" rect = self.target_rect(monitor_spec) if self._wallpapered and layer != "wallpaper": self._require_ops().set_parent(hwnd, None) self._wallpapered = False if layer == "windowed": geom = sanitize_windowed_geometry( (cfg or {}).get("windowed"), rect or (0, 0, 1920, 1080)) self._apply_windowed(hwnd, geom) self.effective_layer = "windowed" return self.effective_layer if rect is None: log.warning("No monitor available; leaving window as-is") self.effective_layer = layer return self.effective_layer if layer == "wallpaper": ok, reason = self._try_wallpaper(hwnd, rect) if ok: self._wallpapered = True self.effective_layer = "wallpaper" self.start_reassert(hwnd, rect, "wallpaper") return self.effective_layer log.warning("Wallpaper layer failed (%s); falling back to bottom", reason) layer = "bottom" self._apply_fullscreen_zorder(hwnd, rect, layer) self.effective_layer = layer self.start_reassert(hwnd, rect, layer) return self.effective_layer def _apply_windowed(self, hwnd, geom: dict): ops = self._require_ops() ops.set_window_style(hwnd, popup=False, caption=True, thickframe=True) ops.set_window_pos(hwnd, _HWND_NOTOPMOST, geom["x"], geom["y"], geom["width"], geom["height"], _SWP_SHOWWINDOW) def _apply_fullscreen_zorder(self, hwnd, rect, layer: str): ops = self._require_ops() left, top, right, bottom = rect ops.set_window_style(hwnd, popup=True, caption=False, thickframe=False) insert_after = _HWND_TOPMOST if layer == "top" else _HWND_BOTTOM ops.set_window_pos(hwnd, insert_after, left, top, right - left, bottom - top, _SWP_SHOWWINDOW) def _try_wallpaper(self, hwnd, rect) -> "tuple[bool, str]": reparenter = self._wallpaper_reparenter or ( _real_wallpaper_reparenter() if IS_WINDOWS else None ) if reparenter is None: return False, "wallpaper reparenting is only supported on Windows" ok, reason = reparenter.reparent(hwnd) if not ok: return False, reason ops = self._require_ops() left, top, right, bottom = rect ops.set_window_style(hwnd, popup=True, caption=False, thickframe=False) ops.set_window_pos(hwnd, _HWND_TOP, left, top, right - left, bottom - top, _SWP_SHOWWINDOW | _SWP_NOACTIVATE) return True, "ok" def _require_ops(self): if self._win32_ops is None: raise RuntimeError("Win32 window operations are unavailable on this platform") return self._win32_ops # -- periodic reassert (top/bottom) -------------------------------------- def start_reassert(self, hwnd, rect, layer: str): """Background thread that periodically re-pushes top/bottom z-order so another app fighting for topmost/foreground doesn't win. wallpaper mode is re-parented once; z-order doesn't apply, so it also uses this loop to detect if the WorkerW window disappeared (explorer restart) and re-reparent on the fly.""" import threading self.stop_reassert() if layer not in ("top", "bottom", "wallpaper"): return stop = threading.Event() def _loop(): while not stop.wait(REASSERT_INTERVAL_S): try: if layer == "wallpaper": ok, _ = self._try_wallpaper(hwnd, rect) if not ok: log.warning("Wallpaper reparent lost; falling back to bottom") self._apply_fullscreen_zorder(hwnd, rect, "bottom") self.effective_layer = "bottom" return else: self._apply_fullscreen_zorder(hwnd, rect, layer) except Exception: log.exception("Periodic layer reassert failed") self._reassert_stop = stop self._reassert_thread = threading.Thread( target=_loop, name="display-agent-layer-reassert", daemon=True) self._reassert_thread.start() def stop_reassert(self): if self._reassert_stop is not None: self._reassert_stop.set() self._reassert_stop = None self._reassert_thread = None class _RealWin32Ops: """Thin real-Windows implementations of the operations LayerEngine needs beyond monitor enumeration and wallpaper reparenting. SetWindowPos MUST be called with explicit argtypes: the special hwndInsertAfter values HWND_TOPMOST/-1 and HWND_NOTOPMOST/-2 are (HWND)-1/-2 — full-width pointers. ctypes' default int marshaling mangles them on 64-bit, SetWindowPos rejects the handle, and the z-order silently never changes (found live in DVI-1227 P3 e2e).""" def __init__(self): import ctypes from ctypes import wintypes self._u32 = ctypes.WinDLL("user32", use_last_error=True) self._u32.SetWindowPos.argtypes = [ wintypes.HWND, wintypes.HWND, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_int, wintypes.UINT] self._u32.SetWindowPos.restype = wintypes.BOOL self._u32.GetWindowLongW.argtypes = [wintypes.HWND, ctypes.c_int] self._u32.GetWindowLongW.restype = wintypes.LONG self._u32.SetWindowLongW.argtypes = [ wintypes.HWND, ctypes.c_int, wintypes.LONG] self._u32.SetWindowLongW.restype = wintypes.LONG self._u32.SetParent.argtypes = [wintypes.HWND, wintypes.HWND] self._u32.SetParent.restype = wintypes.HWND self._ctypes = ctypes self._wt = wintypes def set_parent(self, hwnd, new_parent): """Re-parent hwnd (None/0 restores it as a top-level desktop window).""" prev = self._u32.SetParent(hwnd, new_parent) if not prev: log.warning("SetParent(hwnd=%s, parent=%s) failed (err=%s)", hwnd, new_parent, self._ctypes.get_last_error()) def set_window_pos(self, hwnd, insert_after, x, y, w, h, flags): ok = self._u32.SetWindowPos(hwnd, self._wt.HWND(insert_after), x, y, w, h, flags) if not ok: log.warning("SetWindowPos(hwnd=%s, insert_after=%s) failed (err=%s)", hwnd, insert_after, self._ctypes.get_last_error()) def set_window_style(self, hwnd, popup: bool, caption: bool, thickframe: bool): style = self._u32.GetWindowLongW(hwnd, _GWL_STYLE) if popup: style = (style | _WS_POPUP) & ~_WS_CAPTION & ~_WS_THICKFRAME else: style &= ~_WS_POPUP style = (style | _WS_CAPTION) if caption else (style & ~_WS_CAPTION) style = (style | _WS_THICKFRAME) if thickframe else (style & ~_WS_THICKFRAME) self._u32.SetWindowLongW(hwnd, _GWL_STYLE, style) # --------------------------------------------------------------------------- # Screen-blanking prevention # --------------------------------------------------------------------------- def prevent_sleep(): """Tell Windows this session is actively displaying content, deferring both the display timeout and system sleep. Safe/cheap to call repeatedly (e.g. every couple of minutes from a keep-alive thread) — each call just resets the idle timers; it does not accumulate state.""" if not IS_WINDOWS: return try: _kernel32().SetThreadExecutionState( _ES_CONTINUOUS | _ES_SYSTEM_REQUIRED | _ES_DISPLAY_REQUIRED) except Exception: log.exception("SetThreadExecutionState failed") def allow_sleep(): """Release the ES_CONTINUOUS hold (e.g. on clean shutdown).""" if not IS_WINDOWS: return try: _kernel32().SetThreadExecutionState(_ES_CONTINUOUS) except Exception: log.exception("SetThreadExecutionState (release) failed")