fix: enforce integer storage and weekday_set validation in schema

custom_components/nestquest/schema.py

@@ -12,11 +12,58 @@ caller/DAO rather than a DB default, per the feature guardrails.  No
 produced by callers is not fixed yet (``datetime.isoformat()`` may
 include microseconds, changing the length), so enforcing a fixed length
 here would reject valid values.  Only non-null is enforced for now.
+
+Date policy: ``start_date``/``end_date`` are ISO date strings (YYYY-MM-DD,
+no time component) compared lexically by the ``end_date >= start_date``
+CHECK; due_date/due_time string shapes are caller-validated by the rule
+engine and DAO layers, not by DDL.
+
+``weekday_set`` format: comma-separated weekday integers (CSV), 0 =
+Monday .. 6 = Sunday, e.g. ``'0,2,4'`` for Mon/Wed/Fri.  Chosen over JSON
+because it sorts, compares and parses trivially in SQL.  The format is
+enforced at the DB level: a CHECK matches ``weekday_set`` against GLOB
+patterns for element counts 1..7 of single digits 0-6 (GLOB has no
+repetition quantifier, so the element count is enumerated; more than 7
+elements is impossible for a weekday set).  ``''``, ``'x'``, ``'7'`` and
+``'0,x'`` are all rejected; NULL stays allowed for non-weekly rules.
+This keeps the schema layer storage-only — the DAO layer (tasks
+8c72300c+) does not need its own validator, it can rely on the CHECK.
+
+Type affinity: SQLite applies column affinity on insert, but CHECKs
+still observe text and fractional values ('abc' stays text, '1.5'
+becomes REAL), so every INTEGER column carrying rule semantics carries
+a ``typeof(...) = 'integer'`` guard in its CHECK.
+
+Rule-type coherence: two CHECKs enforce that ``weekly`` carries a
+non-null ``weekday_set``, ``monthly`` carries a ``day_of_month`` or an
+``nth_weekday``, ``yearly`` carries a ``month``, and ``daily``/``custom``
+carry none of the month-specific fields.  Relaxed deliberately: the DDL
+does NOT forbid ``month`` on weekly/monthly rules (an "every 2 weeks"
+rule or a monthly rule could later be pinned to a month) nor extra
+fields on weekly/monthly/yearly rules generally, so the CHECKs only
+reject what the rule engine definitively cannot express now.
 """
 from __future__ import annotations
 
 from .db import NestQuestDatabase
 
+#: GLOB patterns matching a valid ``weekday_set`` CSV: one GLOB per
+#: possible element count 1..7 (a weekday set cannot exceed 7 elements).
+#: Each element is a single digit 0-6, separated by a literal comma.
+#: GLOB has no repetition quantifier, so the element count is enumerated.
+#: ``''`` and NULL are deliberately NOT matched: NULL is allowed for
+#: non-weekly rules via a separate ``IS NULL`` arm, and '' is invalid
+#: because weekly rules need at least one weekday.
+WEEKDAY_SET_GLOB_PATTERNS: list[str] = [
+    "[0-6]",
+    "[0-6],[0-6]",
+    "[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6],[0-6],[0-6]",
+    "[0-6],[0-6],[0-6],[0-6],[0-6],[0-6],[0-6]",
+]
+
 SCHEMA_V1_CHILDREN_DDL: list[str] = [
     """
     CREATE TABLE IF NOT EXISTS children (
@@ -24,8 +71,8 @@ SCHEMA_V1_CHILDREN_DDL: list[str] = [
         display_name TEXT NOT NULL,
         colour TEXT,
         avatar_ref TEXT,
-        sort_order INTEGER NOT NULL DEFAULT 0,
-        is_active INTEGER NOT NULL DEFAULT 1 CHECK (is_active IN (0, 1)),
+        sort_order INTEGER NOT NULL DEFAULT 0 CHECK (typeof(sort_order) = 'integer'),
+        is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
         created_at TEXT NOT NULL
     )
     """,
@@ -40,9 +87,54 @@ SCHEMA_V1_ADMIN_USERS_DDL: list[str] = [
     """,
 ]
 
