diff --git a/solstone/apps/settings/call.py b/solstone/apps/settings/call.py index 8fb2c981f..bd1d13294 100644 --- a/solstone/apps/settings/call.py +++ b/solstone/apps/settings/call.py @@ -53,6 +53,8 @@ observer_app = typer.Typer(help="Observer capture settings.") app.add_typer(observer_app, name="observer") convey_app = typer.Typer(help="Convey access configuration.") app.add_typer(convey_app, name="convey") +processing_app = typer.Typer(help="Deferred-processing mode and drain window.") +app.add_typer(processing_app, name="processing") def _request( @@ -112,6 +114,71 @@ def _moved_stub(command: str) -> None: raise typer.Exit(2) +@processing_app.command("show") +@convey_cli +def processing_show() -> None: + """Show effective deferred-processing settings.""" + result = _request("GET", "/app/settings/api/processing") + _echo_json(result) + + +@processing_app.command("set") +@convey_cli +def processing_set( + mode: str | None = typer.Option(None, "--mode", help="realtime or deferred."), + window_start: str | None = typer.Option( + None, + "--window-start", + help="HH:MM drain window start.", + ), + window_end: str | None = typer.Option( + None, + "--window-end", + help="HH:MM drain window end.", + ), + time_window: bool | None = typer.Option( + None, + "--time-window/--no-time-window", + help="Enable the time-window drain condition.", + ), + display_powersave: bool | None = typer.Option( + None, + "--display-powersave/--no-display-powersave", + help="Enable the display-powersave drain condition.", + ), +) -> None: + """Set deferred-processing mode and drain window.""" + data: dict[str, Any] = {} + if mode is not None: + data["mode"] = mode + tw: dict[str, Any] = {} + if window_start is not None: + tw["start"] = window_start + if window_end is not None: + tw["end"] = window_end + if time_window is not None: + tw["enabled"] = time_window + gate: dict[str, Any] = {} + if tw: + gate["time_window"] = tw + if display_powersave is not None: + gate["display_powersave"] = {"enabled": display_powersave} + if gate: + data["gate"] = gate + if not data: + _exit_with( + "error: provide at least one of " + "--mode/--window-start/--window-end/--time-window/--display-powersave" + ) + try: + response = _post_config("processing", data) + except ConveyClientError as err: + if err.reason_code == INVALID_CONFIG_VALUE.code and err.detail: + _exit_with(err.detail) + raise + _echo_json(response.get("config", {}).get("processing", {})) + + @convey_app.command("host-url") @convey_cli def convey_host_url( diff --git a/solstone/apps/settings/routes.py b/solstone/apps/settings/routes.py index 170e12cbc..bedc176e7 100644 --- a/solstone/apps/settings/routes.py +++ b/solstone/apps/settings/routes.py @@ -57,6 +57,10 @@ from solstone.think.journal_config import ( hold_config_lock, write_journal_config, ) +from solstone.think.processing import ( + load_processing_settings, + validate_processing_update, +) from solstone.think.retention import ( _human_bytes, check_storage_health, @@ -267,6 +271,7 @@ def update_config() -> Any: "support": ["enabled", "proactive", "anonymous_feedback", "portal_url"], "agent": ["name", "name_status", "named_date"], "env": API_KEY_ENV_VARS, + "processing": [], } # Nested config schemas for transcribe backends - built from BACKEND_METADATA @@ -310,6 +315,19 @@ def update_config() -> Any: else: os.environ.pop(key, None) + if section == "processing": + try: + validated = validate_processing_update(old_section, data) + except ValueError as exc: + return error_response(INVALID_CONFIG_VALUE, detail=str(exc)) + new_section = validated.to_dict() + if old_section != new_section: + changed_fields["processing"] = { + "old": old_section, + "new": new_section, + } + config["processing"] = new_section + # Handle nested backend configs for transcribe section if section == "transcribe": for backend_key, allowed_keys in transcribe_nested.items(): @@ -526,6 +544,19 @@ def get_transcribe() -> Any: return _settings_operation_failed() +@settings_bp.route("/api/processing") +def get_processing() -> Any: + """Return effective deferred-processing settings.""" + try: + return jsonify(load_processing_settings().to_dict()) + except Exception: + logger.exception("error loading processing settings") + return error_response( + SETTINGS_OPERATION_FAILED, + detail="unable to load processing settings", + ) + + # --------------------------------------------------------------------------- # Sol Voice API # --------------------------------------------------------------------------- diff --git a/solstone/apps/settings/tests/test_processing_routes.py b/solstone/apps/settings/tests/test_processing_routes.py new file mode 100644 index 000000000..befaf4c51 --- /dev/null +++ b/solstone/apps/settings/tests/test_processing_routes.py @@ -0,0 +1,119 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from solstone.convey import create_app +from solstone.convey.reasons import INVALID_CONFIG_VALUE + + +def _base_config() -> dict: + return { + "setup": {"completed_at": "2026-05-09T00:00:00Z"}, + } + + +def _client(journal_path): + app = create_app(str(journal_path)) + app.config["TESTING"] = True + return app.test_client() + + +def _post_processing(client, data: dict): + return client.post( + "/app/settings/api/config", + json={"section": "processing", "data": data}, + ) + + +def test_processing_api_get_returns_effective_defaults(settings_env): + journal_path, _config = settings_env(_base_config()) + client = _client(journal_path) + + response = client.get("/app/settings/api/processing") + + assert response.status_code == 200 + payload = response.get_json() + assert payload["mode"] == "realtime" + assert payload["gate"]["time_window"]["start"] == "02:00" + + +def test_processing_api_post_round_trips_nested_window(settings_env): + journal_path, _config = settings_env(_base_config()) + client = _client(journal_path) + + response = _post_processing( + client, + { + "mode": "deferred", + "gate": {"time_window": {"start": "01:00", "end": "05:00"}}, + }, + ) + + assert response.status_code == 200 + payload = client.get("/app/settings/api/processing").get_json() + assert payload["mode"] == "deferred" + assert payload["gate"]["time_window"] == { + "enabled": True, + "start": "01:00", + "end": "05:00", + } + assert payload["gate"]["display_powersave"] == {"enabled": False} + + +def test_processing_api_post_bad_hhmm_returns_invalid_config(settings_env): + journal_path, _config = settings_env(_base_config()) + client = _client(journal_path) + + response = _post_processing( + client, + {"gate": {"time_window": {"start": "25:99"}}}, + ) + + assert response.status_code == 400 + payload = response.get_json() + assert payload["reason_code"] == INVALID_CONFIG_VALUE.code + assert "HH:MM" in payload["detail"] + + +def test_processing_api_accepts_deferred_without_enabled_conditions(settings_env): + journal_path, _config = settings_env(_base_config()) + client = _client(journal_path) + + response = _post_processing( + client, + { + "mode": "deferred", + "gate": { + "time_window": {"enabled": False}, + "display_powersave": {"enabled": False}, + }, + }, + ) + + assert response.status_code == 200 + payload = client.get("/app/settings/api/processing").get_json() + assert payload["mode"] == "deferred" + assert payload["gate"]["time_window"]["enabled"] is False + assert payload["gate"]["display_powersave"]["enabled"] is False + + +def test_processing_api_deep_merges_partial_updates(settings_env): + journal_path, _config = settings_env(_base_config()) + client = _client(journal_path) + + first = _post_processing( + client, + { + "mode": "deferred", + "gate": {"time_window": {"start": "01:00", "end": "05:00"}}, + }, + ) + second = _post_processing(client, {"mode": "realtime"}) + + assert first.status_code == 200 + assert second.status_code == 200 + payload = client.get("/app/settings/api/processing").get_json() + assert payload["mode"] == "realtime" + assert payload["gate"]["time_window"]["start"] == "01:00" + assert payload["gate"]["time_window"]["end"] == "05:00" diff --git a/solstone/observe/sense.py b/solstone/observe/sense.py index 592d5f46d..6c2fd252e 100644 --- a/solstone/observe/sense.py +++ b/solstone/observe/sense.py @@ -31,6 +31,7 @@ from solstone.observe.utils import ( VIDEO_EXTENSIONS, ) from solstone.think.callosum import CallosumConnection +from solstone.think.processing import load_processing_settings from solstone.think.runner import KILL_REAP_GRACE_S from solstone.think.runner import ManagedProcess as RunnerManagedProcess from solstone.think.utils import ( @@ -720,6 +721,10 @@ class FileSensor: meta = {} meta["stream"] = stream + deferred_live = ( + not message.get("batch") and load_processing_settings().mode == "deferred" + ) + # Build full paths for all files in this segment. rel_segment = f"{day}/{stream}/{segment}" if stream else f"{day}/{segment}" segment_dir = resolve_journal_path(self.journal_dir, rel_segment) @@ -737,20 +742,25 @@ class FileSensor: self.segment_batch[segment] = True if observer: self.segment_observer[segment] = observer - for file_path in file_paths: - # Only track files that will be processed (match a pattern) - if self._match_pattern(file_path): - self.segment_files[segment].add(file_path) + if not deferred_live: + for file_path in file_paths: + # Only track files that will be processed (match a pattern) + if self._match_pattern(file_path): + self.segment_files[segment].add(file_path) # Process each file (pass segment context for env vars) - for file_path in file_paths: - self._handle_file(file_path, segment=segment, observer=observer, meta=meta) + if not deferred_live: + for file_path in file_paths: + self._handle_file( + file_path, segment=segment, observer=observer, meta=meta + ) # If no files matched any handler patterns, emit observed immediately # (e.g., tmux-only segments with just .jsonl files) with self.lock: if segment in self.segment_files and not self.segment_files[segment]: - self._emit_segment_observed(segment, note="no handlers") + emit_note = "deferred" if deferred_live else "no handlers" + self._emit_segment_observed(segment, note=emit_note) def _emit_status(self): """Emit observe.status event with current processing state (only when active).""" diff --git a/solstone/think/journal_default.json b/solstone/think/journal_default.json index ce8ffb7b7..dbaeab7e5 100644 --- a/solstone/think/journal_default.json +++ b/solstone/think/journal_default.json @@ -65,6 +65,13 @@ }, "debug_show_throttled": false }, + "processing": { + "mode": "realtime", + "gate": { + "time_window": { "enabled": true, "start": "02:00", "end": "06:00" }, + "display_powersave": { "enabled": false } + } + }, "pairing": { "host_url": null }, diff --git a/solstone/think/processing.py b/solstone/think/processing.py new file mode 100644 index 000000000..cdc23dedd --- /dev/null +++ b/solstone/think/processing.py @@ -0,0 +1,362 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Deferred-processing settings and drain-gate read helpers.""" + +from __future__ import annotations + +import copy +import logging +import re +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any + +from solstone.think.utils import day_dirs, get_config + +logger = logging.getLogger(__name__) + +AWAITING_ANALYSIS_TEMPLATE = "{count} segments captured, awaiting analysis" +DRAIN_STATE_REALTIME = "realtime" +DRAIN_STATE_WINDOW_OPEN = "window_open" +DRAIN_STATE_WAITING = "waiting_for_window" +DRAIN_STATE_NO_CONDITION = "no_active_condition" + +_MISSING = object() +_MODES = frozenset({"realtime", "deferred"}) +_HHMM_RE = re.compile(r"^([01]\d|2[0-3]):([0-5]\d)$") + + +@dataclass(frozen=True) +class TimeWindowSettings: + enabled: bool + start: str + end: str + + +@dataclass(frozen=True) +class DisplayPowersaveSettings: + enabled: bool + + +@dataclass(frozen=True) +class GateSettings: + time_window: TimeWindowSettings + display_powersave: DisplayPowersaveSettings + + +@dataclass(frozen=True) +class ProcessingSettings: + mode: str + gate: GateSettings + + def to_dict(self) -> dict[str, Any]: + """Return the on-disk ``processing`` shape.""" + return { + "mode": self.mode, + "gate": { + "time_window": { + "enabled": self.gate.time_window.enabled, + "start": self.gate.time_window.start, + "end": self.gate.time_window.end, + }, + "display_powersave": { + "enabled": self.gate.display_powersave.enabled, + }, + }, + } + + +@dataclass(frozen=True) +class ConditionState: + enabled: bool + available: bool + open: bool + + +@dataclass(frozen=True) +class GateState: + open: bool + conditions: dict[str, ConditionState] + + +DEFAULT_PROCESSING = ProcessingSettings( + mode="realtime", + gate=GateSettings( + time_window=TimeWindowSettings( + enabled=True, + start="02:00", + end="06:00", + ), + display_powersave=DisplayPowersaveSettings(enabled=False), + ), +) + + +def parse_processing_settings(raw: object, *, strict: bool) -> ProcessingSettings: + """Parse processing settings with field defaults or strict validation.""" + if raw is None: + if strict: + _reject("processing", raw, strict) + raw = {} + elif not isinstance(raw, dict): + _reject("processing", raw, strict) + raw = {} + + raw_gate = raw.get("gate", _MISSING) + if raw_gate is _MISSING: + raw_gate = {} + elif not isinstance(raw_gate, dict): + _reject("gate", raw_gate, strict) + raw_gate = {} + + raw_time_window = raw_gate.get("time_window", _MISSING) + if raw_time_window is _MISSING: + raw_time_window = {} + elif not isinstance(raw_time_window, dict): + _reject("gate.time_window", raw_time_window, strict) + raw_time_window = {} + + raw_display_powersave = raw_gate.get("display_powersave", _MISSING) + if raw_display_powersave is _MISSING: + raw_display_powersave = {} + elif not isinstance(raw_display_powersave, dict): + _reject("gate.display_powersave", raw_display_powersave, strict) + raw_display_powersave = {} + + if strict: + _validate_known_keys("processing", raw, {"mode", "gate"}) + _validate_known_keys("gate", raw_gate, {"time_window", "display_powersave"}) + _validate_known_keys( + "gate.time_window", + raw_time_window, + {"enabled", "start", "end"}, + ) + _validate_known_keys( + "gate.display_powersave", + raw_display_powersave, + {"enabled"}, + ) + + return ProcessingSettings( + mode=_mode(raw.get("mode", _MISSING), strict), + gate=GateSettings( + time_window=TimeWindowSettings( + enabled=_bool( + "gate.time_window.enabled", + raw_time_window.get("enabled", _MISSING), + DEFAULT_PROCESSING.gate.time_window.enabled, + strict, + ), + start=_hhmm( + "start", + raw_time_window.get("start", _MISSING), + DEFAULT_PROCESSING.gate.time_window.start, + strict, + ), + end=_hhmm( + "end", + raw_time_window.get("end", _MISSING), + DEFAULT_PROCESSING.gate.time_window.end, + strict, + ), + ), + display_powersave=DisplayPowersaveSettings( + enabled=_bool( + "gate.display_powersave.enabled", + raw_display_powersave.get("enabled", _MISSING), + DEFAULT_PROCESSING.gate.display_powersave.enabled, + strict, + ) + ), + ), + ) + + +def load_processing_settings() -> ProcessingSettings: + """Load processing settings from journal config with default-at-read fallback.""" + return parse_processing_settings(get_config().get("processing"), strict=False) + + +def validate_processing_update( + existing: dict[str, Any], + updates: dict[str, Any], +) -> ProcessingSettings: + """Deep-merge a partial processing update and strictly validate it.""" + return parse_processing_settings(_deep_merge(existing, updates), strict=True) + + +def evaluate_time_window(window: TimeWindowSettings, now: datetime) -> ConditionState: + """Evaluate the local minute-resolution time-window condition.""" + now_min = now.hour * 60 + now.minute + start_min = _hhmm_minutes(window.start) + end_min = _hhmm_minutes(window.end) + + if start_min < end_min: + is_open = start_min <= now_min < end_min + elif start_min > end_min: + is_open = now_min >= start_min or now_min < end_min + else: + is_open = False + + 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_drain_gate(settings: ProcessingSettings, now: datetime) -> 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 + ), + } + return GateState( + open=any( + condition.enabled and condition.available and condition.open + for condition in conditions.values() + ), + conditions=conditions, + ) + + +def format_awaiting_analysis(count: int) -> str: + """Return fixed awaiting-analysis copy for a backlog count.""" + return AWAITING_ANALYSIS_TEMPLATE.format(count=count) + + +def read_last_drained_at() -> int | None: + """Return the latest completed daily drain marker mtime in epoch ms.""" + latest: float | None = None + for path in day_dirs().values(): + marker = Path(path) / "health" / "daily.updated" + try: + mtime = marker.stat().st_mtime + except FileNotFoundError: + continue + if latest is None or mtime > latest: + latest = mtime + if latest is None: + return None + return int(latest * 1000) + + +def derive_drain_state( + settings: ProcessingSettings, + gate_state: GateState, +) -> str: + """Return the stable drain-state token for current settings and gate state.""" + if settings.mode != "deferred": + return DRAIN_STATE_REALTIME + if gate_state.open: + return DRAIN_STATE_WINDOW_OPEN + if any( + condition.enabled and condition.available + for condition in gate_state.conditions.values() + ): + return DRAIN_STATE_WAITING + return DRAIN_STATE_NO_CONDITION + + +def _deep_merge(base: object, updates: dict[str, Any]) -> dict[str, Any]: + if not isinstance(base, dict): + base = {} + merged = copy.deepcopy(base) + for key, value in updates.items(): + if isinstance(value, dict) and isinstance(merged.get(key), dict): + merged[key] = _deep_merge(merged[key], value) + continue + merged[key] = copy.deepcopy(value) + return merged + + +def _warn_invalid(key: str, raw_value: object) -> None: + logger.warning( + "processing settings invalid key=%s value=%r; falling back to default", + key, + raw_value, + ) + + +def _reject(key: str, raw_value: object, strict: bool) -> None: + if raw_value is _MISSING: + return + if strict: + raise ValueError(f"{key} has invalid value: {raw_value!r}") + _warn_invalid(key, raw_value) + + +def _validate_known_keys( + key: str, + raw_value: dict[str, Any], + allowed: set[str], +) -> None: + for candidate in raw_value: + if candidate not in allowed: + raise ValueError(f"{key}.{candidate} is not a recognized setting") + + +def _bool(key: str, raw_value: object, default: bool, strict: bool) -> bool: + if raw_value is _MISSING: + return default + if isinstance(raw_value, bool): + return raw_value + _reject(key, raw_value, strict) + return default + + +def _mode(raw_value: object, strict: bool) -> str: + if raw_value is _MISSING: + return DEFAULT_PROCESSING.mode + if isinstance(raw_value, str) and raw_value in _MODES: + return raw_value + if strict: + raise ValueError("processing mode must be realtime or deferred") + _warn_invalid("mode", raw_value) + return DEFAULT_PROCESSING.mode + + +def _hhmm(field: str, raw_value: object, default: str, strict: bool) -> str: + if raw_value is _MISSING: + return default + if isinstance(raw_value, str) and _HHMM_RE.fullmatch(raw_value): + return raw_value + if strict: + raise ValueError(f"{field} must be HH:MM 24-hour time, got {raw_value!r}") + _warn_invalid(field, raw_value) + return default + + +def _hhmm_minutes(value: str) -> int: + hour, minute = value.split(":", 1) + return int(hour) * 60 + int(minute) + + +__all__ = [ + "AWAITING_ANALYSIS_TEMPLATE", + "DRAIN_STATE_NO_CONDITION", + "DRAIN_STATE_REALTIME", + "DRAIN_STATE_WAITING", + "DRAIN_STATE_WINDOW_OPEN", + "DEFAULT_PROCESSING", + "ConditionState", + "DisplayPowersaveSettings", + "GateSettings", + "GateState", + "ProcessingSettings", + "TimeWindowSettings", + "derive_drain_state", + "evaluate_display_powersave", + "evaluate_drain_gate", + "evaluate_time_window", + "format_awaiting_analysis", + "load_processing_settings", + "parse_processing_settings", + "read_last_drained_at", + "validate_processing_update", +] diff --git a/solstone/think/supervisor.py b/solstone/think/supervisor.py index b4a1c90df..44e7799b0 100644 --- a/solstone/think/supervisor.py +++ b/solstone/think/supervisor.py @@ -41,6 +41,7 @@ from solstone.think.catchup_state import ( ) 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.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 @@ -82,6 +83,7 @@ REACTIVE_TASK_CAPS = { } DEFAULT_THRESHOLD = 60 CHECK_INTERVAL = 30 +GATE_TICK_INTERVAL_S = 60 MAX_UPDATED_CATCHUP = 4 TEMPFAIL_DELAY = 15 # seconds to wait before retrying a tempfail exit CONVEY_READY_WINDOW_SECONDS = 60.0 @@ -176,6 +178,7 @@ _SERVICE_LIFECYCLE_VERBS = { # Global shutdown flag shutdown_requested = False _last_sync_tick: float = 0.0 +_last_gate_tick: float = 0.0 _last_sync_snapshot: "SyncCheckSnapshot | None" = None _sync_conflict_shutdown: bool = False # Supervisor identity (set in main() once ref is assigned) @@ -2335,6 +2338,14 @@ def _handle_segment_observed(message: dict) -> None: ) return + if load_processing_settings().mode == "deferred": + logging.info( + "Deferred mode: live segment %s/%s held for catchup drain; no live think", + day, + segment, + ) + return + stream = message.get("stream") # Update flush state — new segment resets the flush timer @@ -2380,6 +2391,9 @@ def _check_segment_flush(force: bool = False) -> None: if not last_ts or _flush_state["flushed"]: return + if load_processing_settings().mode == "deferred": + return + if not force and time.time() - last_ts < FLUSH_TIMEOUT: return @@ -2567,6 +2581,23 @@ def _run_sync_tick(now: float) -> bool: return True +def _run_gate_tick(now: float) -> None: + global _last_gate_tick + + if now - _last_gate_tick < GATE_TICK_INTERVAL_S: + return + _last_gate_tick = now + if _is_remote_mode: + return + settings = load_processing_settings() + if settings.mode != "deferred": + return + gate = evaluate_drain_gate(settings, datetime.now()) + if not gate.open: + return + run_catchup_drain() + + async def supervise( *, daily: bool = True, @@ -2578,9 +2609,11 @@ async def supervise( Monitors runner health, emits status, triggers daily processing, and checks scheduled agents. """ - global _last_sync_tick, _last_sync_snapshot, _sync_conflict_shutdown + global _last_gate_tick, _last_sync_tick + global _last_sync_snapshot, _sync_conflict_shutdown last_status_emit = 0.0 + _last_gate_tick = 0.0 _last_sync_tick = 0.0 _last_sync_snapshot = None _sync_conflict_shutdown = False @@ -2618,6 +2651,7 @@ async def supervise( # Check for daily processing (non-blocking, submits via task queue) if daily: handle_daily_tasks() + _run_gate_tick(now) # Check periodic task schedules (non-blocking, submits via callosum) if schedule: diff --git a/solstone/think/surfaces/health.py b/solstone/think/surfaces/health.py index 05784f36e..095923f0d 100644 --- a/solstone/think/surfaces/health.py +++ b/solstone/think/surfaces/health.py @@ -25,6 +25,13 @@ from solstone.think.activities import load_activity_records 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 +from solstone.think.processing import ( + derive_drain_state, + evaluate_drain_gate, + format_awaiting_analysis, + load_processing_settings, + read_last_drained_at, +) from solstone.think.surfaces import ledger from solstone.think.surfaces.types import ( CaptureHealth, @@ -499,10 +506,23 @@ def _build_segment_backlog_health() -> SegmentBacklogHealth: days_with_backlog = sum( 1 for completion in backlog.per_day.values() if completion.not_thought > 0 ) + settings = load_processing_settings() + gate = evaluate_drain_gate(settings, datetime.now()) + drain_state = derive_drain_state(settings, gate) + awaiting_total = backlog.not_sensed + backlog.not_thought + awaiting_text = ( + format_awaiting_analysis(awaiting_total) + if settings.mode == "deferred" + else None + ) return SegmentBacklogHealth( not_thought=backlog.not_thought, days_with_backlog=days_with_backlog, errors=backlog.errors, + not_sensed=backlog.not_sensed, + awaiting_analysis_text=awaiting_text, + last_drained_at=read_last_drained_at(), + drain_state=drain_state, ) diff --git a/solstone/think/surfaces/types.py b/solstone/think/surfaces/types.py index 40980258f..a07a6daf8 100644 --- a/solstone/think/surfaces/types.py +++ b/solstone/think/surfaces/types.py @@ -4,6 +4,8 @@ from dataclasses import dataclass from typing import Any +from solstone.think.processing import DRAIN_STATE_REALTIME + @dataclass(frozen=True) class ActivitySourceRef: @@ -123,6 +125,10 @@ class SegmentBacklogHealth: not_thought: int days_with_backlog: int errors: tuple[str, ...] + not_sensed: int = 0 + awaiting_analysis_text: str | None = None + last_drained_at: int | None = None + drain_state: str = DRAIN_STATE_REALTIME @dataclass(frozen=True) diff --git a/tests/test_health_call_parity.py b/tests/test_health_call_parity.py index e60eaf4bd..98931d660 100644 --- a/tests/test_health_call_parity.py +++ b/tests/test_health_call_parity.py @@ -11,6 +11,7 @@ import requests from typer.testing import CliRunner from solstone.think.convey_client import ConveyClient +from solstone.think.processing import DRAIN_STATE_REALTIME from solstone.think.surfaces import health as health_surface from solstone.think.tools.health import app from tests._baseline_harness import make_test_client, mark_setup_complete @@ -86,7 +87,15 @@ def _report() -> dict[str, object]: "ledger_stale_items_count": 0, "profile_entities_total": 0, }, - "segment_backlog": {"not_thought": 0, "days_with_backlog": 0, "errors": []}, + "segment_backlog": { + "not_thought": 0, + "days_with_backlog": 0, + "errors": [], + "not_sensed": 0, + "awaiting_analysis_text": None, + "last_drained_at": None, + "drain_state": DRAIN_STATE_REALTIME, + }, "notes": [], "provider_readiness": { "summary": { diff --git a/tests/test_processing.py b/tests/test_processing.py new file mode 100644 index 000000000..f64c99b84 --- /dev/null +++ b/tests/test_processing.py @@ -0,0 +1,290 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import logging +import os +from datetime import datetime + +import pytest + +from solstone.think.processing import ( + AWAITING_ANALYSIS_TEMPLATE, + DEFAULT_PROCESSING, + DRAIN_STATE_NO_CONDITION, + DRAIN_STATE_REALTIME, + DRAIN_STATE_WAITING, + DRAIN_STATE_WINDOW_OPEN, + ConditionState, + DisplayPowersaveSettings, + GateSettings, + GateState, + ProcessingSettings, + TimeWindowSettings, + derive_drain_state, + evaluate_display_powersave, + evaluate_drain_gate, + evaluate_time_window, + format_awaiting_analysis, + parse_processing_settings, + read_last_drained_at, + validate_processing_update, +) + +EXPECTED_DEFAULT = { + "mode": "realtime", + "gate": { + "time_window": {"enabled": True, "start": "02:00", "end": "06:00"}, + "display_powersave": {"enabled": False}, + }, +} + + +def _time_window( + *, + enabled: bool = True, + start: str = "02:00", + end: str = "06:00", +) -> TimeWindowSettings: + return TimeWindowSettings(enabled=enabled, start=start, end=end) + + +def _settings( + *, + mode: str = "deferred", + time_window: TimeWindowSettings | None = None, + display_powersave: DisplayPowersaveSettings | None = None, +) -> ProcessingSettings: + return ProcessingSettings( + mode=mode, + gate=GateSettings( + time_window=time_window or _time_window(), + display_powersave=display_powersave + or DisplayPowersaveSettings(enabled=False), + ), + ) + + +@pytest.mark.parametrize("raw", [None, {}]) +def test_parse_processing_settings_defaults(raw: object) -> None: + settings = parse_processing_settings(raw, strict=False) + + assert settings == DEFAULT_PROCESSING + assert settings.to_dict() == EXPECTED_DEFAULT + + +def test_parse_processing_settings_missing_section_is_quiet( + caplog: pytest.LogCaptureFixture, +) -> None: + caplog.set_level(logging.WARNING) + + settings = parse_processing_settings(None, strict=False) + + assert settings == DEFAULT_PROCESSING + assert caplog.records == [] + + +def test_strict_validation_rejects_bad_mode() -> None: + with pytest.raises( + ValueError, + match="processing mode must be realtime or deferred", + ): + parse_processing_settings({"mode": "batch"}, strict=True) + + +def test_strict_validation_rejects_non_bool_enabled() -> None: + with pytest.raises( + ValueError, + match="gate.time_window.enabled has invalid value: 'yes'", + ): + parse_processing_settings( + {"gate": {"time_window": {"enabled": "yes"}}}, + strict=True, + ) + + +def test_strict_validation_rejects_bad_start() -> None: + with pytest.raises( + ValueError, + match="start must be HH:MM 24-hour time, got '25:99'", + ): + parse_processing_settings( + {"gate": {"time_window": {"start": "25:99"}}}, + strict=True, + ) + + +def test_strict_validation_rejects_bad_end() -> None: + with pytest.raises(ValueError, match="end must be HH:MM 24-hour time"): + parse_processing_settings( + {"gate": {"time_window": {"end": "6:00"}}}, + strict=True, + ) + + +@pytest.mark.parametrize( + ("raw", "message"), + [ + ({"extra": True}, "processing.extra is not a recognized setting"), + ({"gate": {"extra": True}}, "gate.extra is not a recognized setting"), + ( + {"gate": {"time_window": {"extra": True}}}, + "gate.time_window.extra is not a recognized setting", + ), + ( + {"gate": {"display_powersave": {"extra": True}}}, + "gate.display_powersave.extra is not a recognized setting", + ), + ], +) +def test_strict_validation_rejects_unknown_keys( + raw: dict[str, object], + message: str, +) -> None: + with pytest.raises(ValueError, match=message): + parse_processing_settings(raw, strict=True) + + +def test_validate_processing_update_allows_deferred_without_conditions() -> None: + settings = validate_processing_update( + {}, + { + "mode": "deferred", + "gate": { + "time_window": {"enabled": False}, + "display_powersave": {"enabled": False}, + }, + }, + ) + + assert settings.mode == "deferred" + assert settings.gate.time_window.enabled is False + assert settings.gate.display_powersave.enabled is False + + +def test_validate_processing_update_preserves_nested_defaults() -> None: + settings = validate_processing_update( + DEFAULT_PROCESSING.to_dict(), + {"mode": "deferred"}, + ) + + assert settings.mode == "deferred" + assert settings.gate.time_window.start == "02:00" + assert settings.gate.time_window.end == "06:00" + + +def test_evaluate_time_window_standard_window_boundaries() -> None: + window = _time_window(start="02:00", end="06:00") + + assert evaluate_time_window(window, datetime(2026, 1, 1, 3, 30)).open is True + assert evaluate_time_window(window, datetime(2026, 1, 1, 1, 59)).open is False + assert evaluate_time_window(window, datetime(2026, 1, 1, 2, 0)).open is True + assert evaluate_time_window(window, datetime(2026, 1, 1, 6, 0)).open is False + + +def test_evaluate_time_window_midnight_wrap() -> None: + window = _time_window(start="22:00", end="06:00") + + assert evaluate_time_window(window, datetime(2026, 1, 1, 23, 0)).open is True + assert evaluate_time_window(window, datetime(2026, 1, 1, 5, 0)).open is True + assert evaluate_time_window(window, datetime(2026, 1, 1, 12, 0)).open is False + + +def test_evaluate_time_window_zero_width_is_closed() -> None: + window = _time_window(start="02:00", end="02:00") + + assert evaluate_time_window(window, datetime(2026, 1, 1, 2, 0)).open is False + + +def test_evaluate_drain_gate_uses_or_composition() -> None: + assert ( + evaluate_drain_gate( + _settings( + time_window=_time_window(enabled=True, start="02:00", end="06:00"), + display_powersave=DisplayPowersaveSettings(enabled=True), + ), + datetime(2026, 1, 1, 3, 0), + ).open + is True + ) + + assert ( + evaluate_drain_gate( + _settings( + time_window=_time_window(enabled=False, start="02:00", end="06:00"), + display_powersave=DisplayPowersaveSettings(enabled=True), + ), + datetime(2026, 1, 1, 3, 0), + ).open + is False + ) + + +def test_evaluate_display_powersave_is_inert() -> None: + state = evaluate_display_powersave(DisplayPowersaveSettings(enabled=True)) + + assert state.enabled is True + assert state.available is False + assert state.open is False + + +def test_derive_drain_state_tokens() -> None: + open_gate = GateState( + open=True, + conditions={"time_window": ConditionState(True, True, True)}, + ) + waiting_gate = GateState( + open=False, + conditions={"time_window": ConditionState(True, True, False)}, + ) + unavailable_gate = GateState( + open=False, + conditions={ + "time_window": ConditionState(False, True, False), + "display_powersave": ConditionState(True, False, False), + }, + ) + + assert derive_drain_state(_settings(mode="realtime"), open_gate) == ( + DRAIN_STATE_REALTIME + ) + assert derive_drain_state(_settings(mode="deferred"), open_gate) == ( + DRAIN_STATE_WINDOW_OPEN + ) + assert derive_drain_state(_settings(mode="deferred"), waiting_gate) == ( + DRAIN_STATE_WAITING + ) + assert derive_drain_state(_settings(mode="deferred"), unavailable_gate) == ( + DRAIN_STATE_NO_CONDITION + ) + + +def test_format_awaiting_analysis_uses_fixed_template() -> None: + assert format_awaiting_analysis(1) == AWAITING_ANALYSIS_TEMPLATE.format(count=1) + + +def test_read_last_drained_at_returns_none_without_markers( + tmp_path, + monkeypatch, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + + assert read_last_drained_at() is None + + +def test_read_last_drained_at_returns_latest_marker_mtime( + tmp_path, + monkeypatch, +) -> None: + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + older = tmp_path / "chronicle" / "20260101" / "health" / "daily.updated" + newer = tmp_path / "chronicle" / "20260102" / "health" / "daily.updated" + older.parent.mkdir(parents=True) + newer.parent.mkdir(parents=True) + older.write_text("", encoding="utf-8") + newer.write_text("", encoding="utf-8") + os.utime(older, (1_700_000_000, 1_700_000_000)) + os.utime(newer, (1_700_000_500, 1_700_000_500)) + + assert read_last_drained_at() == 1_700_000_500_000 diff --git a/tests/test_sense.py b/tests/test_sense.py index d8f6d8e39..0875e08fb 100644 --- a/tests/test_sense.py +++ b/tests/test_sense.py @@ -17,6 +17,12 @@ import pytest from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED, WATCHDOG_TIMEOUT from solstone.observe.sense import FileSensor, HandlerProcess, QueuedItem +from solstone.think.processing import ( + DisplayPowersaveSettings, + GateSettings, + ProcessingSettings, + TimeWindowSettings, +) from solstone.think.runner import DailyLogWriter as ProcessLogWriter from solstone.think.runner import _format_log_line @@ -82,6 +88,20 @@ def make_segment_file( return file_path +def _processing_settings(mode: str) -> ProcessingSettings: + return ProcessingSettings( + mode=mode, + gate=GateSettings( + time_window=TimeWindowSettings( + enabled=True, + start="02:00", + end="06:00", + ), + display_powersave=DisplayPowersaveSettings(enabled=False), + ), + ) + + # --- QueuedItem Tests --- @@ -969,6 +989,122 @@ def test_file_sensor_handle_callosum_message_invalid_event(tmp_path): mock_handle.assert_not_called() +def test_file_sensor_deferred_live_observing_suppresses_handlers( + tmp_path, monkeypatch, mock_callosum +): + import solstone.observe.sense as sense_module + from solstone.think.callosum import CallosumConnection + + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + monkeypatch.setattr( + sense_module, + "load_processing_settings", + lambda: _processing_settings("deferred"), + ) + + segment_dir = tmp_path / "chronicle" / "20250101" / "default" / "143022_300" + segment_dir.mkdir(parents=True) + audio_file = segment_dir / "audio.flac" + audio_file.write_text("audio content") + + sensor = FileSensor(tmp_path) + sensor.register("*.flac", "transcribe", ["echo", "{file}"]) + emitted_events = [] + sensor.callosum = CallosumConnection() + sensor.callosum.start(callback=lambda msg: emitted_events.append(msg)) + + with patch.object(sensor, "_handle_file") as mock_handle: + sensor._handle_callosum_message( + { + "tract": "observe", + "event": "observing", + "day": "20250101", + "stream": "default", + "segment": "143022_300", + "files": ["audio.flac"], + } + ) + + mock_handle.assert_not_called() + observed_events = [ + event + for event in emitted_events + if event.get("tract") == "observe" and event.get("event") == "observed" + ] + assert len(observed_events) == 1 + assert observed_events[0].get("day") == "20250101" + assert observed_events[0].get("segment") == "143022_300" + assert (tmp_path / "chronicle" / "20250101" / "health" / "stream.updated").exists() + assert audio_file.exists() + + +def test_file_sensor_realtime_live_observing_dispatches_handler(tmp_path, monkeypatch): + import solstone.observe.sense as sense_module + + monkeypatch.setattr( + sense_module, + "load_processing_settings", + lambda: _processing_settings("realtime"), + ) + + segment_dir = tmp_path / "chronicle" / "20250101" / "default" / "143022_300" + segment_dir.mkdir(parents=True) + audio_file = segment_dir / "audio.flac" + audio_file.write_text("audio content") + + sensor = FileSensor(tmp_path) + sensor.register("*.flac", "transcribe", ["echo", "{file}"]) + + with patch.object(sensor, "_handle_file") as mock_handle: + sensor._handle_callosum_message( + { + "tract": "observe", + "event": "observing", + "day": "20250101", + "stream": "default", + "segment": "143022_300", + "files": ["audio.flac"], + } + ) + + mock_handle.assert_called_once() + assert mock_handle.call_args.args[0] == audio_file + + +def test_file_sensor_deferred_batch_observing_dispatches_handler(tmp_path, monkeypatch): + import solstone.observe.sense as sense_module + + monkeypatch.setattr( + sense_module, + "load_processing_settings", + lambda: _processing_settings("deferred"), + ) + + segment_dir = tmp_path / "chronicle" / "20250101" / "default" / "143022_300" + segment_dir.mkdir(parents=True) + audio_file = segment_dir / "audio.flac" + audio_file.write_text("audio content") + + sensor = FileSensor(tmp_path) + sensor.register("*.flac", "transcribe", ["echo", "{file}"]) + + with patch.object(sensor, "_handle_file") as mock_handle: + sensor._handle_callosum_message( + { + "tract": "observe", + "event": "observing", + "day": "20250101", + "stream": "default", + "segment": "143022_300", + "files": ["audio.flac"], + "batch": True, + } + ) + + mock_handle.assert_called_once() + assert mock_handle.call_args.args[0] == audio_file + + @patch("solstone.think.runner._current_day") def test_file_sensor_segment_observed_includes_day( mock_day, tmp_path, monkeypatch, mock_callosum diff --git a/tests/test_settings_call_parity.py b/tests/test_settings_call_parity.py index 13a662672..fde0805a6 100644 --- a/tests/test_settings_call_parity.py +++ b/tests/test_settings_call_parity.py @@ -7,6 +7,7 @@ import json from datetime import datetime, timezone from pathlib import Path from typing import Any +from urllib.parse import urlsplit import pytest from typer.testing import CliRunner @@ -34,6 +35,27 @@ class _FixedDateTime: return datetime(2026, 4, 17, 12, 0, tzinfo=tz or timezone.utc) +class _CaptureSession: + def __init__(self, inner): + self.inner = inner + self.gets: list[str] = [] + self.posts: list[tuple[str, dict[str, Any] | None]] = [] + + def get(self, url: str): + self.gets.append(urlsplit(url).path) + return self.inner.get(url) + + def post(self, url: str, json=None): + self.posts.append((urlsplit(url).path, json)) + return self.inner.post(url, json=json) + + def put(self, url: str, json=None): + return self.inner.put(url, json=json) + + def delete(self, url: str): + return self.inner.delete(url) + + @pytest.fixture(autouse=True) def _settings_client(journal_copy: Path, monkeypatch: pytest.MonkeyPatch) -> None: for key in API_ENV_KEYS: @@ -257,6 +279,67 @@ def test_transcribe_setters(journal_copy: Path) -> None: assert payload["noise_upgrade"] is False +def test_processing_show_uses_effective_settings_endpoint( + journal_copy: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + session = _CaptureSession(make_test_client(journal_copy)) + client = ConveyClient(session=session, base_url="") + monkeypatch.setattr(settings_call, "get_client", lambda: client) + + result = runner.invoke(settings_call.app, ["processing", "show"]) + + assert result.exit_code == 0 + assert session.gets == ["/app/settings/api/processing"] + payload = json.loads(result.stdout) + assert payload["mode"] == "realtime" + assert payload["gate"]["time_window"]["start"] == "02:00" + assert result.stderr == "" + + +def test_processing_set_posts_only_supplied_fields( + journal_copy: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + session = _CaptureSession(make_test_client(journal_copy)) + client = ConveyClient(session=session, base_url="") + monkeypatch.setattr(settings_call, "get_client", lambda: client) + + result = runner.invoke( + settings_call.app, + ["processing", "set", "--mode", "deferred", "--window-start", "03:00"], + ) + + assert result.exit_code == 0 + assert session.posts == [ + ( + "/app/settings/api/config", + { + "section": "processing", + "data": { + "mode": "deferred", + "gate": {"time_window": {"start": "03:00"}}, + }, + }, + ) + ] + payload = json.loads(result.stdout) + assert payload["mode"] == "deferred" + assert payload["gate"]["time_window"]["start"] == "03:00" + assert result.stderr == "" + + +def test_processing_set_surfaces_invalid_config_detail() -> None: + result = runner.invoke( + settings_call.app, + ["processing", "set", "--mode", "bogus"], + ) + + assert result.exit_code == 1 + assert result.stdout == "" + assert result.stderr == "processing mode must be realtime or deferred\n" + + def test_identity_and_observer_setters(journal_copy: Path) -> None: bad_pronouns = runner.invoke( settings_call.app, diff --git a/tests/test_supervisor.py b/tests/test_supervisor.py index 8f5aed3a2..6ad5dfd3f 100644 --- a/tests/test_supervisor.py +++ b/tests/test_supervisor.py @@ -20,6 +20,13 @@ from unittest.mock import MagicMock import psutil import pytest +from solstone.think.processing import ( + DisplayPowersaveSettings, + GateSettings, + ProcessingSettings, + TimeWindowSettings, +) + def test_sd_notify_no_socket_is_noop(monkeypatch): from solstone.think.supervisor import _sd_notify @@ -619,6 +626,20 @@ class _CaptureTaskQueue: self.submissions.append({"cmd": cmd, "day": day}) +def _supervisor_processing_settings(mode: str) -> ProcessingSettings: + return ProcessingSettings( + mode=mode, + gate=GateSettings( + time_window=TimeWindowSettings( + enabled=True, + start="02:00", + end="06:00", + ), + display_powersave=DisplayPowersaveSettings(enabled=False), + ), + ) + + def test_handle_segment_observed_live_command_marks_live(monkeypatch): mod = importlib.import_module("solstone.think.supervisor") capture = _CaptureTaskQueue() @@ -726,6 +747,125 @@ def test_handle_segment_observed_live_stream_command(monkeypatch): assert cmd[stream_index + 1] == "archon" +def test_handle_segment_observed_deferred_live_submits_nothing(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + capture = _CaptureTaskQueue() + monkeypatch.setattr(mod, "_task_queue", capture) + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("deferred"), + ) + mod._flush_state.update( + { + "last_segment_ts": 123.0, + "day": "20260526", + "segment": "110000_300", + "stream": "archon", + "flushed": True, + } + ) + before = dict(mod._flush_state) + + mod._handle_segment_observed( + { + "tract": "observe", + "event": "observed", + "day": "20260527", + "segment": "120000_300", + } + ) + + assert capture.submissions == [] + assert dict(mod._flush_state) == before + + +def test_handle_segment_observed_reads_processing_mode_fresh(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + capture = _CaptureTaskQueue() + monkeypatch.setattr(mod, "_task_queue", capture) + modes = ["realtime", "deferred"] + + def load_settings(): + return _supervisor_processing_settings(modes.pop(0)) + + monkeypatch.setattr(mod, "load_processing_settings", load_settings) + + mod._handle_segment_observed( + { + "tract": "observe", + "event": "observed", + "day": "20260527", + "segment": "120000_300", + } + ) + after_realtime = dict(mod._flush_state) + mod._handle_segment_observed( + { + "tract": "observe", + "event": "observed", + "day": "20260527", + "segment": "120500_300", + } + ) + + assert len(capture.submissions) == 1 + assert capture.submissions[0]["cmd"][-1] == "--live" + assert dict(mod._flush_state) == after_realtime + + +def test_check_segment_flush_deferred_guard_submits_nothing(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + capture = _CaptureTaskQueue() + monkeypatch.setattr(mod, "_task_queue", capture) + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("deferred"), + ) + mod._flush_state.update( + { + "last_segment_ts": 1.0, + "day": "20260527", + "segment": "120000_300", + "stream": None, + "flushed": False, + } + ) + + mod._check_segment_flush() + + assert capture.submissions == [] + assert mod._flush_state["flushed"] is False + + +def test_check_segment_flush_realtime_submits(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + capture = _CaptureTaskQueue() + monkeypatch.setattr(mod, "_task_queue", capture) + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("realtime"), + ) + monkeypatch.setattr(mod.time, "time", lambda: 10_000.0) + mod._flush_state.update( + { + "last_segment_ts": 1.0, + "day": "20260527", + "segment": "120000_300", + "stream": None, + "flushed": False, + } + ) + + mod._check_segment_flush() + + assert len(capture.submissions) == 1 + assert "--flush" in capture.submissions[0]["cmd"] + assert mod._flush_state["flushed"] is True + + def test_task_queue_daily_and_segment_run_independently(monkeypatch): mod = importlib.import_module("solstone.think.supervisor") queue = _fresh_task_queue(mod) @@ -1786,6 +1926,90 @@ def test_startup_catchup_drain_reconciles_before_drain(monkeypatch): assert order == ["reconcile", "drain"] +def test_run_gate_tick_deferred_open_runs_catchup_drain(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + drain = MagicMock() + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("deferred"), + ) + monkeypatch.setattr( + mod, + "evaluate_drain_gate", + lambda settings, now: SimpleNamespace(open=True), + ) + monkeypatch.setattr(mod, "run_catchup_drain", drain) + mod._last_gate_tick = 0.0 + + mod._run_gate_tick(60.0) + + drain.assert_called_once_with() + + +def test_run_gate_tick_deferred_closed_skips_catchup_drain(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + drain = MagicMock() + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("deferred"), + ) + monkeypatch.setattr( + mod, + "evaluate_drain_gate", + lambda settings, now: SimpleNamespace(open=False), + ) + monkeypatch.setattr(mod, "run_catchup_drain", drain) + mod._last_gate_tick = 0.0 + + mod._run_gate_tick(60.0) + + drain.assert_not_called() + + +def test_run_gate_tick_realtime_skips_catchup_drain(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + drain = MagicMock() + evaluate = MagicMock(return_value=SimpleNamespace(open=True)) + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("realtime"), + ) + 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) + + evaluate.assert_not_called() + drain.assert_not_called() + + +def test_run_gate_tick_throttles_catchup_drain(monkeypatch): + mod = importlib.import_module("solstone.think.supervisor") + drain = MagicMock() + monkeypatch.setattr( + mod, + "load_processing_settings", + lambda: _supervisor_processing_settings("deferred"), + ) + monkeypatch.setattr( + mod, + "evaluate_drain_gate", + lambda settings, now: SimpleNamespace(open=True), + ) + monkeypatch.setattr(mod, "run_catchup_drain", drain) + mod._last_gate_tick = 0.0 + + mod._run_gate_tick(60.0) + mod._run_gate_tick(119.0) + mod._run_gate_tick(121.0) + + assert drain.call_count == 2 + + 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 db9273bc3..b1686db8b 100644 --- a/tests/test_surfaces_health.py +++ b/tests/test_surfaces_health.py @@ -16,6 +16,19 @@ from typer.testing import CliRunner 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 ( + DRAIN_STATE_NO_CONDITION, + DRAIN_STATE_REALTIME, + DRAIN_STATE_WAITING, + DRAIN_STATE_WINDOW_OPEN, + ConditionState, + DisplayPowersaveSettings, + GateSettings, + GateState, + ProcessingSettings, + TimeWindowSettings, + format_awaiting_analysis, +) from solstone.think.surfaces import health as health_surface from tests._baseline_harness import make_test_client @@ -191,6 +204,7 @@ def _write_pipeline_log( def _segment_backlog( per_day_counts: dict[str, int], *, + not_sensed: int = 0, errors: tuple[str, ...] = (), ) -> SegmentBacklog: per_day = { @@ -205,13 +219,30 @@ def _segment_backlog( return SegmentBacklog( days=tuple(per_day_counts), not_thought=sum(per_day_counts.values()), - not_sensed=0, - total=sum(completion.total for completion in per_day.values()), + not_sensed=not_sensed, + total=max( + sum(completion.total for completion in per_day.values()), + sum(per_day_counts.values()) + not_sensed, + ), per_day=per_day, errors=errors, ) +def _processing_settings(mode: str) -> ProcessingSettings: + return ProcessingSettings( + mode=mode, + gate=GateSettings( + time_window=TimeWindowSettings( + enabled=True, + start="02:00", + end="06:00", + ), + display_powersave=DisplayPowersaveSettings(enabled=False), + ), + ) + + def _clear_readiness_snapshot() -> dict[str, object]: return { "summary": { @@ -531,6 +562,144 @@ def test_segment_backlog_health_preserves_errors(tmp_path, monkeypatch): assert report.segment_backlog.errors == ("20260101",) +@pytest.mark.parametrize( + ("not_sensed", "not_thought", "expected_count"), + [(3, 0, 3), (2, 1, 3)], +) +def test_segment_backlog_deferred_awaiting_analysis_uses_unsensed( + monkeypatch, + not_sensed: int, + not_thought: int, + expected_count: int, +) -> None: + per_day = {"20260410": not_thought} if not_thought else {} + monkeypatch.setattr( + health_surface, + "read_segment_backlog", + lambda: _segment_backlog(per_day, not_sensed=not_sensed), + ) + monkeypatch.setattr( + health_surface, + "load_processing_settings", + lambda: _processing_settings("deferred"), + ) + monkeypatch.setattr( + health_surface, + "evaluate_drain_gate", + lambda settings, now: GateState(open=False, conditions={}), + ) + monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + + backlog = health_surface._build_segment_backlog_health() + + assert backlog.not_sensed == not_sensed + assert backlog.not_thought == not_thought + assert backlog.awaiting_analysis_text == format_awaiting_analysis(expected_count) + + +def test_segment_backlog_realtime_omits_awaiting_analysis_text(monkeypatch) -> None: + monkeypatch.setattr( + health_surface, + "read_segment_backlog", + lambda: _segment_backlog({"20260410": 2}, not_sensed=3), + ) + monkeypatch.setattr( + health_surface, + "load_processing_settings", + lambda: _processing_settings("realtime"), + ) + monkeypatch.setattr( + health_surface, + "evaluate_drain_gate", + lambda settings, now: GateState(open=True, conditions={}), + ) + monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + + backlog = health_surface._build_segment_backlog_health() + + assert backlog.awaiting_analysis_text is None + assert backlog.drain_state == DRAIN_STATE_REALTIME + + +@pytest.mark.parametrize( + ("gate", "expected_state"), + [ + ( + GateState( + open=True, + conditions={"time_window": ConditionState(True, True, True)}, + ), + DRAIN_STATE_WINDOW_OPEN, + ), + ( + GateState( + open=False, + conditions={"time_window": ConditionState(True, True, False)}, + ), + DRAIN_STATE_WAITING, + ), + ( + GateState( + open=False, + conditions={ + "time_window": ConditionState(False, True, False), + "display_powersave": ConditionState(True, False, False), + }, + ), + DRAIN_STATE_NO_CONDITION, + ), + ], +) +def test_segment_backlog_drain_state_tokens(monkeypatch, gate, expected_state) -> 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: gate, + ) + monkeypatch.setattr(health_surface, "read_last_drained_at", lambda: None) + + backlog = health_surface._build_segment_backlog_health() + + assert backlog.drain_state == expected_state + + +def test_segment_backlog_last_drained_at_passes_through(monkeypatch) -> None: + monkeypatch.setattr( + health_surface, + "read_segment_backlog", + lambda: _segment_backlog({}), + ) + monkeypatch.setattr( + health_surface, + "load_processing_settings", + lambda: _processing_settings("realtime"), + ) + monkeypatch.setattr( + health_surface, + "evaluate_drain_gate", + lambda settings, now: GateState(open=False, conditions={}), + ) + monkeypatch.setattr( + health_surface, + "read_last_drained_at", + lambda: 1_700_000_000_000, + ) + + backlog = health_surface._build_segment_backlog_health() + + assert backlog.last_drained_at == 1_700_000_000_000 + + def test_silent_facet_note_ladder_thresholds(tmp_path, monkeypatch): _configure_env(tmp_path, monkeypatch) _set_now(monkeypatch, _utc_dt("20260410"))