feat: add schema v1 ddl for task_instances and completion_events (#16)

custom_components/nestquest/schema.py

@@ -85,6 +85,68 @@ unlike ``schedule_rules.end_date`` — an override is always a concrete
 date range; a single-day override stores the same date in both columns.
 ``is_present`` 0/1 marks the child absent/present for the whole range;
 ``note`` is optional free text.
+
+Task instances: ``task_instances`` deliberately carries NO completion
+status column.  An instance's current state derives from the latest row
+in ``completion_events`` (Feature 08), so materialization (Feature 07)
+can insert instances without knowing anything about completion and the
+append-only log stays the single source of truth.  ``definition_id``,
+``due_date`` and ``child_id`` snapshot the materialization-time facts
+("why this instance exists"): reassignment and rule edits change future
+instances only and never rewrite these rows.  ``due_time`` snapshots
+the definition's optional due time at generation for the same reason —
+the instance keeps its generation-time value even if the definition is
+edited later.  ``generated_at`` is a caller-set UTC ISO-8601 timestamp
+per the module timestamp policy (no DB default, so a generation batch
+stamps one coherent time).
+
+Uniqueness: the (definition_id, due_date) pair is declared as a
+table-level ``UNIQUE`` constraint rather than a separate ``CREATE
+UNIQUE INDEX`` statement.  SQLite materializes it as an implicit unique
+index (``sqlite_autoindex_task_instances_1``) — the required "unique
+index prevents duplicate instances" — and the constraint form cannot be
+silently dropped the way a standalone index can, so the duplicate door
+stays shut.  Idempotent materialization keys on exactly this pair.  No
+secondary indexes beyond it are declared yet: history and board queries
+(Features 08/10/13) should justify their indexes as a later migration
+once their query shapes exist.
+
+No ``ON DELETE`` clauses are declared on any foreign key in this
+module: children and definitions are deactivated, never deleted, so a
+deletion that would orphan instances or events is a bug and must fail
+loudly rather than cascade.
+
+Completion events: ``completion_events`` is strictly append-only.  That
+property is enforced at the DAO layer — the completion-events DAO
+exposes no UPDATE or DELETE path (task a9ea75fc) — and deliberately NOT
+by database triggers, matching the module's no-trigger policy: a RAISE
+trigger would be the one exception to "pure CHECK, no triggers",
+complicate the later migration runner, and still prove nothing (any
+writer can run raw SQL).  ``event_type`` is constrained to
+``completed``/``uncompleted``; a ``missed`` row is intentionally not
+representable — missed is a derived state announced on the HA event bus
+(Feature 10), never an appended event.
+
+``child_id`` is deliberately denormalized (it is also derivable via
+``task_instances``): history reporting (Feature 13) filters and rates
+per child directly against the log, and the log should stay
+self-describing without joins.
+
+Actor policy: ``actor_source`` distinguishes an authenticated HA user
+(``'user'``, with ``actor_user_id`` holding the HA user id) from a
+panel tap with no user context (``'panel'``, with ``actor_user_id``
+NULL — the tapped child profile is already ``child_id``).  A coherence
+CHECK makes the pair shape non-negotiable: ``'user'`` requires
+``actor_user_id``, ``'panel'`` forbids it.  Other actor sources are
+rejected; if an unattended system actor is ever needed, that is a
+schema migration, not a silent widening.
+
+On-time policy: ``was_on_time`` is required (integer 0 or 1) for
+``completed`` events — at completion time the due date is known and the
+flag is decidable.  For ``uncompleted`` events the flag is optional:
+Feature 08 decides whether a reversal carries the reversed completion's
+on-time flag or NULL, and the DDL deliberately accepts both rather than
+guessing.
 """
 from __future__ import annotations
 
@@ -209,6 +271,46 @@ SCHEMA_V1_PRESENCE_OVERRIDES_DDL: list[str] = [
     """,
 ]
 
+SCHEMA_V1_TASK_INSTANCES_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS task_instances (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        definition_id INTEGER NOT NULL REFERENCES task_definitions(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        due_date TEXT NOT NULL,
+        due_time TEXT,
+        generated_at TEXT NOT NULL,
+        UNIQUE (definition_id, due_date)
+    )
+    """,
+]
+
+SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS completion_events (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        instance_id INTEGER NOT NULL REFERENCES task_instances(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        event_type TEXT NOT NULL CHECK (event_type IN ('completed', 'uncompleted')),
+        actor_source TEXT NOT NULL CHECK (actor_source IN ('user', 'panel')),
+        actor_user_id TEXT,
+        occurred_at TEXT NOT NULL,
+        was_on_time INTEGER CHECK (
+            was_on_time IS NULL
+            OR (typeof(was_on_time) = 'integer' AND was_on_time IN (0, 1))
+        ),
+        CHECK (
+            (actor_source = 'user' AND actor_user_id IS NOT NULL)
+            OR (actor_source = 'panel' AND actor_user_id IS NULL)
+        ),
+        CHECK (
+            (event_type = 'completed' AND was_on_time IS NOT NULL)
+            OR (event_type = 'uncompleted')
+        )
+    )
+    """,
+]
+
 SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_CHILDREN_DDL,
     *SCHEMA_V1_ADMIN_USERS_DDL,