+_WEEKDAY_SET_CHECK = " OR ".join(
+    f"weekday_set GLOB '{pattern}'" for pattern in WEEKDAY_SET_GLOB_PATTERNS
+)
+
+SCHEMA_V1_SCHEDULE_RULES_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS schedule_rules (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        rule_type TEXT NOT NULL CHECK (rule_type IN ('daily', 'weekly', 'monthly', 'yearly', 'custom')),
+        interval INTEGER NOT NULL DEFAULT 1 CHECK (typeof(interval) = 'integer' AND interval >= 1),
+        weekday_set TEXT CHECK (weekday_set IS NULL OR ({_WEEKDAY_SET_CHECK})),
+        day_of_month INTEGER CHECK (day_of_month IS NULL OR (typeof(day_of_month) = 'integer' AND day_of_month >= 1 AND day_of_month <= 31)),
+        nth_weekday INTEGER CHECK (nth_weekday IS NULL OR (typeof(nth_weekday) = 'integer' AND nth_weekday IN (-1, 1, 2, 3, 4, 5))),
+        month INTEGER CHECK (month IS NULL OR (typeof(month) = 'integer' AND month >= 1 AND month <= 12)),
+        start_date TEXT NOT NULL,
+        end_date TEXT,
+        CHECK (end_date IS NULL OR end_date >= start_date),
+        CHECK (
+            (rule_type = 'weekly' AND weekday_set IS NOT NULL)
+            OR (rule_type = 'monthly' AND (day_of_month IS NOT NULL OR nth_weekday IS NOT NULL))
+            OR (rule_type = 'yearly' AND month IS NOT NULL)
+            OR (rule_type IN ('daily', 'custom') AND day_of_month IS NULL AND nth_weekday IS NULL AND month IS NULL)
+        )
+    )
+    """,
+]
+
+SCHEMA_V1_TASK_DEFINITIONS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS task_definitions (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        title TEXT NOT NULL,
+        description TEXT,
+        icon TEXT,
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
+        due_time TEXT,
+        is_active INTEGER NOT NULL DEFAULT 1 CHECK (typeof(is_active) = 'integer' AND is_active IN (0, 1)),
+        created_at TEXT NOT NULL
+    )
+    """,
+]
+
 SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_CHILDREN_DDL,
     *SCHEMA_V1_ADMIN_USERS_DDL,
+    *SCHEMA_V1_SCHEDULE_RULES_DDL,
+    *SCHEMA_V1_TASK_DEFINITIONS_DDL,
 ]
 
 

tests/test_lifecycle.py

@@ -236,4 +236,23 @@ async def test_unload_closes_database_even_when_listener_removal_raises() -> Non
         await async_unload_entry(hass, entry)
     assert database.connected is False, "database leaked despite remover error"
     assert entry.runtime_data is None
-    assert DOMAIN not in hass.data
+    assert DOMAIN not in hass.data
+
+
+async def test_setup_applies_all_v1_tables(hass, make_entry) -> None:
+    """Setup applies the full v1 DDL: all four v1 tables exist afterwards."""
+    entry = _wire(make_entry(), hass.registry)
+    assert await async_setup_entry(hass, entry) is True
+    database = entry.runtime_data.database
+    rows = await database.fetch_all(
+        "SELECT name FROM sqlite_master WHERE type = 'table' "
+        "AND name IN ('children', 'admin_users', 'schedule_rules', "
+        "'task_definitions') ORDER BY name"
+    )
+    assert [row[0] for row in rows] == [
+        "admin_users",
+        "children",
+        "schedule_rules",
+        "task_definitions",
+    ]
+    await async_unload_entry(hass, entry)

tests/test_schema.py

@@ -10,7 +10,9 @@ from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.schema import (
     SCHEMA_V1_ADMIN_USERS_DDL,
     SCHEMA_V1_CHILDREN_DDL,
+    SCHEMA_V1_SCHEDULE_RULES_DDL,
     SCHEMA_V1_STATEMENTS,
+    SCHEMA_V1_TASK_DEFINITIONS_DDL,
     async_apply_ddl,
 )
 
@@ -37,15 +39,73 @@ def _apply(database) -> None:
     _run(async_apply_ddl(database, SCHEMA_V1_STATEMENTS))
 
 
