diff --git a/solstone/think/display_powersave.py b/solstone/think/display_powersave.py new file mode 100644 index 000000000..47d5a366d --- /dev/null +++ b/solstone/think/display_powersave.py @@ -0,0 +1,233 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Display powersave detection and debounced drain-gate monitoring.""" + +from __future__ import annotations + +import ctypes +import logging +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Callable + +from solstone.think.processing import ( + DISPLAY_POWERSAVE_UNAVAILABLE, + DisplayPowersaveReading, +) + +logger = logging.getLogger(__name__) + +DISPLAY_ASLEEP = "asleep" +DISPLAY_AWAKE = "awake" +DISPLAY_UNDETECTABLE = "undetectable" + +REASON_HEADLESS = "headless" +REASON_UNREADABLE = "unreadable" +REASON_UNSUPPORTED_PLATFORM = "unsupported_platform" +REASON_NO_WINDOW_SERVER = "no_window_server" + +DISPLAY_DEBOUNCE_S = 120.0 +DRM_SYSFS_ROOT = Path("/sys/class/drm") + +_OFF_DPMS = frozenset({"Off", "Standby", "Suspend"}) +_MACOS_DISPLAY_OFF_MARKER = "Display is turned off" +_MACOS_DISPLAY_ON_MARKER = "Display is turned on" +_PMSET_TIMEOUT_S = 5.0 +_CORE_GRAPHICS_PATH = "/System/Library/Frameworks/CoreGraphics.framework/CoreGraphics" + + +@dataclass(frozen=True) +class DisplayReading: + state: str + reason: str | None + + +def _asleep() -> DisplayReading: + return DisplayReading(DISPLAY_ASLEEP, None) + + +def _awake() -> DisplayReading: + return DisplayReading(DISPLAY_AWAKE, None) + + +def _undetectable(reason: str) -> DisplayReading: + return DisplayReading(DISPLAY_UNDETECTABLE, reason) + + +def read_display_power( + *, + platform: str = sys.platform, + sysfs_root: Path = DRM_SYSFS_ROOT, +) -> DisplayReading: + """Return the current display power state, or an undetectable reason.""" + try: + if platform.startswith("linux"): + return _read_linux(sysfs_root) + if platform == "darwin": + return _read_macos() + return _undetectable(REASON_UNSUPPORTED_PLATFORM) + except Exception: + logger.debug("display powersave detection failed", exc_info=True) + return _undetectable(REASON_UNREADABLE) + + +def _read_linux(sysfs_root: Path) -> DisplayReading: + try: + connectors = list(sysfs_root.glob("card*-*")) + except OSError: + return _undetectable(REASON_UNREADABLE) + if not connectors: + return _undetectable(REASON_UNREADABLE) + + considered = 0 + any_unreadable = False + for connector in connectors: + status = _read_attr(connector, "status") + enabled = _read_attr(connector, "enabled") + if status != "connected" or enabled != "enabled": + continue + dpms = _read_attr(connector, "dpms") + considered += 1 + if dpms == "On": + return _awake() + if dpms not in _OFF_DPMS: + any_unreadable = True + + if considered == 0: + return _undetectable(REASON_HEADLESS) + if any_unreadable: + return _undetectable(REASON_UNREADABLE) + return _asleep() + + +def _read_attr(path: Path, name: str) -> str | None: + try: + return (path / name).read_text(encoding="utf-8").strip() + except OSError: + return None + + +def _core_graphics(): + return ctypes.CDLL(_CORE_GRAPHICS_PATH) + + +def _macos_main_display_id() -> int: + core_graphics = _core_graphics() + core_graphics.CGMainDisplayID.restype = ctypes.c_uint32 + return int(core_graphics.CGMainDisplayID()) + + +def _macos_display_is_asleep(display_id: int) -> bool: + core_graphics = _core_graphics() + core_graphics.CGDisplayIsAsleep.argtypes = [ctypes.c_uint32] + core_graphics.CGDisplayIsAsleep.restype = ctypes.c_bool + return bool(core_graphics.CGDisplayIsAsleep(display_id)) + + +def _macos_pmset_log() -> str: + try: + completed = subprocess.run( + ["pmset", "-g", "log"], + capture_output=True, + text=True, + timeout=_PMSET_TIMEOUT_S, + check=False, + ) + except (subprocess.TimeoutExpired, OSError): + return "" + return completed.stdout + + +def _read_macos() -> DisplayReading: + main = _macos_main_display_id() + if main == 0: + return _parse_macos_log(_macos_pmset_log()) + if _macos_display_is_asleep(main): + return _asleep() + return _awake() + + +def _parse_macos_log(text: str) -> DisplayReading: + last: str | None = None + for line in text.splitlines(): + if _MACOS_DISPLAY_OFF_MARKER in line: + last = DISPLAY_ASLEEP + elif _MACOS_DISPLAY_ON_MARKER in line: + last = DISPLAY_AWAKE + if last == DISPLAY_ASLEEP: + return _asleep() + if last == DISPLAY_AWAKE: + return _awake() + return _undetectable(REASON_NO_WINDOW_SERVER) + + +class DisplayPowersaveMonitor: + def __init__(self, *, debounce_s: float = DISPLAY_DEBOUNCE_S): + self._debounce_s = debounce_s + self._snapshot = DISPLAY_POWERSAVE_UNAVAILABLE + self._asleep_since: float | None = None + self._ever_detected = False + + def poll( + self, + *, + now: float, + read: Callable[[], DisplayReading] = read_display_power, + ) -> DisplayPowersaveReading: + reading = read() + if reading.state == DISPLAY_ASLEEP: + if self._asleep_since is None: + self._asleep_since = now + snapshot = DisplayPowersaveReading( + available=True, + asleep=True, + debounced=(now - self._asleep_since) >= self._debounce_s, + ) + elif reading.state == DISPLAY_AWAKE: + self._asleep_since = None + snapshot = DisplayPowersaveReading( + available=True, + asleep=False, + debounced=False, + ) + else: + self._asleep_since = None + snapshot = DISPLAY_POWERSAVE_UNAVAILABLE + + if snapshot.available: + self._ever_detected = True + self._snapshot = snapshot + return snapshot + + def last(self) -> DisplayPowersaveReading: + return self._snapshot + + def capability_known(self) -> bool: + return self._ever_detected + + def reset(self) -> None: + self._snapshot = DISPLAY_POWERSAVE_UNAVAILABLE + self._asleep_since = None + self._ever_detected = False + + +_MONITOR = DisplayPowersaveMonitor() + + +def poll_display_powersave(now: float) -> DisplayPowersaveReading: + return _MONITOR.poll(now=now) + + +def last_display_powersave() -> DisplayPowersaveReading: + return _MONITOR.last() + + +def display_powersave_detectable() -> bool: + return _MONITOR.capability_known() + + +def reset_display_powersave_monitor() -> None: + _MONITOR.reset() diff --git a/solstone/think/processing.py b/solstone/think/processing.py index cdc23dedd..c6d5d2dcf 100644 --- a/solstone/think/processing.py +++ b/solstone/think/processing.py @@ -68,6 +68,16 @@ class ProcessingSettings: } +@dataclass(frozen=True) +class DisplayPowersaveReading: + available: bool + asleep: bool + debounced: bool + + +DISPLAY_POWERSAVE_UNAVAILABLE = DisplayPowersaveReading(False, False, False) + + @dataclass(frozen=True) class ConditionState: enabled: bool @@ -203,17 +213,31 @@ def evaluate_time_window(window: TimeWindowSettings, now: datetime) -> Condition return ConditionState(enabled=window.enabled, available=True, open=is_open) -def evaluate_display_powersave(ps: DisplayPowersaveSettings) -> ConditionState: - """Evaluate the display-powersave forward seam, currently unavailable.""" - return ConditionState(enabled=ps.enabled, available=False, open=False) +def evaluate_display_powersave( + ps: DisplayPowersaveSettings, + reading: DisplayPowersaveReading, +) -> ConditionState: + """Evaluate the display-powersave condition from a debounced reading.""" + if not ps.enabled: + return ConditionState(enabled=False, available=False, open=False) + return ConditionState( + enabled=True, + available=reading.available, + open=reading.asleep and reading.debounced, + ) -def evaluate_drain_gate(settings: ProcessingSettings, now: datetime) -> GateState: +def evaluate_drain_gate( + settings: ProcessingSettings, + now: datetime, + display_reading: DisplayPowersaveReading, +) -> GateState: """Evaluate all drain gate conditions with OR composition.""" conditions = { "time_window": evaluate_time_window(settings.gate.time_window, now), "display_powersave": evaluate_display_powersave( - settings.gate.display_powersave + settings.gate.display_powersave, + display_reading, ), } return GateState( @@ -345,6 +369,8 @@ __all__ = [ "DRAIN_STATE_WINDOW_OPEN", "DEFAULT_PROCESSING", "ConditionState", + "DISPLAY_POWERSAVE_UNAVAILABLE", + "DisplayPowersaveReading", "DisplayPowersaveSettings", "GateSettings", "GateState", diff --git a/solstone/think/supervisor.py b/solstone/think/supervisor.py index 44e7799b0..9e8d7cd56 100644 --- a/solstone/think/supervisor.py +++ b/solstone/think/supervisor.py @@ -39,9 +39,17 @@ from solstone.think.catchup_state import ( record_attempt, record_outcome, ) +from solstone.think.display_powersave import ( + poll_display_powersave, + reset_display_powersave_monitor, +) from solstone.think.maint import run_pending_tasks from solstone.think.models import LOCAL_MODEL, is_local_provider_needed -from solstone.think.processing import evaluate_drain_gate, load_processing_settings +from solstone.think.processing import ( + DISPLAY_POWERSAVE_UNAVAILABLE, + evaluate_drain_gate, + load_processing_settings, +) from solstone.think.providers.mlx_server import MLX_SERVER_PROCESS_NAME from solstone.think.readiness import START_TIME_TOLERANCE_S, clear_ready, signal_ready from solstone.think.runner import ManagedProcess as RunnerManagedProcess @@ -2592,7 +2600,12 @@ def _run_gate_tick(now: float) -> None: settings = load_processing_settings() if settings.mode != "deferred": return - gate = evaluate_drain_gate(settings, datetime.now()) + reading = ( + poll_display_powersave(time.monotonic()) + if settings.gate.display_powersave.enabled + else DISPLAY_POWERSAVE_UNAVAILABLE + ) + gate = evaluate_drain_gate(settings, datetime.now(), reading) if not gate.open: return run_catchup_drain() @@ -2614,6 +2627,7 @@ async def supervise( last_status_emit = 0.0 _last_gate_tick = 0.0 + reset_display_powersave_monitor() _last_sync_tick = 0.0 _last_sync_snapshot = None _sync_conflict_shutdown = False diff --git a/solstone/think/surfaces/health.py b/solstone/think/surfaces/health.py index 095923f0d..149306304 100644 --- a/solstone/think/surfaces/health.py +++ b/solstone/think/surfaces/health.py @@ -22,6 +22,10 @@ from solstone.convey.readiness_snapshot import ( unavailable_snapshot, ) from solstone.think.activities import load_activity_records +from solstone.think.display_powersave import ( + display_powersave_detectable, + last_display_powersave, +) from solstone.think.entities.journal import load_all_journal_entities from solstone.think.facets import get_facets from solstone.think.pipeline_health import read_segment_backlog @@ -507,7 +511,8 @@ def _build_segment_backlog_health() -> SegmentBacklogHealth: 1 for completion in backlog.per_day.values() if completion.not_thought > 0 ) settings = load_processing_settings() - gate = evaluate_drain_gate(settings, datetime.now()) + reading = last_display_powersave() + gate = evaluate_drain_gate(settings, datetime.now(), reading) drain_state = derive_drain_state(settings, gate) awaiting_total = backlog.not_sensed + backlog.not_thought awaiting_text = ( @@ -523,6 +528,7 @@ def _build_segment_backlog_health() -> SegmentBacklogHealth: awaiting_analysis_text=awaiting_text, last_drained_at=read_last_drained_at(), drain_state=drain_state, + display_powersave_detectable=display_powersave_detectable(), ) diff --git a/solstone/think/surfaces/types.py b/solstone/think/surfaces/types.py index a07a6daf8..f11b27f10 100644 --- a/solstone/think/surfaces/types.py +++ b/solstone/think/surfaces/types.py @@ -129,6 +129,7 @@ class SegmentBacklogHealth: awaiting_analysis_text: str | None = None last_drained_at: int | None = None drain_state: str = DRAIN_STATE_REALTIME + display_powersave_detectable: bool = False @dataclass(frozen=True) diff --git a/tests/test_display_powersave.py b/tests/test_display_powersave.py new file mode 100644 index 000000000..bee180c39 --- /dev/null +++ b/tests/test_display_powersave.py @@ -0,0 +1,382 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import subprocess +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from solstone.think import display_powersave as dp +from solstone.think.processing import DISPLAY_POWERSAVE_UNAVAILABLE + + +@pytest.fixture(autouse=True) +def reset_display_monitor(): + dp.reset_display_powersave_monitor() + yield + dp.reset_display_powersave_monitor() + + +def _connector( + root: Path, + name: str, + *, + status: str | None, + enabled: str | None, + dpms: str | None, +) -> Path: + path = root / name + path.mkdir(parents=True) + for attr_name, value in { + "status": status, + "enabled": enabled, + "dpms": dpms, + }.items(): + if value is not None: + (path / attr_name).write_text(value, encoding="utf-8") + return path + + +def _read_linux(root: Path): + return dp.read_display_power(platform="linux", sysfs_root=root) + + +def _reader(state: str, reason: str | None = None): + return lambda: dp.DisplayReading(state, reason) + + +def test_read_linux_returns_awake_when_any_considered_connector_is_on( + tmp_path: Path, +) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="connected", + enabled="enabled", + dpms="On", + ) + + assert _read_linux(tmp_path) == dp.DisplayReading(dp.DISPLAY_AWAKE, None) + + +@pytest.mark.parametrize("dpms", ["Off", "Standby", "Suspend"]) +def test_read_linux_returns_asleep_when_all_considered_connectors_are_off_states( + tmp_path: Path, + dpms: str, +) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="connected", + enabled="enabled", + dpms=dpms, + ) + + assert _read_linux(tmp_path) == dp.DisplayReading(dp.DISPLAY_ASLEEP, None) + + +def test_read_linux_ignores_disconnected_and_disabled_stale_dpms( + tmp_path: Path, +) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="disconnected", + enabled="enabled", + dpms="On", + ) + _connector( + tmp_path, + "card1-DP-1", + status="connected", + enabled="disabled", + dpms="On", + ) + + assert _read_linux(tmp_path) == dp.DisplayReading( + dp.DISPLAY_UNDETECTABLE, + dp.REASON_HEADLESS, + ) + + +def test_read_linux_mixed_on_and_off_returns_awake(tmp_path: Path) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="connected", + enabled="enabled", + dpms="Off", + ) + _connector( + tmp_path, + "card1-DP-1", + status="connected", + enabled="enabled", + dpms="On", + ) + + assert _read_linux(tmp_path) == dp.DisplayReading(dp.DISPLAY_AWAKE, None) + + +def test_read_linux_returns_unreadable_for_missing_or_unknown_considered_dpms( + tmp_path: Path, +) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="connected", + enabled="enabled", + dpms="Off", + ) + _connector( + tmp_path, + "card1-DP-1", + status="connected", + enabled="enabled", + dpms=None, + ) + + assert _read_linux(tmp_path) == dp.DisplayReading( + dp.DISPLAY_UNDETECTABLE, + dp.REASON_UNREADABLE, + ) + + +def test_read_linux_on_wins_over_unreadable_sibling(tmp_path: Path) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="connected", + enabled="enabled", + dpms=None, + ) + _connector( + tmp_path, + "card1-DP-1", + status="connected", + enabled="enabled", + dpms="On", + ) + + assert _read_linux(tmp_path) == dp.DisplayReading(dp.DISPLAY_AWAKE, None) + + +def test_read_linux_returns_headless_when_no_connected_enabled_connectors( + tmp_path: Path, +) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="disconnected", + enabled="enabled", + dpms="Off", + ) + + assert _read_linux(tmp_path) == dp.DisplayReading( + dp.DISPLAY_UNDETECTABLE, + dp.REASON_HEADLESS, + ) + + +def test_read_linux_returns_unreadable_when_no_connectors(tmp_path: Path) -> None: + assert _read_linux(tmp_path) == dp.DisplayReading( + dp.DISPLAY_UNDETECTABLE, + dp.REASON_UNREADABLE, + ) + + +def test_read_linux_returns_unreadable_when_root_missing(tmp_path: Path) -> None: + assert _read_linux(tmp_path / "missing") == dp.DisplayReading( + dp.DISPLAY_UNDETECTABLE, + dp.REASON_UNREADABLE, + ) + + +def test_read_linux_returns_unreadable_when_glob_raises() -> None: + class BrokenRoot: + def glob(self, _pattern): + raise OSError("no drm") + + assert dp.read_display_power(platform="linux", sysfs_root=BrokenRoot()) == ( + dp.DisplayReading(dp.DISPLAY_UNDETECTABLE, dp.REASON_UNREADABLE) + ) + + +def test_read_display_power_wraps_detector_exception_as_unreadable( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _connector( + tmp_path, + "card1-HDMI-A-2", + status="connected", + enabled="enabled", + dpms="On", + ) + + def fail_read_attr(_path: Path, _name: str) -> str: + raise RuntimeError("boom") + + monkeypatch.setattr(dp, "_read_attr", fail_read_attr) + + assert _read_linux(tmp_path) == dp.DisplayReading( + dp.DISPLAY_UNDETECTABLE, + dp.REASON_UNREADABLE, + ) + + +def test_read_display_power_dispatches_linux_and_darwin( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + dp, + "_read_linux", + lambda root: dp.DisplayReading(dp.DISPLAY_AWAKE, None), + ) + monkeypatch.setattr( + dp, + "_read_macos", + lambda: dp.DisplayReading(dp.DISPLAY_ASLEEP, None), + ) + + assert dp.read_display_power(platform="linux", sysfs_root=tmp_path).state == ( + dp.DISPLAY_AWAKE + ) + assert dp.read_display_power(platform="darwin", sysfs_root=tmp_path).state == ( + dp.DISPLAY_ASLEEP + ) + + +def test_read_display_power_returns_unsupported_platform(tmp_path: Path) -> None: + assert dp.read_display_power(platform="win32", sysfs_root=tmp_path) == ( + dp.DisplayReading(dp.DISPLAY_UNDETECTABLE, dp.REASON_UNSUPPORTED_PLATFORM) + ) + + +def test_parse_macos_log_uses_last_display_marker() -> None: + text = "\n".join( + [ + "2026-01-01 Display is turned off", + "2026-01-01 Display is turned on", + ] + ) + + assert dp._parse_macos_log(text) == dp.DisplayReading(dp.DISPLAY_AWAKE, None) + + +def test_parse_macos_log_without_marker_returns_no_window_server() -> None: + assert dp._parse_macos_log("unrelated line") == dp.DisplayReading( + dp.DISPLAY_UNDETECTABLE, + dp.REASON_NO_WINDOW_SERVER, + ) + + +@pytest.mark.parametrize( + ("asleep", "expected_state"), + [(True, dp.DISPLAY_ASLEEP), (False, dp.DISPLAY_AWAKE)], +) +def test_read_macos_uses_coregraphics_sleep_state_when_main_display_exists( + monkeypatch: pytest.MonkeyPatch, + asleep: bool, + expected_state: str, +) -> None: + monkeypatch.setattr(dp, "_macos_main_display_id", lambda: 1) + monkeypatch.setattr(dp, "_macos_display_is_asleep", lambda display_id: asleep) + + assert dp._read_macos() == dp.DisplayReading(expected_state, None) + + +def test_read_macos_falls_back_to_pmset_log_when_main_display_is_zero( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(dp, "_macos_main_display_id", lambda: 0) + monkeypatch.setattr(dp, "_macos_pmset_log", lambda: "Display is turned off") + + assert dp._read_macos() == dp.DisplayReading(dp.DISPLAY_ASLEEP, None) + + +def test_macos_pmset_log_contains_timeout_and_oserror( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def timeout(*_args, **_kwargs): + raise subprocess.TimeoutExpired(["pmset", "-g", "log"], timeout=1) + + monkeypatch.setattr(dp.subprocess, "run", timeout) + assert dp._macos_pmset_log() == "" + + def oserror(*_args, **_kwargs): + raise OSError("pmset missing") + + monkeypatch.setattr(dp.subprocess, "run", oserror) + assert dp._macos_pmset_log() == "" + + +def test_monitor_debounces_only_after_threshold_from_injected_now() -> None: + monitor = dp.DisplayPowersaveMonitor(debounce_s=120.0) + + first = monitor.poll(now=0.0, read=_reader(dp.DISPLAY_ASLEEP)) + before_threshold = monitor.poll(now=119.0, read=_reader(dp.DISPLAY_ASLEEP)) + after_threshold = monitor.poll(now=240.0, read=_reader(dp.DISPLAY_ASLEEP)) + + assert first.debounced is False + assert before_threshold.debounced is False + assert after_threshold.debounced is True + + +def test_monitor_wake_closes_immediately_and_resets_since() -> None: + monitor = dp.DisplayPowersaveMonitor(debounce_s=120.0) + monitor.poll(now=0.0, read=_reader(dp.DISPLAY_ASLEEP)) + monitor.poll(now=240.0, read=_reader(dp.DISPLAY_ASLEEP)) + + awake = monitor.poll(now=241.0, read=_reader(dp.DISPLAY_AWAKE)) + next_asleep = monitor.poll(now=242.0, read=_reader(dp.DISPLAY_ASLEEP)) + + assert awake.available is True + assert awake.asleep is False + assert awake.debounced is False + assert next_asleep.debounced is False + + +def test_monitor_undetectable_resets_snapshot_and_debounce() -> None: + monitor = dp.DisplayPowersaveMonitor(debounce_s=120.0) + monitor.poll(now=0.0, read=_reader(dp.DISPLAY_ASLEEP)) + undetectable = monitor.poll( + now=60.0, + read=_reader(dp.DISPLAY_UNDETECTABLE, dp.REASON_UNREADABLE), + ) + next_asleep = monitor.poll(now=180.0, read=_reader(dp.DISPLAY_ASLEEP)) + + assert undetectable == DISPLAY_POWERSAVE_UNAVAILABLE + assert next_asleep.debounced is False + + +def test_monitor_capability_known_is_sticky_until_reset() -> None: + monitor = dp.DisplayPowersaveMonitor(debounce_s=120.0) + + assert monitor.capability_known() is False + monitor.poll(now=0.0, read=_reader(dp.DISPLAY_AWAKE)) + assert monitor.capability_known() is True + monitor.poll( + now=1.0, + read=_reader(dp.DISPLAY_UNDETECTABLE, dp.REASON_UNREADABLE), + ) + assert monitor.capability_known() is True + monitor.reset() + assert monitor.capability_known() is False + + +def test_singleton_last_and_reset_are_non_polling( + monkeypatch: pytest.MonkeyPatch, +) -> None: + snapshot = dp._MONITOR.poll(now=0.0, read=_reader(dp.DISPLAY_AWAKE)) + read = MagicMock(side_effect=AssertionError("last should not poll")) + monkeypatch.setattr(dp, "read_display_power", read) + + assert dp.last_display_powersave() == snapshot + read.assert_not_called() + dp.reset_display_powersave_monitor() + assert dp.last_display_powersave() == DISPLAY_POWERSAVE_UNAVAILABLE diff --git a/tests/test_health_call_parity.py b/tests/test_health_call_parity.py index 98931d660..d23442b31 100644 --- a/tests/test_health_call_parity.py +++ b/tests/test_health_call_parity.py @@ -95,6 +95,7 @@ def _report() -> dict[str, object]: "awaiting_analysis_text": None, "last_drained_at": None, "drain_state": DRAIN_STATE_REALTIME, + "display_powersave_detectable": False, }, "notes": [], "provider_readiness": { diff --git a/tests/test_processing.py b/tests/test_processing.py index f64c99b84..f9325cdf5 100644 --- a/tests/test_processing.py +++ b/tests/test_processing.py @@ -12,11 +12,13 @@ import pytest from solstone.think.processing import ( AWAITING_ANALYSIS_TEMPLATE, DEFAULT_PROCESSING, + DISPLAY_POWERSAVE_UNAVAILABLE, DRAIN_STATE_NO_CONDITION, DRAIN_STATE_REALTIME, DRAIN_STATE_WAITING, DRAIN_STATE_WINDOW_OPEN, ConditionState, + DisplayPowersaveReading, DisplayPowersaveSettings, GateSettings, GateState, @@ -205,6 +207,7 @@ def test_evaluate_drain_gate_uses_or_composition() -> None: display_powersave=DisplayPowersaveSettings(enabled=True), ), datetime(2026, 1, 1, 3, 0), + DISPLAY_POWERSAVE_UNAVAILABLE, ).open is True ) @@ -216,19 +219,70 @@ def test_evaluate_drain_gate_uses_or_composition() -> None: display_powersave=DisplayPowersaveSettings(enabled=True), ), datetime(2026, 1, 1, 3, 0), + DISPLAY_POWERSAVE_UNAVAILABLE, ).open is False ) -def test_evaluate_display_powersave_is_inert() -> None: - state = evaluate_display_powersave(DisplayPowersaveSettings(enabled=True)) +def test_evaluate_display_powersave_disabled_ignores_reading() -> None: + state = evaluate_display_powersave( + DisplayPowersaveSettings(enabled=False), + DisplayPowersaveReading(available=True, asleep=True, debounced=True), + ) - assert state.enabled is True + assert state.enabled is False assert state.available is False assert state.open is False +def test_evaluate_display_powersave_enabled_uses_reading() -> None: + asleep = evaluate_display_powersave( + DisplayPowersaveSettings(enabled=True), + DisplayPowersaveReading(available=True, asleep=True, debounced=True), + ) + awake = evaluate_display_powersave( + DisplayPowersaveSettings(enabled=True), + DisplayPowersaveReading(available=True, asleep=False, debounced=False), + ) + undetectable = evaluate_display_powersave( + DisplayPowersaveSettings(enabled=True), + DISPLAY_POWERSAVE_UNAVAILABLE, + ) + + assert asleep == ConditionState(enabled=True, available=True, open=True) + assert awake == ConditionState(enabled=True, available=True, open=False) + assert undetectable == ConditionState(enabled=True, available=False, open=False) + + +def test_evaluate_drain_gate_opens_on_debounced_display_powersave() -> None: + gate = evaluate_drain_gate( + _settings( + time_window=_time_window(enabled=False), + display_powersave=DisplayPowersaveSettings(enabled=True), + ), + datetime(2026, 1, 1, 12, 0), + DisplayPowersaveReading(available=True, asleep=True, debounced=True), + ) + + assert gate.open is True + + +def test_display_powersave_undetectable_has_no_active_condition() -> None: + settings = _settings( + time_window=_time_window(enabled=False), + display_powersave=DisplayPowersaveSettings(enabled=True), + ) + gate = evaluate_drain_gate( + settings, + datetime(2026, 1, 1, 12, 0), + DISPLAY_POWERSAVE_UNAVAILABLE, + ) + + assert gate.open is False + assert derive_drain_state(settings, gate) == DRAIN_STATE_NO_CONDITION + + def test_derive_drain_state_tokens() -> None: open_gate = GateState( open=True, diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index 6ad5dfd3f..5400035c1 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -21,6 +21,8 @@ import psutil import pytest from solstone.think.processing import ( + DISPLAY_POWERSAVE_UNAVAILABLE, + DisplayPowersaveReading, DisplayPowersaveSettings, GateSettings, ProcessingSettings, @@ -626,7 +628,11 @@ class _CaptureTaskQueue: self.submissions.append({"cmd": cmd, "day": day}) -def _supervisor_processing_settings(mode: str) -> ProcessingSettings: +def _supervisor_processing_settings( + mode: str, + *, + display_powersave_enabled: bool = False, +) -> ProcessingSettings: return ProcessingSettings( mode=mode, gate=GateSettings( @@ -635,7 +641,9 @@ def _supervisor_processing_settings(mode: str) -> ProcessingSettings: start="02:00", end="06:00", ), - display_powersave=DisplayPowersaveSettings(enabled=False), + display_powersave=DisplayPowersaveSettings( + enabled=display_powersave_enabled + ), ), ) @@ -1937,7 +1945,7 @@ def test_run_gate_tick_deferred_open_runs_catchup_drain(monkeypatch): monkeypatch.setattr( mod, "evaluate_drain_gate", - lambda settings, now: SimpleNamespace(open=True), + lambda settings, now, reading: SimpleNamespace(open=True), ) monkeypatch.setattr(mod, "run_catchup_drain", drain) mod._last_gate_tick = 0.0 @@ -1958,7 +1966,7 @@ def test_run_gate_tick_deferred_closed_skips_catchup_drain(monkeypatch): monkeypatch.setattr( mod, "evaluate_drain_gate", - lambda settings, now: SimpleNamespace(open=False), + lambda settings, now, reading: SimpleNamespace(open=False), ) monkeypatch.setattr(mod, "run_catchup_drain", drain) mod._last_gate_tick = 0.0 @@ -1998,7 +2006,7 @@ def test_run_gate_tick_throttles_catchup_drain(monkeypatch): monkeypatch.setattr( mod, "evaluate_drain_gate", - lambda settings, now: SimpleNamespace(open=True), + lambda settings, now, reading: SimpleNamespace(open=True), ) monkeypatch.setattr(mod, "run_catchup_drain", drain) mod._last_gate_tick = 0.0 @@ -2010,6 +2018,74 @@ def test_run_gate_tick_throttles_catchup_drain(monkeypatch): assert drain.call_count == 2 +def test_run_gate_tick_disabled_display_powersave_does_not_poll(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + drain = MagicMock() + poll = MagicMock(side_effect=AssertionError("poll should not be called")) + + def evaluate(settings, now, reading): + assert reading == DISPLAY_POWERSAVE_UNAVAILABLE + return SimpleNamespace(open=True) + + monkeypatch.setattr(mod, "_is_remote_mode", False) + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("deferred"), + ) + monkeypatch.setattr(mod, "poll_display_powersave", poll) + monkeypatch.setattr(mod, "evaluate_drain_gate", evaluate) + monkeypatch.setattr(mod, "run_catchup_drain", drain) + mod._last_gate_tick = 0.0 + + mod._run_gate_tick(60.0) + + poll.assert_not_called() + drain.assert_called_once_with() + + +def test_run_gate_tick_enabled_display_powersave_polls(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + reading = DisplayPowersaveReading(available=True, asleep=True, debounced=True) + poll = MagicMock(return_value=reading) + captured = {} + + def evaluate(settings, now, display_reading): + captured["reading"] = display_reading + return SimpleNamespace(open=False) + + monkeypatch.setattr(mod, "_is_remote_mode", False) + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings( + "deferred", + display_powersave_enabled=True, + ), + ) + monkeypatch.setattr(mod, "poll_display_powersave", poll) + monkeypatch.setattr(mod, "evaluate_drain_gate", evaluate) + monkeypatch.setattr(mod, "run_catchup_drain", MagicMock()) + mod._last_gate_tick = 0.0 + + mod._run_gate_tick(60.0) + + poll.assert_called_once() + assert isinstance(poll.call_args.args[0], float) + assert captured["reading"] == reading + + +def test_supervise_resets_display_powersave_monitor_on_entry(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + reset = MagicMock() + monkeypatch.setattr(mod, "reset_display_powersave_monitor", reset) + monkeypatch.setattr(mod, "shutdown_requested", True) + + asyncio.run(mod.supervise(daily=False, schedule=False, procs=[])) + + reset.assert_called_once_with() + + def test_record_scheduler_completion_serializes_concurrent_writes( tmp_path, monkeypatch ): diff --git a/tests/test_surfaces_health.py b/tests/test_surfaces_health.py index b1686db8b..fec392e79 100644 --- a/tests/test_surfaces_health.py +++ b/tests/test_surfaces_health.py @@ -8,6 +8,7 @@ import os import re from datetime import UTC, datetime, timedelta from pathlib import Path +from unittest.mock import MagicMock import pytest from click.testing import Result @@ -17,11 +18,13 @@ from solstone.convey.readiness_snapshot import unavailable_snapshot from solstone.think.convey_client import ConveyClient from solstone.think.pipeline_health import SegmentBacklog, SegmentCompletion from solstone.think.processing import ( + DISPLAY_POWERSAVE_UNAVAILABLE, DRAIN_STATE_NO_CONDITION, DRAIN_STATE_REALTIME, DRAIN_STATE_WAITING, DRAIN_STATE_WINDOW_OPEN, ConditionState, + DisplayPowersaveReading, DisplayPowersaveSettings, GateSettings, GateState, @@ -243,6 +246,20 @@ def _processing_settings(mode: str) -> ProcessingSettings: ) +def _stub_display_powersave( + monkeypatch: pytest.MonkeyPatch, + *, + reading=DISPLAY_POWERSAVE_UNAVAILABLE, + detectable: bool = False, +) -> None: + monkeypatch.setattr(health_surface, "last_display_powersave", lambda: reading) + monkeypatch.setattr( + health_surface, + "display_powersave_detectable", + lambda: detectable, + ) + + def _clear_readiness_snapshot() -> dict[str, object]: return { "summary": { @@ -586,9 +603,10 @@ def test_segment_backlog_deferred_awaiting_analysis_uses_unsensed( monkeypatch.setattr( health_surface, "evaluate_drain_gate", - lambda settings, now: GateState(open=False, conditions={}), + lambda settings, now, reading: GateState(open=False, conditions={}), ) monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + _stub_display_powersave(monkeypatch) backlog = health_surface._build_segment_backlog_health() @@ -611,9 +629,10 @@ def test_segment_backlog_realtime_omits_awaiting_analysis_text(monkeypatch) -> N monkeypatch.setattr( health_surface, "evaluate_drain_gate", - lambda settings, now: GateState(open=True, conditions={}), + lambda settings, now, reading: GateState(open=True, conditions={}), ) monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + _stub_display_powersave(monkeypatch) backlog = health_surface._build_segment_backlog_health() @@ -664,9 +683,10 @@ def test_segment_backlog_drain_state_tokens(monkeypatch, gate, expected_state) - monkeypatch.setattr( health_surface, "evaluate_drain_gate", - lambda settings, now: gate, + lambda settings, now, reading: gate, ) monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + _stub_display_powersave(monkeypatch) backlog = health_surface._build_segment_backlog_health() @@ -687,19 +707,99 @@ def test_segment_backlog_last_drained_at_passes_through(monkeypatch) -> None: monkeypatch.setattr( health_surface, "evaluate_drain_gate", - lambda settings, now: GateState(open=False, conditions={}), + lambda settings, now, reading: GateState(open=False, conditions={}), ) monkeypatch.setattr( health_surface, "read_last_drained_at", lambda: 1_700_000_000_000, ) + _stub_display_powersave(monkeypatch) backlog = health_surface._build_segment_backlog_health() assert backlog.last_drained_at == 1_700_000_000_000 +def test_segment_backlog_uses_last_display_powersave_snapshot(monkeypatch) -> None: + expected = DisplayPowersaveReading(available=True, asleep=True, debounced=True) + captured = {} + monkeypatch.setattr( + health_surface, + "read_segment_backlog", + lambda: _segment_backlog({}), + ) + monkeypatch.setattr( + health_surface, + "load_processing_settings", + lambda: _processing_settings("deferred"), + ) + + def evaluate(settings, now, reading): + captured["reading"] = reading + return GateState(open=True, conditions={}) + + monkeypatch.setattr(health_surface, "evaluate_drain_gate", evaluate) + monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + _stub_display_powersave(monkeypatch, reading=expected) + + health_surface._build_segment_backlog_health() + + assert captured["reading"] == expected + + +def test_segment_backlog_exposes_display_powersave_detectable(monkeypatch) -> None: + monkeypatch.setattr( + health_surface, + "read_segment_backlog", + lambda: _segment_backlog({}), + ) + monkeypatch.setattr( + health_surface, + "load_processing_settings", + lambda: _processing_settings("deferred"), + ) + monkeypatch.setattr( + health_surface, + "evaluate_drain_gate", + lambda settings, now, reading: GateState(open=False, conditions={}), + ) + monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + _stub_display_powersave(monkeypatch, detectable=True) + + backlog = health_surface._build_segment_backlog_health() + + assert backlog.display_powersave_detectable is True + + +def test_segment_backlog_health_never_polls_display(monkeypatch) -> None: + from solstone.think import display_powersave + + poll = MagicMock() + monkeypatch.setattr(display_powersave, "poll_display_powersave", poll) + monkeypatch.setattr( + health_surface, + "read_segment_backlog", + lambda: _segment_backlog({}), + ) + monkeypatch.setattr( + health_surface, + "load_processing_settings", + lambda: _processing_settings("deferred"), + ) + monkeypatch.setattr( + health_surface, + "evaluate_drain_gate", + lambda settings, now, reading: GateState(open=False, conditions={}), + ) + monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + _stub_display_powersave(monkeypatch) + + health_surface._build_segment_backlog_health() + + poll.assert_not_called() + + def test_silent_facet_note_ladder_thresholds(tmp_path, monkeypatch): _configure_env(tmp_path, monkeypatch) _set_now(monkeypatch, _utc_dt("20260410"))