@@ -216,6 +318,8 @@ SCHEMA_V1_STATEMENTS: list[str] = [
     *SCHEMA_V1_TASK_DEFINITIONS_DDL,
     *SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
     *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
+    *SCHEMA_V1_TASK_INSTANCES_DDL,
+    *SCHEMA_V1_COMPLETION_EVENTS_DDL,
 ]
 
 

tests/test_schema.py

@@ -10,11 +10,13 @@ from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.schema import (
     SCHEMA_V1_ADMIN_USERS_DDL,
     SCHEMA_V1_CHILDREN_DDL,
+    SCHEMA_V1_COMPLETION_EVENTS_DDL,
     SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
     SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
     SCHEMA_V1_SCHEDULE_RULES_DDL,
     SCHEMA_V1_STATEMENTS,
     SCHEMA_V1_TASK_DEFINITIONS_DDL,
+    SCHEMA_V1_TASK_INSTANCES_DDL,
     async_apply_ddl,
 )
 
@@ -162,6 +164,8 @@ def test_schema_v1_statements_compose_the_six_tables() -> None:
         *SCHEMA_V1_TASK_DEFINITIONS_DDL,
         *SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
         *SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
+        *SCHEMA_V1_TASK_INSTANCES_DDL,
+        *SCHEMA_V1_COMPLETION_EVENTS_DDL,
     ]
 
 
@@ -1399,3 +1403,635 @@ def test_applying_presence_ddl_twice_keeps_rows(tmp_path) -> None:
         assert counts == [1, 1]
     finally:
         _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# task_instances and completion_events: helpers