+# Sentinel used by the insert helpers: omit the column from the INSERT so
+# its DDL DEFAULT applies (explicit NULL would not trigger a DEFAULT).
+_OMIT = object()
+
+
+def _insert_rule(database, **overrides):
+    """Insert one valid schedule_rule, returning the rule id.
+
+    Pass ``_OMIT`` as a value to leave that column out of the INSERT so
+    its DDL DEFAULT applies.
+    """
+    values: dict[str, object] = {
+        "rule_type": "weekly",
+        "weekday_set": "0,2,4",
+        "start_date": "2026-09-13",
+    }
+    values.update(overrides)
+    values = {k: v for k, v in values.items() if v is not _OMIT}
+    columns = ", ".join(values)
+    placeholders = ", ".join("?" for _ in values)
+    result = _run(
+        database.execute(
+            f"INSERT INTO schedule_rules ({columns}) VALUES ({placeholders})",
+            tuple(values.values()),
+        )
+    )
+    assert result.lastrowid is not None
+    return result.lastrowid
+
+
+def _insert_definition(database, child_id=1, rule_id=1, **overrides):
+    """Insert one valid task_definition, returning the definition id.
+
+    Pass ``_OMIT`` as a value to leave that column out of the INSERT so
+    its DDL DEFAULT applies.
+    """
+    values: dict[str, object] = {
+        "title": "Brush teeth",
+        "child_id": child_id,
+        "schedule_rule_id": rule_id,
+        "created_at": "2026-09-13T00:00:00+00:00",
+    }
+    values.update(overrides)
+    values = {k: v for k, v in values.items() if v is not _OMIT}
+    columns = ", ".join(values)
+    placeholders = ", ".join("?" for _ in values)
+    result = _run(
+        database.execute(
+            f"INSERT INTO task_definitions ({columns}) VALUES ({placeholders})",
+            tuple(values.values()),
+        )
+    )
+    assert result.lastrowid is not None
+    return result.lastrowid
+
+
 # ---------------------------------------------------------------------------
 # DDL constants
 # ---------------------------------------------------------------------------
 
 
-def test_schema_v1_statements_compose_the_two_tables() -> None:
+def test_schema_v1_statements_compose_the_four_tables() -> None:
     assert SCHEMA_V1_STATEMENTS == [
         *SCHEMA_V1_CHILDREN_DDL,
         *SCHEMA_V1_ADMIN_USERS_DDL,
+        *SCHEMA_V1_SCHEDULE_RULES_DDL,
+        *SCHEMA_V1_TASK_DEFINITIONS_DDL,
     ]
 
 
@@ -121,6 +181,91 @@ def test_admin_users_columns_types_and_constraints(tmp_path) -> None:
         _run(database.close())
 
 
