"""Persistent storage for dynamic maintenance state. Frequently-changing data (history, last_performed, trigger runtime, adaptive config) lives here instead of in ConfigEntry.data to avoid excessive writes to ``core.config_entries`` which wears out SD cards. Static task configuration (name, type, interval, trigger thresholds, etc.) remains in ConfigEntry.data and is only written on explicit user edits. """ from __future__ import annotations import logging from collections.abc import Mapping from datetime import UTC, datetime from typing import Any from homeassistant.core import HomeAssistant from homeassistant.helpers.storage import Store from .const import CONF_TASKS, DEFAULT_MAX_HISTORY_ENTRIES, DOMAIN, LIFECYCLE_HISTORY_TYPES from .helpers.dates import local_date_from_iso from .helpers.parts import round_qty from .helpers.pause import write_anchor _LOGGER = logging.getLogger(__name__) STORE_VERSION = 1 STORE_KEY_PREFIX = f"{DOMAIN}" STORE_SAVE_DELAY = 60.0 # seconds — debounce writes to protect SD cards # Fields that migrate from ConfigEntry.data to Store _DYNAMIC_TASK_FIELDS = ("last_performed", "last_planned_due", "due_override", "history", "adaptive_config", "phase_cursor") # Presentation-state fields that SPLIT/migrate into the Store like the dynamic # fields but are deliberately NOT overlaid by merge_task_data: merged dicts feed # MaintenanceTask.from_dict all over the coordinator and the model never needs # these — helpers.aggregate.merged_tasks overlays them explicitly for the read # surfaces instead. Split-listing them lets a backup import (which lands them in # entry.data) carry them into the fresh entry's Store on first setup. _SPLIT_ONLY_TASK_FIELDS = ("checklist_progress",) # Flat trigger_config keys that should move to Store trigger_runtime _LEGACY_TRIGGER_RUNTIME_KEYS = ( "trigger_baseline_value", "trigger_accumulated_seconds", "trigger_change_count", "trigger_on_since", ) # Flat legacy key -> the per-entity ``_trigger_state`` key the trigger classes # restore from (entity/triggers/_inject_per_entity_state). _LEGACY_TO_PER_ENTITY_KEY = { "trigger_baseline_value": "baseline_value", "trigger_accumulated_seconds": "accumulated_seconds", "trigger_change_count": "change_count", "trigger_on_since": "on_since", } def legacy_runtime_to_trigger_state(trigger_config: Mapping[str, Any], legacy: Mapping[str, Any]) -> dict[str, Any]: """Shape ``trigger_runtime_legacy`` (flat keys) like a per-entity ``_trigger_state``. The flat keys are split off the static config on first setup and parked under ``trigger_runtime_legacy`` — but the triggers only ever read the per-entity shape ``{entity_id: {baseline_value: …}}``. Handing them the flat dict verbatim meant an adopted "last service was at 20 000 km" baseline (#102), an imported runtime total or a hand-edited counter baseline was silently replaced by the current reading on the first start. Keyed by the first entity: legacy configs only ever had one. """ if legacy and all(isinstance(v, Mapping) for v in legacy.values()): # Already per-entity (defensive: nothing writes this shape today). return dict(legacy) from .entity.triggers import primary_entity_id # The shared "which entity does this trigger watch" rule (DRY audit # 2026-09-26 B) — a compound trigger has none and gets nothing. first = primary_entity_id(dict(trigger_config)) per_entity = { _LEGACY_TO_PER_ENTITY_KEY[k]: v for k, v in legacy.items() if k in _LEGACY_TO_PER_ENTITY_KEY and v is not None } if not first or not per_entity: return {} return {first: per_entity} class MaintenanceStore: """Manages dynamic state storage for a single maintenance object. Each maintenance object config entry gets its own store file at ``.storage/maintenance_supporter.``. """ def __init__(self, hass: HomeAssistant, entry_id: str) -> None: """Initialize the store.""" self._store: Store[dict[str, Any]] = Store(hass, STORE_VERSION, f"{STORE_KEY_PREFIX}.{entry_id}") self._data: dict[str, Any] = {"version": STORE_VERSION, "tasks": {}} # ------------------------------------------------------------------ # Load / Save / Remove # ------------------------------------------------------------------ async def async_load(self) -> dict[str, Any] | None: """Load store data. Returns None if no store file exists.""" raw = await self._store.async_load() if raw is not None: self._data = self._sanitize_loaded(raw) return raw @staticmethod def _sanitize_loaded(raw: Any) -> dict[str, Any]: """Coerce loaded data to the expected structure (journey I2). HA core already quarantines syntactically corrupt store files, but a hand-edited or partially-written file can still be valid JSON with the wrong SHAPE — and every accessor here assumes dicts and lists. Broken parts must degrade to "never performed" instead of bricking the entry at boot with an AttributeError. """ if not isinstance(raw, dict): _LOGGER.warning( "Store payload is %s, expected dict — resetting dynamic state", type(raw).__name__, ) return {"version": STORE_VERSION, "tasks": {}} data = dict(raw) tasks = data.get("tasks") if not isinstance(tasks, dict): if tasks is not None: _LOGGER.warning( "Store 'tasks' is %s, expected dict — resetting dynamic state", type(tasks).__name__, ) data["tasks"] = {} return data clean: dict[str, Any] = {} for tid, state in tasks.items(): if not isinstance(state, dict): _LOGGER.warning( "Dropping malformed store state for task %s (%s)", tid, type(state).__name__, ) continue state = dict(state) if "history" in state and not isinstance(state["history"], list): _LOGGER.warning("Resetting malformed history for task %s", tid) state["history"] = [] for key in ("trigger_runtime", "adaptive_config", "checklist_progress"): if key in state and not isinstance(state[key], dict): _LOGGER.warning("Dropping malformed %s for task %s", key, tid) del state[key] # Element level — the container checks above are not enough: # a non-dict history entry crashes the statistics/history readers, # a non-dict per-entity runtime crashes the trigger restore, and a # non-string date marker crashes date.fromisoformat at read time. # Drop the bad elements, keep the rest (never the whole task). if "history" in state: history = state["history"] kept = [h for h in history if isinstance(h, dict)] if len(kept) != len(history): _LOGGER.warning("Dropping %d malformed history entries for task %s", len(history) - len(kept), tid) state["history"] = kept # A non-string timestamp (an epoch number from an imported # backup) broke every "latest entry" comparison — completing # the task raised TypeError (bug audit 2026-09-27). Epoch # numbers become ISO stamps, anything else is dropped. fixed = [_coerce_history_timestamp(h) for h in state["history"]] if any(a is not b for a, b in zip(fixed, state["history"], strict=True)): _LOGGER.warning("Repaired malformed history timestamps for task %s", tid) state["history"] = fixed if "trigger_runtime" in state: runtime = state["trigger_runtime"] kept_rt = {eid: rt for eid, rt in runtime.items() if isinstance(rt, dict)} if len(kept_rt) != len(runtime): _LOGGER.warning("Dropping malformed trigger_runtime entries for task %s", tid) state["trigger_runtime"] = kept_rt for key in ("last_performed", "last_planned_due", "due_override"): if key in state and state[key] is not None and not isinstance(state[key], str): _LOGGER.warning("Dropping malformed %s for task %s", key, tid) del state[key] clean[tid] = state data["tasks"] = clean # Spare-part stock counts ({part_id: {"stock": int}}) — same shape # discipline as tasks: a malformed section degrades to "untracked". parts = data.get("parts") if parts is not None and not isinstance(parts, dict): _LOGGER.warning("Store 'parts' is %s, expected dict — resetting stock state", type(parts).__name__) data["parts"] = {} elif isinstance(parts, dict): data["parts"] = { pid: dict(state) for pid, state in parts.items() if isinstance(state, dict) } return data def async_delay_save(self) -> None: """Schedule a debounced save (coalesces rapid writes).""" self._store.async_delay_save(self._get_data, STORE_SAVE_DELAY) async def async_save(self) -> None: """Immediate save (for migration or critical writes).""" await self._store.async_save(self._data) async def async_remove(self) -> None: """Delete the store file (called when config entry is removed).""" await self._store.async_remove() def _get_data(self) -> dict[str, Any]: """Return current data for save callback.""" return self._data # ------------------------------------------------------------------ # Task State Accessors # ------------------------------------------------------------------ def _ensure_task(self, task_id: str) -> dict[str, Any]: """Ensure a task state dict exists and return it.""" tasks: dict[str, Any] = self._data.setdefault("tasks", {}) if task_id not in tasks: tasks[task_id] = {} result: dict[str, Any] = tasks[task_id] return result def get_task_state(self, task_id: str) -> dict[str, Any]: """Return the dynamic state for a task (or empty dict).""" tasks: dict[str, Any] = self._data.get("tasks", {}) result: dict[str, Any] = tasks.get(task_id, {}) return result def init_task(self, task_id: str, last_performed: str | None = None) -> None: """Initialize state for a newly created task.""" state = self._ensure_task(task_id) if last_performed is not None: state["last_performed"] = last_performed state.setdefault("history", []) def remove_task(self, task_id: str) -> None: """Remove all state for a deleted task.""" self._data.get("tasks", {}).pop(task_id, None) def put_task_state(self, task_id: str, state: dict[str, Any]) -> None: """Replace a task's dynamic state wholesale — the task/move path carries history, last_performed, adaptive config and trigger runtime from one object's Store to another's.""" self._data.setdefault("tasks", {})[task_id] = dict(state) def update_task_state(self, task_id: str, **fields: Any) -> None: """Write plain scalar fields of a task's dynamic state; a ``None`` value removes the key. The public door for the few callers that used to reach into ``_ensure_task`` (the coordinator's cycle modifiers, the reference-number counter).""" state = self._ensure_task(task_id) for key, value in fields.items(): if value is None: state.pop(key, None) else: state[key] = value def all_task_states(self) -> dict[str, dict[str, Any]]: """Every task's dynamic state, keyed by task id — the LIVE dicts, for in-place rewrites that then save (entity renames).""" tasks: dict[str, dict[str, Any]] = self._data.setdefault("tasks", {}) return tasks # --- in-cycle checklist progress (#73) ----------------------------------- def set_checklist_progress(self, task_id: str, state: dict[str, bool]) -> None: """Replace the in-cycle checklist ticks (keys = item texts). Keyed by item TEXT, not index: reordering the checklist keeps the ticks with their steps, and renaming a step deliberately drops its tick (the wording changed — re-confirm it). An empty dict clears. """ task_state = self._ensure_task(task_id) if state: task_state["checklist_progress"] = dict(state) else: task_state.pop("checklist_progress", None) def clear_checklist_progress(self, task_id: str) -> None: """Drop the in-cycle ticks — a completed/skipped cycle starts fresh.""" self._data.get("tasks", {}).get(task_id, {}).pop("checklist_progress", None) def prune_orphans(self, valid_task_ids: set[str]) -> int: """Drop task states whose ids are not in *valid_task_ids*. Boot-time reconciliation for the entry.data <-> Store split (journey I1): a crash between the ConfigEntry write and the Store save on a task deletion leaves the dynamic state orphaned forever — the merge only iterates entry.data tasks, so orphans are invisible but accumulate. Returns the number of pruned states; the caller decides whether to save. """ tasks = self._data.get("tasks", {}) orphaned = [tid for tid in tasks if tid not in valid_task_ids] for tid in orphaned: del tasks[tid] return len(orphaned) # --- spare-part stock (mutable count; static part defs live in entry.data) --- def get_part_stock(self, part_id: str) -> float | None: """On-hand count for a part, or None when inventory isn't tracked.""" state = self._data.get("parts", {}).get(part_id, {}) stock = state.get("stock") return stock if isinstance(stock, (int, float)) and not isinstance(stock, bool) else None def all_part_stocks(self) -> dict[str, float | None]: """{part_id: stock} for every tracked part.""" return {pid: self.get_part_stock(pid) for pid in self._data.get("parts", {})} def set_part_stock(self, part_id: str, stock: float | None) -> None: """Set (or untrack with None) a part's on-hand count, clamped to [0, MAX_PART_STOCK]. The WS schemas cap each INPUT at MAX_PART_STOCK, but additive paths (restock delta / buy-task completion) could push the SUM past it — this chokepoint clamp keeps every writer honest.""" from .helpers.parts import MAX_PART_STOCK parts: dict[str, Any] = self._data.setdefault("parts", {}) if stock is None: parts.pop(part_id, None) else: parts.setdefault(part_id, {})["stock"] = min(max(0, round_qty(stock)), MAX_PART_STOCK) def remove_part(self, part_id: str) -> None: """Drop a deleted part's stock state.""" self._data.get("parts", {}).pop(part_id, None) def prune_part_orphans(self, valid_part_ids: set[str]) -> int: """Drop stock states whose part ids no longer exist (journey-I1 twin).""" parts = self._data.get("parts", {}) orphaned = [pid for pid in parts if pid not in valid_part_ids] for pid in orphaned: del parts[pid] return len(orphaned) # --- last_performed --- def get_last_performed(self, task_id: str) -> str | None: """Return last_performed date string (or None).""" return self.get_task_state(task_id).get("last_performed") def set_last_performed(self, task_id: str, date_str: str) -> None: """Set last_performed date string (modifiers untouched — the coordinator's ``_persist_dynamic_state`` writes those itself).""" self._ensure_task(task_id)["last_performed"] = date_str def set_anchor(self, task_id: str, date_str: str | None, *, clear_modifiers: bool = True) -> None: """Move (or with ``None`` clear) a task's ``last_performed`` anchor. Every anchor poke outside the coordinator goes through here: the history edit/delete re-derivation, the dialog's "last performed" field, unarchive/resume. With ``clear_modifiers`` (the default) the cycle's per-occurrence modifiers (``due_override``, ``last_planned_due``) go too — see :func:`helpers.pause.write_anchor`. """ write_anchor(self._ensure_task(task_id), date_str, clear_modifiers=clear_modifiers) def set_phase_cursor(self, task_id: str, cursor: int) -> None: """Set the phase cursor (#139) — which cycle step is due next.""" self._ensure_task(task_id)["phase_cursor"] = cursor # --- history --- def get_history(self, task_id: str) -> list[dict[str, Any]]: """Return the history list for a task.""" result: list[dict[str, Any]] = self.get_task_state(task_id).get("history", []) return result def append_history(self, task_id: str, entry: dict[str, Any]) -> None: """Append a history entry and trim to max size.""" state = self._ensure_task(task_id) history = state.setdefault("history", []) history.append(entry) if len(history) > DEFAULT_MAX_HISTORY_ENTRIES: state["history"] = history[-DEFAULT_MAX_HISTORY_ENTRIES:] def set_history(self, task_id: str, history: list[dict[str, Any]]) -> None: """Replace the entire history list.""" self._ensure_task(task_id)["history"] = history # --- adaptive_config --- def get_adaptive_config(self, task_id: str) -> dict[str, Any] | None: """Return the adaptive config (or None).""" return self.get_task_state(task_id).get("adaptive_config") def set_adaptive_config(self, task_id: str, config: dict[str, Any]) -> None: """Set the adaptive config.""" self._ensure_task(task_id)["adaptive_config"] = config # --- trigger_runtime (per-entity) --- def get_trigger_runtime(self, task_id: str, entity_id: str | None = None) -> dict[str, Any]: """Return trigger runtime data. If *entity_id* is given, return per-entity runtime data. Otherwise, return the whole trigger_runtime dict. """ state = self.get_task_state(task_id) runtime: dict[str, Any] = state.get("trigger_runtime", {}) if entity_id is not None: result: dict[str, Any] = runtime.get(entity_id, {}) return result return runtime def set_trigger_runtime( self, task_id: str, entity_id: str, data: dict[str, Any], ) -> None: """Set trigger runtime data for a specific entity.""" state = self._ensure_task(task_id) runtime = state.setdefault("trigger_runtime", {}) # REPLACE, not merge: every trigger persists its COMPLETE per-entity # dict in one write (each type owns its slot; compound conditions get # _compound_-prefixed slots). The old .update() merge meant a key # written once could never be REMOVED — a state-change hold window's # pending_since survived its own clear on disk, and a restart during a # fresh flicker then treated the stale anchor as an elapsed window, # bypassing the #136 filter (bug audit 2026-08-22). runtime[entity_id] = dict(data) def clear_trigger_runtime(self, task_id: str) -> None: """Clear all trigger runtime data for a task. Both slots: the first-setup snapshot in ``trigger_runtime_legacy`` is reshaped into live per-entity state whenever ``trigger_runtime`` is empty (merge_task_data), so leaving it behind resurrected the adopted baseline / accumulated runtime after every clear — a baseline edited in the panel was silently replaced by the first-setup value, a switched runtime entity inherited the old hours (bug review 2026-09-04). """ state = self.get_task_state(task_id) state.pop("trigger_runtime", None) state.pop("trigger_runtime_legacy", None) # ------------------------------------------------------------------ # Merge / Split Helpers # ------------------------------------------------------------------ def merge_task_data(self, task_id: str, static_data: dict[str, Any]) -> dict[str, Any]: """Merge static ConfigEntry data with dynamic Store data. Returns a unified task dict compatible with ``MaintenanceTask.from_dict``. """ result = dict(static_data) state = self.get_task_state(task_id) # Overlay dynamic fields for field in _DYNAMIC_TASK_FIELDS: if field in state: result[field] = state[field] # Merge trigger_runtime into trigger_config._trigger_state trigger_runtime = state.get("trigger_runtime") if not trigger_runtime and state.get("trigger_runtime_legacy") and "trigger_config" in result: # No per-entity data yet: reshape the legacy flat keys so the # triggers actually restore them (see legacy_runtime_to_trigger_state). trigger_runtime = legacy_runtime_to_trigger_state(result["trigger_config"], state["trigger_runtime_legacy"]) if trigger_runtime and "trigger_config" in result: tc = dict(result["trigger_config"]) # Restructure compound flat keys (_compound_N_key) into nested # conditions list that CompoundTrigger.async_setup() expects compound_keys = [k for k in trigger_runtime if k.startswith("_compound_")] if compound_keys: non_compound = {k: v for k, v in trigger_runtime.items() if not k.startswith("_compound_")} conditions: dict[int, dict[str, Any]] = {} for k in compound_keys: # Key format: _compound_ or _compound__ # Split into: ['', 'compound', ''] or ['', 'compound', '', ''] parts = k.split("_", 3) if len(parts) < 3: continue try: idx = int(parts[2]) except ValueError: continue if len(parts) == 3: # No entity_id suffix — merge directly into condition dict conditions.setdefault(idx, {}).update(trigger_runtime[k]) else: sub_key = parts[3] conditions.setdefault(idx, {})[sub_key] = trigger_runtime[k] non_compound["conditions"] = [ conditions.get(i, {}) for i in range(max(conditions.keys()) + 1 if conditions else 0) ] tc["_trigger_state"] = non_compound else: tc["_trigger_state"] = trigger_runtime result["trigger_config"] = tc return result def merge_all_tasks(self, static_tasks: dict[str, Any]) -> dict[str, Any]: """Merge all tasks' static data with Store dynamic data.""" return {task_id: self.merge_task_data(task_id, task_data) for task_id, task_data in static_tasks.items()} # ------------------------------------------------------------------ # Migration # ------------------------------------------------------------------ def _coerce_history_timestamp(entry: dict[str, Any]) -> dict[str, Any]: """``entry`` unchanged when its timestamp is a string (or absent); a copy with an epoch number turned into a UTC ISO stamp, or without the key when the value is anything else.""" ts = entry.get("timestamp") if ts is None or isinstance(ts, str): return entry fixed = dict(entry) fixed.pop("timestamp") if isinstance(ts, (int, float)) and not isinstance(ts, bool): try: fixed["timestamp"] = datetime.fromtimestamp(ts, tz=UTC).isoformat() except (OverflowError, OSError, ValueError): pass return fixed def reanchor_from_history(store: MaintenanceStore, task_id: str, history: list[dict[str, Any]]) -> str | None: """Re-derive a task's ``last_performed`` from its history and write it. The single rule behind the history edit AND delete commands: the anchor is the day of the LATEST lifecycle entry (completed / reset / skipped / missed — ``LIFECYCLE_HISTORY_TYPES``); with none left the task reads as never performed. The cycle modifiers (a postpone's ``due_override``, ``last_planned_due``) are dropped only when the anchor actually MOVES — editing a note on the last completion keeps a postponed occurrence, deleting that completion abandons it. Returns the new anchor. """ # String stamps only: an imported epoch number made max() compare int # with str and the history edit / delete raised (bug audit 2026-09-27). latest = max( ( stamp for h in history if isinstance(h, dict) and h.get("type") in LIFECYCLE_HISTORY_TYPES and isinstance(stamp := h.get("timestamp"), str) ), default="", ) anchor = local_date_from_iso(latest) if latest else None # the stamp's local calendar day store.set_anchor(task_id, anchor, clear_modifiers=anchor != store.get_last_performed(task_id)) return anchor def extract_dynamic_from_task( task_data: dict[str, Any], ) -> tuple[dict[str, Any], dict[str, Any]]: """Split a task dict into (static, dynamic) parts. Modifies neither input; returns new dicts. """ static = dict(task_data) dynamic: dict[str, Any] = {} # Extract top-level dynamic fields (+ the split-only presentation state). for field in _DYNAMIC_TASK_FIELDS + _SPLIT_ONLY_TASK_FIELDS: if field in static: dynamic[field] = static.pop(field) # Extract trigger runtime from trigger_config if "trigger_config" in static: tc = dict(static["trigger_config"]) runtime = tc.pop("_trigger_state", None) # Also extract legacy flat runtime keys into trigger_runtime flat_runtime: dict[str, Any] = {} for flat_key in _LEGACY_TRIGGER_RUNTIME_KEYS: val = tc.pop(flat_key, None) if val is not None: flat_runtime[flat_key] = val static["trigger_config"] = tc if runtime: dynamic["trigger_runtime"] = runtime elif flat_runtime: # Legacy: flat keys exist but no _trigger_state yet. # Store them as-is; the trigger objects will migrate to # per-entity format on first setup. dynamic["trigger_runtime_legacy"] = flat_runtime return static, dynamic async def async_migrate_to_store( hass: HomeAssistant, entry_id: str, entry_data: Mapping[str, Any], store: MaintenanceStore, ) -> dict[str, Any] | Mapping[str, Any]: """Migrate dynamic state from ConfigEntry.data into a Store. Returns the cleaned static-only entry data (for updating the ConfigEntry). The Store is saved immediately. This is idempotent: if the Store already has data, nothing happens and the original *entry_data* is returned unchanged. """ existing = await store.async_load() if existing is not None: return entry_data # Already migrated tasks_data = entry_data.get(CONF_TASKS, {}) static_tasks: dict[str, Any] = {} for task_id, task_data in tasks_data.items(): static, dynamic = extract_dynamic_from_task(task_data) static_tasks[task_id] = static # Populate store store.init_task(task_id) state = store._ensure_task(task_id) state.update(dynamic) # Save store immediately (must succeed before we strip ConfigEntry) await store.async_save() # Build cleaned entry data new_data = dict(entry_data) new_data[CONF_TASKS] = static_tasks _LOGGER.info( "Migrated %d tasks from ConfigEntry to Store for entry %s", len(tasks_data), entry_id, ) return new_data