+# ---------------------------------------------------------------------------
+
+
+def _insert_instance(database, definition_id=1, **overrides):
+    """Insert one valid task_instance, returning the instance id.
+
+    Pass ``_OMIT`` as a value to leave that column out of the INSERT so
+    its DDL DEFAULT applies.
+    """
+    values: dict[str, object] = {
+        "definition_id": definition_id,
+        "child_id": 1,
+        "due_date": "2026-09-14",
+        "generated_at": "2026-09-14T00: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_instances ({columns}) VALUES ({placeholders})",
+            tuple(values.values()),
+        )
+    )
+    assert result.lastrowid is not None
+    return result.lastrowid
+
+
+def _insert_event(database, instance_id=1, **overrides):
+    """Insert one valid completion_event, returning the event id.
+
+    Pass ``_OMIT`` as a value to leave that column out of the INSERT so
+    its DDL DEFAULT applies.
+    """
+    values: dict[str, object] = {
+        "instance_id": instance_id,
+        "child_id": 1,
+        "event_type": "completed",
+        "actor_source": "user",
+        "actor_user_id": "user-1",
+        "occurred_at": "2026-09-14T08:30:00+00:00",
+        "was_on_time": 1,
+    }
+    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 completion_events ({columns}) VALUES ({placeholders})",
+            tuple(values.values()),
+        )
+    )
+    assert result.lastrowid is not None
+    return result.lastrowid
+
+
+def _setup_definition(database) -> None:
+    """Create one child, one schedule rule and one task definition."""
+    _child(database)
+    _insert_rule(database)
+    _insert_definition(database, child_id=1, rule_id=1)
+
+
+# ---------------------------------------------------------------------------
+# task_instances and completion_events: DDL shape
+# ---------------------------------------------------------------------------
+
+
+def test_task_instances_table_exists_after_applying_ddl(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'task_instances'"
+            )
+        )
+        assert row is not None and row[0] == "task_instances"
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_table_exists_after_applying_ddl(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'completion_events'"
+            )
+        )
+        assert row is not None and row[0] == "completion_events"
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_columns_types_and_constraints(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-columns.db")
+    try:
+        _apply(database)
+        columns = _run(database.fetch_all("PRAGMA table_info(task_instances)"))
+        # 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),
+            ("definition_id", "INTEGER", 1, None, 0),
+            ("child_id", "INTEGER", 1, None, 0),
+            ("due_date", "TEXT", 1, None, 0),
+            ("due_time", "TEXT", 0, None, 0),
+            ("generated_at", "TEXT", 1, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_columns_types_and_constraints(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-columns.db")
+    try:
+        _apply(database)
+        columns = _run(database.fetch_all("PRAGMA table_info(completion_events)"))
+        assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
+            ("id", "INTEGER", 0, None, 1),
+            ("instance_id", "INTEGER", 1, None, 0),
+            ("child_id", "INTEGER", 1, None, 0),
+            ("event_type", "TEXT", 1, None, 0),
+            ("actor_source", "TEXT", 1, None, 0),
+            ("actor_user_id", "TEXT", 0, None, 0),
+            ("occurred_at", "TEXT", 1, None, 0),
+            ("was_on_time", "INTEGER", 0, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_foreign_keys_declared(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-fks.db")
+    try:
+        _apply(database)
+        fks = _run(database.fetch_all("PRAGMA foreign_key_list(task_instances)"))
+        declared = {(row[2], row[3], row[4]) for row in fks}
+        assert declared == {
+            ("task_definitions", "definition_id", "id"),
+            ("children", "child_id", "id"),
+        }
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_foreign_keys_declared(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-fks.db")
+    try:
+        _apply(database)
+        fks = _run(database.fetch_all("PRAGMA foreign_key_list(completion_events)"))
+        declared = {(row[2], row[3], row[4]) for row in fks}
+        assert declared == {
+            ("task_instances", "instance_id", "id"),
+            ("children", "child_id", "id"),
+        }
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-unique.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        rows = _run(
+            database.fetch_all(
+                "SELECT name, sql FROM sqlite_master WHERE type = 'index' "
+                "AND tbl_name = 'task_instances'"
+            )
+        )
+        implicit = [(name, sql) for name, sql in rows if sql is None]
+        assert len(implicit) == 1, (
+            "the (definition_id, due_date) UNIQUE constraint must be backed "
+            "by exactly one implicit unique index"
+        )
+        columns = _run(
+            database.fetch_all(
+                f"PRAGMA index_info('{implicit[0][0]}')"
+            )
+        )
+        assert [row[2] for row in columns] == ["definition_id", "due_date"]
+        # The implicit index is genuinely unique: UNIQUE(...) is the only
+        # constraint SQLite backs with a sql-less autoindex on this table.
+        row = _run(
+            database.fetch_one(
+                "SELECT COUNT(DISTINCT name) FROM pragma_index_list"
+                "('task_instances') WHERE \"unique\" = 1"
+            )
+        )
+        assert row == (1,)
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# task_instances: constraint enforcement
+# ---------------------------------------------------------------------------
+
+
+def test_task_instances_valid_insert_round_trips(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-valid.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1, child_id=1)
+        row = _run(
+            database.fetch_one(
+                "SELECT definition_id, child_id, due_date, due_time, generated_at "
+                "FROM task_instances WHERE id = ?",
+                (instance_id,),
+            )
+        )
+        assert row == (1, 1, "2026-09-14", None, "2026-09-14T00:00:00+00:00")
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_due_time_round_trips(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-due-time.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(
+            database, definition_id=1, due_time="17:30"
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT due_time FROM task_instances WHERE id = ?",
+                (instance_id,),
+            )
+        )
+        assert row == ("17:30",)
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_duplicate_definition_date_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-duplicate.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _insert_instance(database, definition_id=1, due_date="2026-09-14")
+        with pytest.raises(sqlite3.IntegrityError, match="UNIQUE"):
+            _insert_instance(database, definition_id=1, due_date="2026-09-14")
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_same_date_different_definitions_allowed(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "task-instances-multi-def.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Bo", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        rule_id = _insert_rule(database)
+        _insert_definition(database, child_id=2, rule_id=rule_id)
+        _insert_instance(database, definition_id=1, due_date="2026-09-14")
+        _insert_instance(database, definition_id=2, due_date="2026-09-14")
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM task_instances"))
+        assert row == (2,)
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_same_definition_different_dates_allowed(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-multi-date.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _insert_instance(database, definition_id=1, due_date="2026-09-14")
+        _insert_instance(database, definition_id=1, due_date="2026-09-15")
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM task_instances"))
+        assert row == (2,)
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_unknown_definition_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-bad-def.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _insert_instance(database, definition_id=999)
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_unknown_child_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-bad-child.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _insert_instance(database, definition_id=1, child_id=999)
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_due_date_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-due-null.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_instance(database, definition_id=1, due_date=None)
+    finally:
+        _run(database.close())
+
+
+def test_task_instances_generated_at_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-generated-null.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_instance(database, definition_id=1, generated_at=None)
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# completion_events: constraint enforcement
+# ---------------------------------------------------------------------------
+
+
+def test_completion_events_valid_user_completion_round_trips(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-valid.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        event_id = _insert_event(
+            database, instance_id=instance_id, child_id=1
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT instance_id, child_id, event_type, actor_source, "
+                "actor_user_id, occurred_at, was_on_time "
+                "FROM completion_events WHERE id = ?",
+                (event_id,),
+            )
+        )
+        assert row == (
+            instance_id,
+            1,
+            "completed",
+            "user",
+            "user-1",
+            "2026-09-14T08:30:00+00:00",
+            1,
+        )
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_valid_panel_completion_round_trips(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-panel.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        event_id = _insert_event(
+            database,
+            instance_id=instance_id,
+            actor_source="panel",
+            actor_user_id=None,
+            was_on_time=0,
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT actor_source, actor_user_id, was_on_time "
+                "FROM completion_events WHERE id = ?",
+                (event_id,),
+            )
+        )
+        assert row == ("panel", None, 0)
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_valid_uncompleted_round_trips(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-uncompleted.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        event_id = _insert_event(
+            database,
+            instance_id=instance_id,
+            event_type="uncompleted",
+            actor_source="user",
+            was_on_time=_OMIT,
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT event_type, actor_source, was_on_time "
+                "FROM completion_events WHERE id = ?",
+                (event_id,),
+            )
+        )
+        assert row == ("uncompleted", "user", None)
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_uncompleted_with_on_time_flag_allowed(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "completion-events-uncompleted-ontime.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        event_id = _insert_event(
+            database,
+            instance_id=instance_id,
+            event_type="uncompleted",
+            was_on_time=1,
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT was_on_time FROM completion_events WHERE id = ?",
+                (event_id,),
+            )
+        )
+        assert row == (1,)
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_unknown_instance_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-bad-instance.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _insert_event(database, instance_id=999)
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_unknown_child_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-bad-child.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
+            _insert_event(database, instance_id=instance_id, child_id=999)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize("bad_event_type", ["missed", "COMPLETE", ""])
+def test_completion_events_event_type_check_enforced(
+    tmp_path, bad_event_type
+) -> None:
+    database = _open_db(tmp_path / f"completion-events-type-{hash(bad_event_type)}.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_event(
+                database, instance_id=instance_id, event_type=bad_event_type
+            )
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize("bad_source", ["kiosk", "system", "USER", ""])
+def test_completion_events_actor_source_check_enforced(
+    tmp_path, bad_source
+) -> None:
+    database = _open_db(tmp_path / f"completion-events-source-{hash(bad_source)}.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_event(
+                database, instance_id=instance_id, actor_source=bad_source
+            )
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize(
+    ("actor_source", "actor_user_id"),
+    [
+        ("user", None),  # user without id
+        ("panel", "user-1"),  # panel with id
+    ],
+)
+def test_completion_events_actor_pair_coherence_enforced(
+    tmp_path, actor_source, actor_user_id
+) -> None:
+    database = _open_db(tmp_path / f"completion-events-actor-{actor_source}.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_event(
+                database,
+                instance_id=instance_id,
+                actor_source=actor_source,
+                actor_user_id=actor_user_id,
+            )
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize("bad_was_on_time", ["abc", 1.5, 2])
+def test_completion_events_was_on_time_non_integer_or_out_of_range_fails(
+    tmp_path, bad_was_on_time
+) -> None:
+    database = _open_db(tmp_path / f"completion-events-ontime-{type(bad_was_on_time).__name__}.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_event(
+                database, instance_id=instance_id, was_on_time=bad_was_on_time
+            )
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_completed_requires_was_on_time(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-completed-ontime-null.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_event(
+                database,
+                instance_id=instance_id,
+                event_type="completed",
+                was_on_time=_OMIT,
+            )
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_occurred_at_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-occurred-null.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_event(
+                database, instance_id=instance_id, occurred_at=None
+            )
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_multiple_events_per_instance_allowed(
+    tmp_path,
+) -> None:
+    database = _open_db(tmp_path / "completion-events-multi.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        _insert_event(
+            database, instance_id=instance_id, event_type="completed",
+            occurred_at="2026-09-14T08:30:00+00:00",
+        )
+        _insert_event(
+            database, instance_id=instance_id, event_type="uncompleted",
+            actor_source="user", was_on_time=1,
+            occurred_at="2026-09-14T09:00:00+00:00",
+        )
+        _insert_event(
+            database, instance_id=instance_id, event_type="completed",
+            occurred_at="2026-09-14T09:15:00+00:00",
+        )
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM completion_events"))
+        assert row == (3,)
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# Idempotence across the full eight-table v1 list
+# ---------------------------------------------------------------------------
+
+
+def test_applying_full_ddl_twice_keeps_rows(tmp_path) -> None:
+    database = _open_db(tmp_path / "idempotent-full.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        instance_id = _insert_instance(database, definition_id=1)
+        _insert_event(database, instance_id=instance_id)
+        _apply(database)
+        counts = []
+        for table in (
+            "children",
+            "schedule_rules",
+            "task_definitions",
+            "task_instances",
+            "completion_events",
+        ):
+            row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))
+            counts.append(row[0])
+        assert counts == [1, 1, 1, 1, 1]
+    finally:
+        _run(database.close())