+def test_schedule_rules_table_exists_after_applying_ddl(tmp_path) -> None:
+    database = _open_db(tmp_path / "rules.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'schedule_rules'"
+            )
+        )
+        assert row is not None and row[0] == "schedule_rules"
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_table_exists_after_applying_ddl(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'task_definitions'"
+            )
+        )
+        assert row is not None and row[0] == "task_definitions"
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_columns_types_and_constraints(tmp_path) -> None:
+    database = _open_db(tmp_path / "rules-columns.db")
+    try:
+        _apply(database)
+        columns = _run(database.fetch_all("PRAGMA table_info(schedule_rules)"))
+        # cid, name, type, notnull, dflt_value, pk
+        assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
+            ("id", "INTEGER", 0, None, 1),
+            ("rule_type", "TEXT", 1, None, 0),
+            ("interval", "INTEGER", 1, "1", 0),
+            ("weekday_set", "TEXT", 0, None, 0),
+            ("day_of_month", "INTEGER", 0, None, 0),
+            ("nth_weekday", "INTEGER", 0, None, 0),
+            ("month", "INTEGER", 0, None, 0),
+            ("start_date", "TEXT", 1, None, 0),
+            ("end_date", "TEXT", 0, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_columns_types_and_constraints(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-columns.db")
+    try:
+        _apply(database)
+        columns = _run(database.fetch_all("PRAGMA table_info(task_definitions)"))
+        assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
+            ("id", "INTEGER", 0, None, 1),
+            ("title", "TEXT", 1, None, 0),
+            ("description", "TEXT", 0, None, 0),
+            ("icon", "TEXT", 0, None, 0),
+            ("child_id", "INTEGER", 1, None, 0),
+            ("schedule_rule_id", "INTEGER", 1, None, 0),
+            ("due_time", "TEXT", 0, None, 0),
+            ("is_active", "INTEGER", 1, "1", 0),
+            ("created_at", "TEXT", 1, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_foreign_keys_declared(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-fks.db")
+    try:
+        _apply(database)
+        fks = _run(database.fetch_all("PRAGMA foreign_key_list(task_definitions)"))
+        declared = {(row[2], row[3], row[4]) for row in fks}
+        assert declared == {
+            ("children", "child_id", "id"),
+            ("schedule_rules", "schedule_rule_id", "id"),
+        }
+    finally:
+        _run(database.close())
+
+
 # ---------------------------------------------------------------------------
 # Constraint enforcement
 # ---------------------------------------------------------------------------
@@ -158,6 +303,33 @@ def test_children_is_active_check_enforced(tmp_path) -> None:
         _run(database.close())
 
 
+@pytest.mark.parametrize(
+    ("column", "bad_value"),
+    [
+        ("sort_order", "abc"),
+        ("sort_order", 1.5),
+        ("is_active", "abc"),
+        ("is_active", 1.5),
+    ],
+)
+def test_children_integer_columns_reject_text_and_fraction(
+    tmp_path, column, bad_value
+) -> None:
+    database = _open_db(tmp_path / f"children-{column}-typeof.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _run(
+                database.execute(
+                    f"INSERT INTO children (display_name, {column}, created_at) "
+                    "VALUES (?, ?, ?)",
+                    ("Ada", bad_value, "2026-09-13T00:00:00+00:00"),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
 def test_children_created_at_not_null_enforced(tmp_path) -> None:
     database = _open_db(tmp_path / "created-at.db")
     try:
@@ -282,5 +454,464 @@ def test_failed_statement_rolls_back_whole_application(tmp_path) -> None:
             )
         )
         assert row is None, "partial DDL must not survive a failed application"
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# schedule_rules and task_definitions constraint enforcement
+# ---------------------------------------------------------------------------
+
+
+def test_schedule_rules_valid_insert_round_trips(tmp_path) -> None:
+    database = _open_db(tmp_path / "valid-rule.db")
+    try:
+        _apply(database)
+        rule_id = _insert_rule(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT rule_type, interval, weekday_set, start_date "
+                "FROM schedule_rules WHERE id = ?",
+                (rule_id,),
+            )
+        )
+        assert row == ("weekly", 1, "0,2,4", "2026-09-13")
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_interval_default_applies(tmp_path) -> None:
+    database = _open_db(tmp_path / "interval-default.db")
+    try:
+        _apply(database)
+        rule_id = _insert_rule(database, interval=_OMIT)
+        row = _run(
+            database.fetch_one(
+                "SELECT interval FROM schedule_rules WHERE id = ?", (rule_id,)
+            )
+        )
+        assert row == (1,)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_rule_type_check_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "rule-type.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, rule_type="biweekly", weekday_set=None)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_interval_zero_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "interval-zero.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, interval=0)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize("bad_interval", ["abc", 1.5])
+def test_schedule_rules_interval_non_integer_fails(
+    tmp_path, bad_interval
+) -> None:
+    database = _open_db(tmp_path / f"interval-{type(bad_interval).__name__}.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, interval=bad_interval)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_interval_integer_one_succeeds(tmp_path) -> None:
+    database = _open_db(tmp_path / "interval-one.db")
+    try:
+        _apply(database)
+        rule_id = _insert_rule(database, interval=1)
+        row = _run(
+            database.fetch_one(
+                "SELECT interval FROM schedule_rules WHERE id = ?", (rule_id,)
+            )
+        )
+        assert row == (1,)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize(
+    ("column", "bad_value"),
+    [
+        ("day_of_month", "abc"),
+        ("day_of_month", 1.5),
+        ("nth_weekday", "abc"),
+        ("nth_weekday", 1.5),
+        ("month", "abc"),
+        ("month", 1.5),
+    ],
+)
+def test_schedule_rules_integer_columns_reject_text_and_fraction(
+    tmp_path, column, bad_value
+) -> None:
+    database = _open_db(tmp_path / f"{column}-typeof.db")
+    try:
+        _apply(database)
+        overrides: dict[str, object] = {"rule_type": "monthly", "weekday_set": None}
+        overrides[column] = bad_value
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, **overrides)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize(
+    "bad_weekday_set",
+    ["", "x", "7", "0,x", "0,8", "0,", "0,1,2,3,4,5,6,0"],
+)
+def test_schedule_rules_weekday_set_invalid_csv_fails(
+    tmp_path, bad_weekday_set
+) -> None:
+    database = _open_db(tmp_path / "weekday-set-invalid.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, weekday_set=bad_weekday_set)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize("weekday_set", ["0", "6", "0,1,2,3,4,5,6"])
+def test_schedule_rules_weekday_set_valid_csv_succeeds(
+    tmp_path, weekday_set
+) -> None:
+    database = _open_db(tmp_path / "weekday-set-valid.db")
+    try:
+        _apply(database)
+        rule_id = _insert_rule(database, weekday_set=weekday_set)
+        row = _run(
+            database.fetch_one(
+                "SELECT weekday_set FROM schedule_rules WHERE id = ?", (rule_id,)
+            )
+        )
+        assert row == (weekday_set,)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_day_of_month_out_of_range_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "day-of-month.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(
+                database,
+                rule_type="monthly",
+                weekday_set=None,
+                day_of_month=32,
+            )
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_nth_weekday_zero_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "nth-weekday.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(
+                database,
+                rule_type="monthly",
+                weekday_set=None,
+                day_of_month=None,
+                nth_weekday=0,
+            )
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_nth_weekday_minus_one_last_week_allowed(tmp_path) -> None:
+    database = _open_db(tmp_path / "nth-weekday-last.db")
+    try:
+        _apply(database)
+        rule_id = _insert_rule(
+            database,
+            rule_type="monthly",
+            weekday_set=None,
+            day_of_month=None,
+            nth_weekday=-1,
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT nth_weekday FROM schedule_rules WHERE id = ?", (rule_id,)
+            )
+        )
+        assert row == (-1,)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_month_out_of_range_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "month.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(
+                database, rule_type="yearly", weekday_set=None, month=13
+            )
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_weekly_without_weekday_set_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "weekly-no-weekdays.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, rule_type="weekly", weekday_set=None)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_monthly_without_day_or_nth_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "monthly-no-fields.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(
+                database,
+                rule_type="monthly",
+                weekday_set=None,
+                day_of_month=None,
+                nth_weekday=None,
+            )
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_yearly_without_month_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "yearly-no-month.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(
+                database,
+                rule_type="yearly",
+                weekday_set=None,
+                day_of_month=None,
+                month=None,
+            )
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_daily_with_month_fields_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "daily-month-fields.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, rule_type="daily", month=3)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_end_date_before_start_date_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "end-before-start.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_rule(database, end_date="2026-09-12")
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_end_date_equal_to_start_date_allowed(tmp_path) -> None:
+    database = _open_db(tmp_path / "end-equals-start.db")
+    try:
+        _apply(database)
+        rule_id = _insert_rule(database, end_date="2026-09-13")
+        row = _run(
+            database.fetch_one(
+                "SELECT end_date FROM schedule_rules WHERE id = ?", (rule_id,)
+            )
+        )
+        assert row == ("2026-09-13",)
+    finally:
+        _run(database.close())
+
+
+def test_schedule_rules_start_date_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "start-date-null.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_rule(database, start_date=None)
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_insert_requires_existing_child_and_rule(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "definitions-fk-unknown.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        rule_id = _insert_rule(database)
+        definition_id = _insert_definition(
+            database, child_id=1, rule_id=rule_id
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT title, child_id, schedule_rule_id FROM task_definitions "
+                "WHERE id = ?",
+                (definition_id,),
+            )
+        )
+        assert row == ("Brush teeth", 1, rule_id)
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_unknown_child_id_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-bad-child.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _insert_definition(database, child_id=999)
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-bad-rule.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _insert_definition(database, child_id=1, rule_id=999)
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_is_active_check_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-is-active.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_definition(database, child_id=1, rule_id=1, is_active=2)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize("bad_is_active", ["abc", 1.5])
+def test_task_definitions_is_active_non_integer_fails(
+    tmp_path, bad_is_active
+) -> None:
+    database = _open_db(tmp_path / "definitions-is-active-typeof.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _insert_rule(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_definition(
+                database, child_id=1, rule_id=1, is_active=bad_is_active
+            )
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_title_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-title.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_definition(database, child_id=1, rule_id=1, title=None)
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_created_at_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-created-at.db")
+    try:
+        _apply(database)
+        _insert_rule(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_definition(
+                database, child_id=1, rule_id=1, created_at=None
+            )
+    finally:
+        _run(database.close())
+
+
+def test_task_definitions_defaults_apply_on_insert(tmp_path) -> None:
+    database = _open_db(tmp_path / "definitions-defaults.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _insert_rule(database)
+        definition_id = _insert_definition(
+            database, child_id=1, rule_id=1, is_active=_OMIT
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT is_active, due_time FROM task_definitions WHERE id = ?",
+                (definition_id,),
+            )
+        )
+        assert row == (1, None)
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# Idempotence across the extended v1 list
+# ---------------------------------------------------------------------------
+
+
+def test_applying_extended_ddl_twice_keeps_new_table_rows(tmp_path) -> None:
+    database = _open_db(tmp_path / "idempotent-extended.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _insert_rule(database)
+        _insert_definition(database, child_id=1, rule_id=1)
+        _apply(database)
+        counts = []
+        for table in ("children", "schedule_rules", "task_definitions"):
+            row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))
+            counts.append(row[0])
+        assert counts == [1, 1, 1]
     finally:
         _run(database.close())