From b2ee5029468416a6be178cf94e579fab693c086a Mon Sep 17 00:00:00 2001 From: Joshua Date: Thu, 17 Sep 2026 13:26:10 -0400 Subject: [PATCH 1/4] Implement core materialization walk (Feature 07) --- custom_components/nestquest/dao_presence.py | 73 ++++- custom_components/nestquest/materialize.py | 164 +++++++++++ tests/test_materialize.py | 288 ++++++++++++++++++++ 3 files changed, 524 insertions(+), 1 deletion(-) create mode 100644 custom_components/nestquest/materialize.py create mode 100644 tests/test_materialize.py diff --git a/custom_components/nestquest/dao_presence.py b/custom_components/nestquest/dao_presence.py index 5279a25..07f3111 100644 --- a/custom_components/nestquest/dao_presence.py +++ b/custom_components/nestquest/dao_presence.py @@ -276,4 +276,75 @@ class PresenceOverridesDao: "DELETE FROM presence_overrides WHERE id = ?", (override_id,), ) - return result.rowcount > 0 \ No newline at end of file + return result.rowcount > 0 + + +async def load_snapshot( + database: NestQuestDatabase, + child_ids: list[int], + range_start: str, + range_end: str, +) -> tuple[ + dict[int, PresenceScheduleRecord], + dict[int, list[PresenceOverrideRecord]], +]: + """Read schedules and overrides for ``child_ids`` in ONE transaction. + + Returns ``(schedules, overrides)``: + + - ``schedules`` maps ``child_id`` to its :class:`PresenceScheduleRecord` + for every child that HAS a schedule row. A child absent from the + mapping has no schedule, i.e. is present every day (Feature 05). + - ``overrides`` maps ``child_id`` to that child's overrides overlapping + ``[range_start, range_end]`` (inclusive boundaries), ordered by start + date; a child with no matching override is absent from the mapping. + + Both tables are read inside one BEGIN..COMMIT span under the + connection-scoped lock, so the returned pairing is a coherent snapshot + — a concurrent schedule upsert or override create cannot interleave + between the two reads. Children are validated as existing up front so + an unknown ``child_id`` raises ValueError rather than being silently + read as "present every day", mirroring + :meth:`PresenceSchedulesDao.get_by_child`. + """ + _validate_date(range_start, "range_start") + _validate_date(range_end, "range_end") + if range_end < range_start: + raise ValueError( + "range_end must be on or after range_start, got " + f"{range_end!r} < {range_start!r}" + ) + children = sorted(set(child_ids)) + if not children: + return {}, {} + placeholders = ",".join("?" for _ in children) + async with _connection_lock(database): + async with database.transaction(): + existing_rows = await database.fetch_all( + f"SELECT id FROM children WHERE id IN ({placeholders})", + tuple(children), + ) + existing = {row[0] for row in existing_rows} + unknown = sorted(set(children) - existing) + if unknown: + raise ValueError(f"child {unknown[0]} does not exist") + schedule_rows = await database.fetch_all( + f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules " + f"WHERE child_id IN ({placeholders})", + tuple(children), + ) + override_rows = await database.fetch_all( + f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides " + f"WHERE child_id IN ({placeholders}) AND end_date >= ? " + "AND start_date <= ? ORDER BY start_date", + tuple(children) + (range_start, range_end), + ) + schedules = { + record.child_id: record + for record in map(_schedule_from_row, schedule_rows) + } + overrides: dict[int, list[PresenceOverrideRecord]] = {} + for row in override_rows: + record = _override_from_row(row) + overrides.setdefault(record.child_id, []).append(record) + return schedules, overrides \ No newline at end of file diff --git a/custom_components/nestquest/materialize.py b/custom_components/nestquest/materialize.py new file mode 100644 index 0000000..2a41a2e --- /dev/null +++ b/custom_components/nestquest/materialize.py @@ -0,0 +1,164 @@ +"""The materialization walk (Feature 07 core). + +:func:`materialize` turns the schedule rules + custody presence of the +active quest definitions into concrete ``quest_instances`` rows over a +date range. It is the pure walk over definitions x dates x assignees x +windows; HA scheduling (day-rollover triggers) and skip-completed +generation are separate later tasks and live elsewhere. + +The walk is deterministic and idempotent: + +- Every active definition is read with its decoded + :class:`~.recurrence.ScheduleRule`, assignees and windows through the + Feature 06 snapshot helpers, so each bundle is a coherent read. +- Presence is snapshotted into an immutable + :class:`~.presence.PresenceEngine` from the schedules and overrides of + every assignee child, read in one locked transaction (see + :func:`~.dao_presence.load_snapshot`). +- For each date in the range, each definition whose decoded rule + :func:`~.recurrence.occurs_on` on that date, each assignee who is + ``is_present`` on that date and each declared window, one instance is + upserted through :class:`~.dao_instances.QuestInstancesDao.upsert` with + the window's ``due_time`` snapshotted and a ``generated_at`` stamp + shared by the whole batch (one coherent time per run). + +The upsert is idempotent on (definition_id, child_id, due_date, window), +so re-running the materialization over the same range never duplicates a +row; skipped-completed handling is deliberately NOT done here. +""" +from __future__ import annotations + +import datetime + +from .dao_instances import QuestInstancesDao +from .dao_presence import load_snapshot +from .dao_rules import ( + QuestDefinitionsDao, + _validate_date, + schedule_rule_from_storage, + schedule_rule_storage_from_record, +) +from .db import NestQuestDatabase +from .presence import PresenceEngine, PresenceOverride, PresenceSchedule +from .recurrence import ScheduleRule, occurs_on + + +def _now_stamp() -> str: + """Return the batch's shared UTC second-precision generation stamp.""" + return datetime.datetime.now(datetime.timezone.utc).isoformat( + timespec="seconds" + ) + + +def _decode_rule(snapshot) -> ScheduleRule: + """Decode a snapshot's rule record back into a ScheduleRule. + + The record carried by the snapshot is the one the definition + referenced at read time, so decoding it here (rather than re-reading + through ``ScheduleRulesDao``) preserves the coherence the snapshot + guarantees. + """ + return schedule_rule_from_storage( + schedule_rule_storage_from_record(snapshot.rule) + ) + + +def _build_engine( + schedules_records, + overrides_records, +) -> PresenceEngine: + """Convert raw presence records into an immutable PresenceEngine.""" + schedules: dict[int, PresenceSchedule] = {} + for child_id, record in schedules_records.items(): + schedules[child_id] = PresenceSchedule.decode( + child_id, record.anchor_date, record.pattern + ) + overrides: dict[int, list[PresenceOverride]] = {} + for child_id, records in overrides_records.items(): + overrides[child_id] = [ + PresenceOverride( + record.child_id, + record.start_date, + record.end_date, + record.is_present, + record.note, + ) + for record in records + ] + return PresenceEngine(schedules, overrides) + + +async def materialize( + database: NestQuestDatabase, + start_date: str, + end_date: str, +) -> int: + """Generate quest instances for the closed range [start_date, end_date]. + + ``start_date`` and ``end_date`` must be strict ISO calendar dates + (YYYY-MM-DD) with ``end_date`` on or after ``start_date``; anything + else raises ValueError naming the offending field. Returns the number + of instances upserted (idempotent — a re-run for the same batch + refreshes the same rows, never duplicates them). + + The active definitions and the presence inputs are each read as a + coherent snapshot, and the presence schedule/overrides are read under + one locked transaction so the walk sees a single stable view of who + is where on each date. + """ + _validate_date(start_date, "start_date") + _validate_date(end_date, "end_date") + if end_date < start_date: + raise ValueError( + f"end_date must be on or after start_date, got " + f"{end_date!r} < {start_date!r}" + ) + + snapshots = await QuestDefinitionsDao(database).list_snapshots_active() + assignee_child_ids = sorted( + { + child.id + for snapshot in snapshots + for child in snapshot.assignees + } + ) + + schedules_records, overrides_records = await load_snapshot( + database, assignee_child_ids, start_date, end_date + ) + engine = _build_engine(schedules_records, overrides_records) + + definitions = [ + (snapshot.definition.id, _decode_rule(snapshot), + snapshot.assignees, snapshot.windows) + for snapshot in snapshots + ] + + generated_at = _now_stamp() + instances = QuestInstancesDao(database) + + start = datetime.datetime.strptime(start_date, "%Y-%m-%d").date() + end = datetime.datetime.strptime(end_date, "%Y-%m-%d").date() + + count = 0 + cursor = start + while cursor <= end: + iso = cursor.isoformat() + for definition_id, rule, assignees, windows in definitions: + if not occurs_on(rule, cursor): + continue + for child in assignees: + if not engine.is_present(child.id, cursor): + continue + for window in windows: + await instances.upsert( + definition_id, + child.id, + iso, + generated_at, + window=window.window, + due_time=window.due_time, + ) + count += 1 + cursor += datetime.timedelta(days=1) + return count \ No newline at end of file diff --git a/tests/test_materialize.py b/tests/test_materialize.py new file mode 100644 index 0000000..1e6f621 --- /dev/null +++ b/tests/test_materialize.py @@ -0,0 +1,288 @@ +"""Tests for materialize.py: the Feature 07 materialization walk.""" +from __future__ import annotations + +import asyncio +import datetime + +import pytest + +from custom_components.nestquest.dao_children import ChildrenDao +from custom_components.nestquest.dao_instances import QuestInstancesDao +from custom_components.nestquest.dao_presence import PresenceSchedulesDao +from custom_components.nestquest.db import NestQuestDatabase +from custom_components.nestquest.materialize import materialize +from custom_components.nestquest.migrations import apply_migrations +from custom_components.nestquest.quest_definitions import ( + create_quest_definition, +) +from custom_components.nestquest.recurrence import RuleType, ScheduleRule + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _make_hass_mock(): + from unittest.mock import AsyncMock, MagicMock + + hass = MagicMock() + hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a))) + return hass + + +NOW = "2026-09-14T12:00:00+00:00" + + +async def _prepare(path): + database = NestQuestDatabase(_make_hass_mock()) + await database.open(path) + await apply_migrations(database) + return database + + +def _with_db(tmp_path, name): + def _run_test(body): + async def _main(): + database = await _prepare(tmp_path / name) + try: + return await body(database) + finally: + await database.close() + + return _run(_main()) + + return _run_test + + +def _future_monday() -> datetime.date: + """The Monday of next week (strictly future, all 28 days eligible).""" + today = datetime.date.today() + this_monday = today - datetime.timedelta(days=today.weekday()) + return this_monday + datetime.timedelta(days=7) + + +def _build_expected(start, def1_id, def2_id, a_id, b_id, c_id) -> set: + """Hand-derived expected (definition_id, child_id, window, due_date, + due_time) tuples over the four-week window. + + Custody: A is present week 0 and 2 of the two-week cycle, absent + 1 and 3; B is the mirror; C has no schedule (present every day). + """ + expected: set = set() + start_dt = datetime.date.fromisoformat(start) + for offset in range(28): + day = start_dt + datetime.timedelta(days=offset) + iso = day.isoformat() + week = offset // 7 + monday = day.weekday() == 0 + + # Def1: DAILY, assignees A + C, window morning due 09:00. + if week % 2 == 0: + expected.add((def1_id, a_id, "morning", iso, "09:00")) + expected.add((def1_id, c_id, "morning", iso, "09:00")) + + # Def2: WEEKLY Monday, assignees B + C, morning 07:00 + evening 19:00. + if monday: + if week % 2 == 1: + expected.add((def2_id, b_id, "morning", iso, "07:00")) + expected.add((def2_id, b_id, "evening", iso, "19:00")) + expected.add((def2_id, c_id, "morning", iso, "07:00")) + expected.add((def2_id, c_id, "evening", iso, "19:00")) + return expected + + +async def _collect_instances(database, child_ids, start, end) -> set: + dao = QuestInstancesDao(database) + collected = set() + for child_id in child_ids: + for record in await dao.list_by_date_range(child_id, start, end): + collected.add( + ( + record.definition_id, + record.child_id, + record.window, + record.due_date, + record.due_time, + ) + ) + return collected + + +def test_materialize_two_custody_one_always_present(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + schedules = PresenceSchedulesDao(database) + + a = await children.create("Ada", NOW) # present weeks 0, 2 + b = await children.create("Bo", NOW) # present weeks 1, 3 + c = await children.create("Cleo", NOW) # no schedule: always present + + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=27)).isoformat() + + await schedules.upsert_by_child( + a.id, 2, start_iso, "0,1,2,3,4,5,6|" + ) + await schedules.upsert_by_child( + b.id, 2, start_iso, "|0,1,2,3,4,5,6" + ) + + def1 = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id, c.id], + [("morning", "09:00")], + ) + def2 = await create_quest_definition( + database, + "Weekly chore", + ScheduleRule( + rule_type=RuleType.WEEKLY, + weekday_set={0}, + start_date=start_iso, + ), + [b.id, c.id], + [("morning", "07:00"), ("evening", "19:00")], + ) + + expected = _build_expected( + start_iso, def1.definition.id, def2.definition.id, + a.id, b.id, c.id, + ) + + count = await materialize(database, start_iso, end_iso) + assert count == len(expected) + + actual = await _collect_instances( + database, (a.id, b.id, c.id), start_iso, end_iso + ) + assert actual == expected + return count + + _with_db(tmp_path, "materialize.db")(_body) + + +def test_materialize_shares_one_generated_at_per_run(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + child = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=6)).isoformat() + await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [child.id], + ["morning"], + ) + await materialize(database, start_iso, end_iso) + dao = QuestInstancesDao(database) + stamps = { + record.generated_at + for record in await dao.list_by_date_range( + child.id, start_iso, end_iso + ) + } + assert len(stamps) == 1 + return stamps + + _with_db(tmp_path, "materialize-stamp.db")(_body) + + +def test_materialize_is_idempotent(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + child = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=6)).isoformat() + await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [child.id], + ["morning"], + ) + first = await materialize(database, start_iso, end_iso) + second = await materialize(database, start_iso, end_iso) + assert first == second == 7 + return first + + _with_db(tmp_path, "materialize-idempotent.db")(_body) + + +def test_materialize_honours_overrides(tmp_path) -> None: + async def _body(database): + from custom_components.nestquest.dao_presence import ( + PresenceOverridesDao, + ) + + children = ChildrenDao(database) + overrides = PresenceOverridesDao(database) + child = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=6)).isoformat() + # No schedule means the child is present every day; a blocking + # override covering the whole week must still suppress every day. + await overrides.create( + child.id, start_iso, end_iso, False, note="away all week" + ) + await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [child.id], + ["morning"], + ) + count = await materialize(database, start_iso, end_iso) + assert count == 0 + return count + + _with_db(tmp_path, "materialize-override.db")(_body) + + +def test_materialize_rejects_malformed_or_inverted_bounds(tmp_path) -> None: + async def _body(database): + start = (_future_monday()).isoformat() + end = (datetime.date.fromisoformat(start) + + datetime.timedelta(days=1)).isoformat() + for bad in ("not-a-date", "2026-9-7", ""): + with pytest.raises(ValueError, match="start_date"): + await materialize(database, bad, end) + with pytest.raises(ValueError, match="end_date"): + await materialize(database, start, bad) + with pytest.raises(ValueError, match="end_date must be on or after"): + await materialize(database, end, start) + return None + + _with_db(tmp_path, "materialize-bounds.db")(_body) + + +def test_materialize_skips_inactive_definitions(tmp_path) -> None: + async def _body(database): + from custom_components.nestquest.quest_definitions import ( + set_quest_definition_active, + ) + + children = ChildrenDao(database) + child = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=6)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [child.id], + ["morning"], + ) + await set_quest_definition_active(database, created.definition.id, False) + count = await materialize(database, start_iso, end_iso) + assert count == 0 + return count + + _with_db(tmp_path, "materialize-inactive.db")(_body) \ No newline at end of file -- 2.55.0 From cc418a997c61e19aac1cb49497ce4fb06139ef0e Mon Sep 17 00:00:00 2001 From: Joshua Date: Thu, 17 Sep 2026 13:43:42 -0400 Subject: [PATCH 2/4] fix: single coherent input snapshot and skip-invalidated-tuple (NQ-75-001, NQ-75-002) --- custom_components/nestquest/dao_presence.py | 58 ++--- custom_components/nestquest/dao_rules.py | 91 +++++++- custom_components/nestquest/materialize.py | 89 +++++--- tests/test_materialize.py | 233 ++++++++++++++++++++ 4 files changed, 401 insertions(+), 70 deletions(-) diff --git a/custom_components/nestquest/dao_presence.py b/custom_components/nestquest/dao_presence.py index 07f3111..cf1707a 100644 --- a/custom_components/nestquest/dao_presence.py +++ b/custom_components/nestquest/dao_presence.py @@ -279,7 +279,7 @@ class PresenceOverridesDao: return result.rowcount > 0 -async def load_snapshot( +async def read_snapshot_unlocked( database: NestQuestDatabase, child_ids: list[int], range_start: str, @@ -288,7 +288,13 @@ async def load_snapshot( dict[int, PresenceScheduleRecord], dict[int, list[PresenceOverrideRecord]], ]: - """Read schedules and overrides for ``child_ids`` in ONE transaction. + """Read schedules and overrides for ``child_ids`` WITHOUT locking. + + MUST be called inside the connection lock and an open transaction, so + the schedules and overrides returned here stay coherent with every + other table the caller reads inside that same transaction (the + materialization walk's single input snapshot does exactly that). The + date range comes already validated by the caller. Returns ``(schedules, overrides)``: @@ -299,46 +305,24 @@ async def load_snapshot( ``[range_start, range_end]`` (inclusive boundaries), ordered by start date; a child with no matching override is absent from the mapping. - Both tables are read inside one BEGIN..COMMIT span under the - connection-scoped lock, so the returned pairing is a coherent snapshot - — a concurrent schedule upsert or override create cannot interleave - between the two reads. Children are validated as existing up front so - an unknown ``child_id`` raises ValueError rather than being silently - read as "present every day", mirroring - :meth:`PresenceSchedulesDao.get_by_child`. + ``child_ids`` must name existing children (the caller derives them + from a trusted snapshot, within the same transaction). """ - _validate_date(range_start, "range_start") - _validate_date(range_end, "range_end") - if range_end < range_start: - raise ValueError( - "range_end must be on or after range_start, got " - f"{range_end!r} < {range_start!r}" - ) children = sorted(set(child_ids)) if not children: return {}, {} placeholders = ",".join("?" for _ in children) - async with _connection_lock(database): - async with database.transaction(): - existing_rows = await database.fetch_all( - f"SELECT id FROM children WHERE id IN ({placeholders})", - tuple(children), - ) - existing = {row[0] for row in existing_rows} - unknown = sorted(set(children) - existing) - if unknown: - raise ValueError(f"child {unknown[0]} does not exist") - schedule_rows = await database.fetch_all( - f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules " - f"WHERE child_id IN ({placeholders})", - tuple(children), - ) - override_rows = await database.fetch_all( - f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides " - f"WHERE child_id IN ({placeholders}) AND end_date >= ? " - "AND start_date <= ? ORDER BY start_date", - tuple(children) + (range_start, range_end), - ) + schedule_rows = await database.fetch_all( + f"SELECT {_SCHEDULE_COLUMNS} FROM presence_schedules " + f"WHERE child_id IN ({placeholders})", + tuple(children), + ) + override_rows = await database.fetch_all( + f"SELECT {_OVERRIDE_COLUMNS} FROM presence_overrides " + f"WHERE child_id IN ({placeholders}) AND end_date >= ? " + "AND start_date <= ? ORDER BY start_date", + tuple(children) + (range_start, range_end), + ) schedules = { record.child_id: record for record in map(_schedule_from_row, schedule_rows) diff --git a/custom_components/nestquest/dao_rules.py b/custom_components/nestquest/dao_rules.py index 5669368..f0da787 100644 --- a/custom_components/nestquest/dao_rules.py +++ b/custom_components/nestquest/dao_rules.py @@ -951,6 +951,21 @@ class QuestDefinitionsDao: definitions = await self.list_active() return await self._snapshots_for(definitions) + async def read_snapshots_active(self) -> list[QuestDefinitionSnapshot]: + """Return every active definition's snapshot WITHOUT locking. + + Mirrors :meth:`list_snapshots_active` minus the lock and + transaction: the definitions and their rules, assignees and + windows are read on the caller's connection under whatever + transaction the caller is already holding, so they stay coherent + with every OTHER table the caller reads inside that same + transaction. MUST be called inside the connection lock and an + open transaction (the single-transaction input snapshot the + materialization walk uses is that caller). + """ + definitions = await self.list_active() + return await self._snapshots_for(definitions) + async def list_snapshots_by_child( self, child_id: int ) -> list[QuestDefinitionSnapshot]: @@ -1175,6 +1190,37 @@ class QuestDefinitionsDao: ) return result.rowcount > 0 + async def still_materializable( + self, definition_id: int, child_id: int, window: str + ) -> bool: + """Return True when the (definition, child, window) tuple is still + a valid materialization target. + + Checks, in one query under the connection lock, that the + definition exists AND is active, the child exists AND is active, + the (definition, child) assignment link is still present, and the + window is still declared on the definition. The materialization + walk re-checks each tuple against this verdict right before its + upsert so a config change landed between the input snapshot and + the upsert (an assignment removed, a definition or child + deactivated, a window removed) causes that tuple to be SKIPPED + instead of aborting the batch. This is a point read under the + lock; the walk additionally treats the upsert's own "does not + exist" / "not assigned" rejections as skippable, closing the + assignment/existence gap. + """ + async with _connection_lock(self._database): + row = await self._database.fetch_one( + "SELECT 1 FROM quest_definitions d " + "JOIN quest_definition_assignees a ON a.definition_id = d.id " + "JOIN children c ON c.id = a.child_id " + "JOIN quest_definition_windows w ON w.definition_id = d.id " + "WHERE d.id = ? AND a.child_id = ? AND w.window = ? " + "AND d.is_active = 1 AND c.is_active = 1 LIMIT 1", + (definition_id, child_id, window), + ) + return row is not None + async def _validate_assignable_child(self, child_id: int) -> None: """Raise ValueError unless the child exists and is active. @@ -1189,4 +1235,47 @@ class QuestDefinitionsDao: if not child[0]: raise ValueError( f"child {child_id} is inactive and cannot be assigned" - ) \ No newline at end of file + ) + + +async def load_materialization_input( + database: NestQuestDatabase, + range_start: str, + range_end: str, +): + """Read every materialization input in ONE locked transaction. + + Returns ``(snapshots, schedules, overrides)``: + + - ``snapshots``: every active definition with its rule, assignees and + windows (the :class:`QuestDefinitionSnapshot` shape). + - ``schedules`` / ``overrides``: the presence schedules and + range-scoped overrides of every assignee child, in + :mod:`.dao_presence`'s record shapes. + + The whole read — definitions, rules, assignees, windows, and the + assignees' presence schedules and overrides — runs inside a single + BEGIN..COMMIT span under the connection lock, so a concurrent edit + (definition, assignment, window, or presence) cannot interleave + between the parts and the walk never sees a mixed-time view. Each + table's SQL stays in its own DAO module: the presence portion is + delegated to :mod:`.dao_presence` (a lazy import here, since that + module imports this one). + """ + definitions_dao = QuestDefinitionsDao(database) + async with _connection_lock(database): + async with database.transaction(): + snapshots = await definitions_dao.read_snapshots_active() + child_ids = sorted( + { + child.id + for snapshot in snapshots + for child in snapshot.assignees + } + ) + from . import dao_presence # lazy: dao_presence imports us + + schedules, overrides = await dao_presence.read_snapshot_unlocked( + database, child_ids, range_start, range_end + ) + return snapshots, schedules, overrides \ No newline at end of file diff --git a/custom_components/nestquest/materialize.py b/custom_components/nestquest/materialize.py index 2a41a2e..131f569 100644 --- a/custom_components/nestquest/materialize.py +++ b/custom_components/nestquest/materialize.py @@ -8,13 +8,14 @@ generation are separate later tasks and live elsewhere. The walk is deterministic and idempotent: -- Every active definition is read with its decoded - :class:`~.recurrence.ScheduleRule`, assignees and windows through the - Feature 06 snapshot helpers, so each bundle is a coherent read. +- All inputs — the active definitions with their rules, assignees and + windows, AND the assignee children's presence schedules and scoped + overrides — are read inside ONE locked transaction (see + :func:`~.dao_rules.load_materialization_input`), so the walk sees a + single coherent snapshot rather than a definition state from one + instant and a presence state from another. - Presence is snapshotted into an immutable - :class:`~.presence.PresenceEngine` from the schedules and overrides of - every assignee child, read in one locked transaction (see - :func:`~.dao_presence.load_snapshot`). + :class:`~.presence.PresenceEngine` from that same coherent read. - For each date in the range, each definition whose decoded rule :func:`~.recurrence.occurs_on` on that date, each assignee who is ``is_present`` on that date and each declared window, one instance is @@ -22,6 +23,13 @@ The walk is deterministic and idempotent: the window's ``due_time`` snapshotted and a ``generated_at`` stamp shared by the whole batch (one coherent time per run). +A config change that lands AFTER the snapshot is read but BEFORE a +tuple's upsert (an assignment removed, a definition or child +deactivated, a window removed) makes that tuple SKIP rather than abort +the batch: each tuple is re-validated against the current state right +before its upsert, and only the narrow "no longer valid" conditions are +treated as skippable. + The upsert is idempotent on (definition_id, child_id, due_date, window), so re-running the materialization over the same range never duplicates a row; skipped-completed handling is deliberately NOT done here. @@ -31,10 +39,10 @@ from __future__ import annotations import datetime from .dao_instances import QuestInstancesDao -from .dao_presence import load_snapshot from .dao_rules import ( QuestDefinitionsDao, _validate_date, + load_materialization_input, schedule_rule_from_storage, schedule_rule_storage_from_record, ) @@ -88,6 +96,22 @@ def _build_engine( return PresenceEngine(schedules, overrides) +def _is_deleted_tuple(error: ValueError) -> bool: + """True for the narrow 'tuple no longer valid' upsert rejections. + + A change landing between the per-tuple validity re-check and the + upsert can still surface as the upsert's "definition does not exist" + or "not assigned to child" error; those are skippable. Every other + rejection (past due date, immutable completed instance, bad window) + is a real error and propagates. + """ + message = str(error) + return ( + "does not exist" in message + or "is not assigned to child" in message + ) + + async def materialize( database: NestQuestDatabase, start_date: str, @@ -98,13 +122,13 @@ async def materialize( ``start_date`` and ``end_date`` must be strict ISO calendar dates (YYYY-MM-DD) with ``end_date`` on or after ``start_date``; anything else raises ValueError naming the offending field. Returns the number - of instances upserted (idempotent — a re-run for the same batch - refreshes the same rows, never duplicates them). + of instances upserted for the range (idempotent — a re-run refreshes + the same rows, never duplicates them). - The active definitions and the presence inputs are each read as a - coherent snapshot, and the presence schedule/overrides are read under - one locked transaction so the walk sees a single stable view of who - is where on each date. + All inputs are read in ONE locked transaction, and each tuple is + re-validated against the current state just before its upsert so a + config change that arrives mid-batch skips the affected tuples rather + than aborting the batch or leaving a partial result. """ _validate_date(start_date, "start_date") _validate_date(end_date, "end_date") @@ -114,17 +138,8 @@ async def materialize( f"{end_date!r} < {start_date!r}" ) - snapshots = await QuestDefinitionsDao(database).list_snapshots_active() - assignee_child_ids = sorted( - { - child.id - for snapshot in snapshots - for child in snapshot.assignees - } - ) - - schedules_records, overrides_records = await load_snapshot( - database, assignee_child_ids, start_date, end_date + snapshots, schedules_records, overrides_records = await ( + load_materialization_input(database, start_date, end_date) ) engine = _build_engine(schedules_records, overrides_records) @@ -136,6 +151,7 @@ async def materialize( generated_at = _now_stamp() instances = QuestInstancesDao(database) + definitions_dao = QuestDefinitionsDao(database) start = datetime.datetime.strptime(start_date, "%Y-%m-%d").date() end = datetime.datetime.strptime(end_date, "%Y-%m-%d").date() @@ -151,14 +167,23 @@ async def materialize( if not engine.is_present(child.id, cursor): continue for window in windows: - await instances.upsert( - definition_id, - child.id, - iso, - generated_at, - window=window.window, - due_time=window.due_time, - ) + if not await definitions_dao.still_materializable( + definition_id, child.id, window.window + ): + continue + try: + await instances.upsert( + definition_id, + child.id, + iso, + generated_at, + window=window.window, + due_time=window.due_time, + ) + except ValueError as error: + if _is_deleted_tuple(error): + continue + raise count += 1 cursor += datetime.timedelta(days=1) return count \ No newline at end of file diff --git a/tests/test_materialize.py b/tests/test_materialize.py index 1e6f621..b6c2736 100644 --- a/tests/test_materialize.py +++ b/tests/test_materialize.py @@ -9,6 +9,7 @@ import pytest from custom_components.nestquest.dao_children import ChildrenDao from custom_components.nestquest.dao_instances import QuestInstancesDao from custom_components.nestquest.dao_presence import PresenceSchedulesDao +from custom_components.nestquest.dao_rules import QuestDefinitionsDao from custom_components.nestquest.db import NestQuestDatabase from custom_components.nestquest.materialize import materialize from custom_components.nestquest.migrations import apply_migrations @@ -209,11 +210,243 @@ def test_materialize_is_idempotent(tmp_path) -> None: first = await materialize(database, start_iso, end_iso) second = await materialize(database, start_iso, end_iso) assert first == second == 7 + # Running twice refreshes the SAME seven rows: exactly one + # physical row per (definition_id, child_id, due_date, window) + # key, never a duplicate. + records = await QuestInstancesDao(database).list_by_date_range( + child.id, start_iso, end_iso + ) + keys = [ + (r.definition_id, r.child_id, r.due_date, r.window) + for r in records + ] + assert len(keys) == 7 + assert len(set(keys)) == 7 return first _with_db(tmp_path, "materialize-idempotent.db")(_body) +async def _materialize_with_change( + database, start_iso, end_iso, change +) -> int: + """Run materialize, pause before its first per-tuple re-check, apply + ``change``, then release. Returns the materialize count. + + The gate parks the walk AFTER the single input snapshot is read but + BEFORE the first :meth:`QuestDefinitionsDao.still_materializable` + re-check acquires the connection lock, so ``change`` commits against + the live state and the walk's per-tuple re-check then skips the + tuples it invalidated (or, for a presence-only change, continues with + the snapped presence). This deterministically exercises the + snapshot-to-upsert gap that a config edit can interleave into. + """ + original_check = QuestDefinitionsDao.still_materializable + entered = asyncio.Event() + release = asyncio.Event() + seen = {"n": 0} + + async def _gated_check(self, definition_id, child_id, window): + seen["n"] += 1 + if seen["n"] == 1: + seen["first"] = (definition_id, child_id, window) + entered.set() + await release.wait() + return await original_check(self, definition_id, child_id, window) + + QuestDefinitionsDao.still_materializable = _gated_check + try: + task = asyncio.ensure_future( + materialize(database, start_iso, end_iso) + ) + await entered.wait() + await change() + release.set() + return await task + finally: + QuestDefinitionsDao.still_materializable = original_check + + +def test_materialize_skips_removed_assignee_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + c = await children.create("Cleo", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id, c.id], + ["morning"], + ) + + async def _remove_assignee(): + await QuestDefinitionsDao(database).remove_assignee( + created.definition.id, a.id + ) + + count = await _materialize_with_change( + database, start_iso, end_iso, _remove_assignee + ) + assert count == 3 # Cleo's three days survive + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + assert len(await QuestInstancesDao(database).list_by_date_range( + c.id, start_iso, end_iso + )) == 3 + return count + + _with_db(tmp_path, "materialize-removed-assignee.db")(_body) + + +def test_materialize_skips_deactivated_definition_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + c = await children.create("Cleo", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id, c.id], + ["morning"], + ) + + async def _deactivate(): + await QuestDefinitionsDao(database).set_active( + created.definition.id, False + ) + + count = await _materialize_with_change( + database, start_iso, end_iso, _deactivate + ) + assert count == 0 + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + assert await QuestInstancesDao(database).list_by_date_range( + c.id, start_iso, end_iso + ) == [] + return count + + _with_db(tmp_path, "materialize-deactivated-definition.db")(_body) + + +def test_materialize_skips_removed_window_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id], + [("morning", "09:00")], + ) + + async def _remove_window(): + await QuestDefinitionsDao(database).remove_window( + created.definition.id, "morning" + ) + + count = await _materialize_with_change( + database, start_iso, end_iso, _remove_window + ) + assert count == 0 + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + return count + + _with_db(tmp_path, "materialize-removed-window.db")(_body) + + +def test_materialize_skips_deactivated_child_mid_walk(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + a = await children.create("Ada", NOW) + c = await children.create("Cleo", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [a.id, c.id], + ["morning"], + ) + + async def _deactivate_child(): + await ChildrenDao(database).set_active(a.id, False) + + count = await _materialize_with_change( + database, start_iso, end_iso, _deactivate_child + ) + assert count == 3 # Cleo's three days survive + assert await QuestInstancesDao(database).list_by_date_range( + a.id, start_iso, end_iso + ) == [] + assert len(await QuestInstancesDao(database).list_by_date_range( + c.id, start_iso, end_iso + )) == 3 + return count + + _with_db(tmp_path, "materialize-deactivated-child.db")(_body) + + +def test_materialize_uses_single_presence_snapshot_mid_walk(tmp_path) -> None: + async def _body(database): + from custom_components.nestquest.dao_presence import ( + PresenceOverridesDao, + ) + + children = ChildrenDao(database) + overrides = PresenceOverridesDao(database) + child = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [child.id], + ["morning"], + ) + + async def _add_blocking_override(): + await overrides.create( + child.id, start_iso, end_iso, False, note="away" + ) + + # Presence is read once, inside the input snapshot; a blocking + # override landing AFTER that read must not make the walk panic + # or half-apply it. The walk stays fully governed by the + # snapshot it read: Ada was present every day there, so all + # three days still materialize, deterministically. + count = await _materialize_with_change( + database, start_iso, end_iso, _add_blocking_override + ) + assert count == 3 + assert len(await QuestInstancesDao(database).list_by_date_range( + child.id, start_iso, end_iso + )) == 3 + return count + + _with_db(tmp_path, "materialize-presence-snapshot.db")(_body) + + def test_materialize_honours_overrides(tmp_path) -> None: async def _body(database): from custom_components.nestquest.dao_presence import ( -- 2.55.0 From 6df3c720859e91f7df260a558470a7956b926178 Mon Sep 17 00:00:00 2001 From: Joshua Date: Thu, 17 Sep 2026 13:50:38 -0400 Subject: [PATCH 3/4] fix: atomic materialization upsert validates active/assignee/window (NQ-75-CR-001) --- custom_components/nestquest/dao_instances.py | 106 +++++++++++++++++++ custom_components/nestquest/dao_rules.py | 31 ------ custom_components/nestquest/materialize.py | 73 +++++-------- tests/test_materialize.py | 29 ++--- 4 files changed, 148 insertions(+), 91 deletions(-) diff --git a/custom_components/nestquest/dao_instances.py b/custom_components/nestquest/dao_instances.py index 93437f1..36d7000 100644 --- a/custom_components/nestquest/dao_instances.py +++ b/custom_components/nestquest/dao_instances.py @@ -277,6 +277,112 @@ class QuestInstancesDao: assert instance is not None return instance + async def upsert_if_valid( + self, + definition_id: int, + child_id: int, + due_date: str, + generated_at: str, + *, + window: str, + ) -> QuestInstanceRecord | None: + """Insert one dated instance iff its tuple is still materializable. + + This is the materialization walk's single write path. Unlike the + generic :meth:`upsert`, which validates only existence and + assignment, this method re-validates EVERY materialization + precondition in the SAME connection-lock-held transaction as the + INSERT — closing the check-then-insert gap a concurrent config + write could otherwise slip through: + + - the definition exists AND is active (a definition deactivated + after the input snapshot is skipped, never written); + - the child exists AND is active (a deactivated child is skipped); + - the (definition, child) assignment link still exists (a removed + assignment is skipped); + - the window is still declared on the definition (a removed window + is skipped), and its CURRENT ``due_time`` is read from the live + row so a due-time edit landing after the snapshot is honored, + never the stale snapshot value. + + Returns the stored instance, or ``None`` when a precondition + failed — the tuple no longer materializable, which the walk SKIPS + rather than raising, so the rest of the batch is unaffected. The + no-past rule and the immutable-completed-instance refusal are + preserved from :meth:`upsert` (those are not configuration + invalidations and still raise ValueError). + """ + _validate_date(due_date, "due_date") + _validate_window(window) + # Fail fast on obviously-past dates before queuing on the lock; + # the authoritative check re-runs under the lock (see upsert). + today = datetime.date.today().isoformat() + if due_date < today: + raise ValueError( + f"instances are never generated in the past: due_date " + f"{due_date!r} is before today {today!r}" + ) + async with _connection_lock(self._database): + async with self._database.transaction(): + today = datetime.date.today().isoformat() + if due_date < today: + raise ValueError( + f"instances are never generated in the past: " + f"due_date {due_date!r} is before today {today!r}" + ) + valid = await self._database.fetch_one( + "SELECT w.due_time FROM quest_definitions d " + "JOIN quest_definition_assignees a " + "ON a.definition_id = d.id " + "JOIN children c ON c.id = a.child_id " + "JOIN quest_definition_windows w " + "ON w.definition_id = d.id " + "WHERE d.id = ? AND a.child_id = ? AND w.window = ? " + "AND d.is_active = 1 AND c.is_active = 1 LIMIT 1", + (definition_id, child_id, window), + ) + if valid is None: + # A config change invalidated the tuple since the + # snapshot: skip it rather than abort the batch. + return None + due_time = valid[0] + existing = await self._database.fetch_one( + "SELECT 1 FROM completion_events " + "WHERE instance_id = (SELECT id FROM quest_instances " + "WHERE definition_id = ? AND child_id = ? " + "AND due_date = ? AND window = ?)", + (definition_id, child_id, due_date, window), + ) + if existing is not None: + raise ValueError( + f"instance for (definition {definition_id}, " + f"child {child_id}, due_date {due_date!r}, " + f"window {window!r}) already has a completion " + "event and is immutable; it cannot be regenerated" + ) + await self._database.execute( + "INSERT INTO quest_instances (definition_id, child_id, " + "window, due_date, due_time, generated_at) " + "VALUES (?, ?, ?, ?, ?, ?) " + "ON CONFLICT (definition_id, child_id, due_date, " + "window) DO UPDATE SET " + "due_time = excluded.due_time, " + "generated_at = excluded.generated_at", + ( + definition_id, + child_id, + window, + due_date, + due_time, + generated_at, + ), + ) + instance = await self.get( + definition_id, child_id, due_date, window + ) + assert instance is not None + return instance + async def get( self, definition_id: int, diff --git a/custom_components/nestquest/dao_rules.py b/custom_components/nestquest/dao_rules.py index f0da787..ec3f6d9 100644 --- a/custom_components/nestquest/dao_rules.py +++ b/custom_components/nestquest/dao_rules.py @@ -1190,37 +1190,6 @@ class QuestDefinitionsDao: ) return result.rowcount > 0 - async def still_materializable( - self, definition_id: int, child_id: int, window: str - ) -> bool: - """Return True when the (definition, child, window) tuple is still - a valid materialization target. - - Checks, in one query under the connection lock, that the - definition exists AND is active, the child exists AND is active, - the (definition, child) assignment link is still present, and the - window is still declared on the definition. The materialization - walk re-checks each tuple against this verdict right before its - upsert so a config change landed between the input snapshot and - the upsert (an assignment removed, a definition or child - deactivated, a window removed) causes that tuple to be SKIPPED - instead of aborting the batch. This is a point read under the - lock; the walk additionally treats the upsert's own "does not - exist" / "not assigned" rejections as skippable, closing the - assignment/existence gap. - """ - async with _connection_lock(self._database): - row = await self._database.fetch_one( - "SELECT 1 FROM quest_definitions d " - "JOIN quest_definition_assignees a ON a.definition_id = d.id " - "JOIN children c ON c.id = a.child_id " - "JOIN quest_definition_windows w ON w.definition_id = d.id " - "WHERE d.id = ? AND a.child_id = ? AND w.window = ? " - "AND d.is_active = 1 AND c.is_active = 1 LIMIT 1", - (definition_id, child_id, window), - ) - return row is not None - async def _validate_assignable_child(self, child_id: int) -> None: """Raise ValueError unless the child exists and is active. diff --git a/custom_components/nestquest/materialize.py b/custom_components/nestquest/materialize.py index 131f569..387380a 100644 --- a/custom_components/nestquest/materialize.py +++ b/custom_components/nestquest/materialize.py @@ -19,16 +19,18 @@ The walk is deterministic and idempotent: - For each date in the range, each definition whose decoded rule :func:`~.recurrence.occurs_on` on that date, each assignee who is ``is_present`` on that date and each declared window, one instance is - upserted through :class:`~.dao_instances.QuestInstancesDao.upsert` with - the window's ``due_time`` snapshotted and a ``generated_at`` stamp - shared by the whole batch (one coherent time per run). + written through :class:`~.dao_instances.QuestInstancesDao.upsert_if_valid` + with a ``generated_at`` stamp shared by the whole batch (one coherent + time per run). A config change that lands AFTER the snapshot is read but BEFORE a -tuple's upsert (an assignment removed, a definition or child -deactivated, a window removed) makes that tuple SKIP rather than abort -the batch: each tuple is re-validated against the current state right -before its upsert, and only the narrow "no longer valid" conditions are -treated as skippable. +tuple's write (an assignment removed, a definition or child deactivated, +a window removed) makes that tuple SKIP rather than abort the batch: +:meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid` re-validates +every materialization precondition (active definition, active child, +live assignment, declared window, and its CURRENT ``due_time``) inside +the SAME transaction as the INSERT, so the check and the write cannot be +split by a racing config edit. The upsert is idempotent on (definition_id, child_id, due_date, window), so re-running the materialization over the same range never duplicates a @@ -40,7 +42,6 @@ import datetime from .dao_instances import QuestInstancesDao from .dao_rules import ( - QuestDefinitionsDao, _validate_date, load_materialization_input, schedule_rule_from_storage, @@ -96,22 +97,6 @@ def _build_engine( return PresenceEngine(schedules, overrides) -def _is_deleted_tuple(error: ValueError) -> bool: - """True for the narrow 'tuple no longer valid' upsert rejections. - - A change landing between the per-tuple validity re-check and the - upsert can still surface as the upsert's "definition does not exist" - or "not assigned to child" error; those are skippable. Every other - rejection (past due date, immutable completed instance, bad window) - is a real error and propagates. - """ - message = str(error) - return ( - "does not exist" in message - or "is not assigned to child" in message - ) - - async def materialize( database: NestQuestDatabase, start_date: str, @@ -125,10 +110,12 @@ async def materialize( of instances upserted for the range (idempotent — a re-run refreshes the same rows, never duplicates them). - All inputs are read in ONE locked transaction, and each tuple is - re-validated against the current state just before its upsert so a - config change that arrives mid-batch skips the affected tuples rather - than aborting the batch or leaving a partial result. + All inputs are read in ONE locked transaction; each tuple's write then + goes through the atomic :meth:`~.dao_instances.QuestInstancesDao.upsert_if_valid`, + which re-validates the tuple and its current ``due_time`` in the same + transaction as the INSERT, so a config change that arrives mid-batch + skips the affected tuples rather than aborting the batch or leaving a + partial result. """ _validate_date(start_date, "start_date") _validate_date(end_date, "end_date") @@ -151,7 +138,6 @@ async def materialize( generated_at = _now_stamp() instances = QuestInstancesDao(database) - definitions_dao = QuestDefinitionsDao(database) start = datetime.datetime.strptime(start_date, "%Y-%m-%d").date() end = datetime.datetime.strptime(end_date, "%Y-%m-%d").date() @@ -167,23 +153,14 @@ async def materialize( if not engine.is_present(child.id, cursor): continue for window in windows: - if not await definitions_dao.still_materializable( - definition_id, child.id, window.window - ): - continue - try: - await instances.upsert( - definition_id, - child.id, - iso, - generated_at, - window=window.window, - due_time=window.due_time, - ) - except ValueError as error: - if _is_deleted_tuple(error): - continue - raise - count += 1 + written = await instances.upsert_if_valid( + definition_id, + child.id, + iso, + generated_at, + window=window.window, + ) + if written is not None: + count += 1 cursor += datetime.timedelta(days=1) return count \ No newline at end of file diff --git a/tests/test_materialize.py b/tests/test_materialize.py index b6c2736..ff9dace 100644 --- a/tests/test_materialize.py +++ b/tests/test_materialize.py @@ -230,31 +230,36 @@ def test_materialize_is_idempotent(tmp_path) -> None: async def _materialize_with_change( database, start_iso, end_iso, change ) -> int: - """Run materialize, pause before its first per-tuple re-check, apply + """Run materialize, pause before its first atomic write, apply ``change``, then release. Returns the materialize count. The gate parks the walk AFTER the single input snapshot is read but - BEFORE the first :meth:`QuestDefinitionsDao.still_materializable` - re-check acquires the connection lock, so ``change`` commits against - the live state and the walk's per-tuple re-check then skips the - tuples it invalidated (or, for a presence-only change, continues with - the snapped presence). This deterministically exercises the - snapshot-to-upsert gap that a config edit can interleave into. + BEFORE the first :meth:`QuestInstancesDao.upsert_if_valid` call + acquires the connection lock, so ``change`` commits against the live + state and the write's own in-transaction precondition check then + skips the tuples it invalidated (or, for a presence-only change, + proceeds with the snapped presence). This deterministically + exercises the snapshot-to-insert gap that a config edit can + interleave into. """ - original_check = QuestDefinitionsDao.still_materializable + original_upsert = QuestInstancesDao.upsert_if_valid entered = asyncio.Event() release = asyncio.Event() seen = {"n": 0} - async def _gated_check(self, definition_id, child_id, window): + async def _gated(self, definition_id, child_id, due_date, generated_at, + *, window): seen["n"] += 1 if seen["n"] == 1: seen["first"] = (definition_id, child_id, window) entered.set() await release.wait() - return await original_check(self, definition_id, child_id, window) + return await original_upsert( + self, definition_id, child_id, due_date, generated_at, + window=window, + ) - QuestDefinitionsDao.still_materializable = _gated_check + QuestInstancesDao.upsert_if_valid = _gated try: task = asyncio.ensure_future( materialize(database, start_iso, end_iso) @@ -264,7 +269,7 @@ async def _materialize_with_change( release.set() return await task finally: - QuestDefinitionsDao.still_materializable = original_check + QuestInstancesDao.upsert_if_valid = original_upsert def test_materialize_skips_removed_assignee_mid_walk(tmp_path) -> None: -- 2.55.0 From fac9f483ef39fababcfc3a02505b512600ba44e6 Mon Sep 17 00:00:00 2001 From: Joshua Date: Thu, 17 Sep 2026 13:59:59 -0400 Subject: [PATCH 4/4] fix: store the snapshot due_time, not the live value (NQ-75-CR-002) --- custom_components/nestquest/dao_instances.py | 19 +++++---- custom_components/nestquest/materialize.py | 1 + tests/test_materialize.py | 42 +++++++++++++++++++- 3 files changed, 53 insertions(+), 9 deletions(-) diff --git a/custom_components/nestquest/dao_instances.py b/custom_components/nestquest/dao_instances.py index 36d7000..841c1a7 100644 --- a/custom_components/nestquest/dao_instances.py +++ b/custom_components/nestquest/dao_instances.py @@ -285,6 +285,7 @@ class QuestInstancesDao: generated_at: str, *, window: str, + due_time: str | None = None, ) -> QuestInstanceRecord | None: """Insert one dated instance iff its tuple is still materializable. @@ -301,9 +302,14 @@ class QuestInstancesDao: - the (definition, child) assignment link still exists (a removed assignment is skipped); - the window is still declared on the definition (a removed window - is skipped), and its CURRENT ``due_time`` is read from the live - row so a due-time edit landing after the snapshot is honored, - never the stale snapshot value. + is skipped). + + The live checks above ONLY decide whether the tuple is still + valid; they never read the window's ``due_time``. The instance is + written with the caller-supplied ``due_time`` — the value snapshotted + from the coherent input read — so a window-time edit landing + mid-walk cannot leak a newer time into a batch that was already + snapshotted against older rule/presence/assignment decisions. Returns the stored instance, or ``None`` when a precondition failed — the tuple no longer materializable, which the walk SKIPS @@ -330,8 +336,8 @@ class QuestInstancesDao: f"instances are never generated in the past: " f"due_date {due_date!r} is before today {today!r}" ) - valid = await self._database.fetch_one( - "SELECT w.due_time FROM quest_definitions d " + still_valid = await self._database.fetch_one( + "SELECT 1 FROM quest_definitions d " "JOIN quest_definition_assignees a " "ON a.definition_id = d.id " "JOIN children c ON c.id = a.child_id " @@ -341,11 +347,10 @@ class QuestInstancesDao: "AND d.is_active = 1 AND c.is_active = 1 LIMIT 1", (definition_id, child_id, window), ) - if valid is None: + if still_valid is None: # A config change invalidated the tuple since the # snapshot: skip it rather than abort the batch. return None - due_time = valid[0] existing = await self._database.fetch_one( "SELECT 1 FROM completion_events " "WHERE instance_id = (SELECT id FROM quest_instances " diff --git a/custom_components/nestquest/materialize.py b/custom_components/nestquest/materialize.py index 387380a..5c4de5b 100644 --- a/custom_components/nestquest/materialize.py +++ b/custom_components/nestquest/materialize.py @@ -159,6 +159,7 @@ async def materialize( iso, generated_at, window=window.window, + due_time=window.due_time, ) if written is not None: count += 1 diff --git a/tests/test_materialize.py b/tests/test_materialize.py index ff9dace..86dfe31 100644 --- a/tests/test_materialize.py +++ b/tests/test_materialize.py @@ -248,7 +248,7 @@ async def _materialize_with_change( seen = {"n": 0} async def _gated(self, definition_id, child_id, due_date, generated_at, - *, window): + *, window, due_time=None): seen["n"] += 1 if seen["n"] == 1: seen["first"] = (definition_id, child_id, window) @@ -256,7 +256,7 @@ async def _materialize_with_change( await release.wait() return await original_upsert( self, definition_id, child_id, due_date, generated_at, - window=window, + window=window, due_time=due_time, ) QuestInstancesDao.upsert_if_valid = _gated @@ -452,6 +452,44 @@ def test_materialize_uses_single_presence_snapshot_mid_walk(tmp_path) -> None: _with_db(tmp_path, "materialize-presence-snapshot.db")(_body) +def test_materialize_stores_snapshot_due_time_not_live(tmp_path) -> None: + async def _body(database): + children = ChildrenDao(database) + child = await children.create("Ada", NOW) + start = _future_monday() + start_iso = start.isoformat() + end_iso = (start + datetime.timedelta(days=2)).isoformat() + created = await create_quest_definition( + database, + "Daily chore", + ScheduleRule(rule_type=RuleType.DAILY, start_date=start_iso), + [child.id], + [("morning", "09:00")], + ) + + async def _edit_due_time(): + await QuestDefinitionsDao(database).upsert_window( + created.definition.id, "morning", due_time="10:15" + ) + + # The window's due_time is edited to 10:15 between the input + # snapshot and the insert. The walk must store the SNAPSHOT's + # 09:00, not the live 10:15 — the batch is governed entirely by + # the coherent snapshot it read. + count = await _materialize_with_change( + database, start_iso, end_iso, _edit_due_time + ) + assert count == 3 + records = await QuestInstancesDao(database).list_by_date_range( + child.id, start_iso, end_iso + ) + assert len(records) == 3 + assert {r.due_time for r in records} == {"09:00"} + return records + + _with_db(tmp_path, "materialize-snapshot-due-time.db")(_body) + + def test_materialize_honours_overrides(tmp_path) -> None: async def _body(database): from custom_components.nestquest.dao_presence import ( -- 2.55.0