Review fix: set_child_active rejects non-bool is_active

Review round 1 (Codex gpt-5.6-terra): the DAO's int() silently coerced
strings and numerics (1, 0, 1.0) into an unintended active state.
The business layer now requires a real bool, before the existence
check's caller-intent can be lost, with a coercible-input test matrix;
the refused transition leaves the record untouched. Suite: 860 passed.

custom_components/nestquest/children.py

@@ -0,0 +1,246 @@
+"""Business layer for the children registry (Feature 03).
+
+Sits between the Feature 09 service gate and the typed
+:class:`~.dao_children.ChildrenDao`: callers get validation, name
+policies and the duplicate-name warning here, and never talk to the
+DAO or SQL directly.  Children are never hard-deleted — deactivation
+is a separate explicit operation (:func:`set_child_active`), so
+completion history keeps its references (feature guardrail).
+
+Name policy: the display name is required and trimmed; an empty or
+whitespace-only name is rejected with ValueError.  A duplicate display
+name is ALLOWED (two children may legitimately share a nickname) but
+logged as a warning so an accidental double-add is visible in the HA
+log.  The duplicate check is case-insensitive and whitespace-normalised
+on both sides, because "ada" and "Ada " are the same name to a parent.
+
+Colour and avatar are optional free-form strings, trimmed when present;
+``sort_order`` is an int with the DAO/schema default of 0.  Clearing an
+optional field back to NULL is deliberately rejected: the DAO's update
+writes only provided fields (it cannot express NULL), so the business
+layer refuses an explicit ``None`` instead of silently ignoring it.
+Timestamps follow the module timestamp policy: the business layer
+stamps UTC ISO-8601 at second precision so every row it creates is
+uniform.
+"""
+from __future__ import annotations
+
+import datetime
+import logging
+
+from .dao_children import ChildRecord, ChildrenDao, _connection_lock
+from .db import NestQuestDatabase
+
+LOGGER = logging.getLogger(__name__)
+
+#: Sentinel distinguishing "argument omitted" from "explicit SQL NULL"
+#: in :func:`edit_child`.
+_UNSET = object()
+
+
+def _now_stamp() -> str:
+    """Return the module's strict UTC second-precision stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(
+        timespec="seconds"
+    )
+
+
+def _validate_text(
+    value: object, field: str, *, required: bool = False
+) -> str | None:
+    """Validate a free-text field and return it trimmed.
+
+    ``None`` raises when ``required`` and otherwise returns None.  A
+    non-string value is rejected; an empty result is rejected for
+    required fields (whitespace-only names are empty to a parent).
+    """
+    if value is None:
+        if required:
+            raise ValueError(f"{field} is required")
+        return None
+    if not isinstance(value, str):
+        raise ValueError(f"{field} must be a string, got {value!r}")
+    trimmed = value.strip()
+    if required and not trimmed:
+        raise ValueError(f"{field} is required")
+    return trimmed
+
+
+def _validate_sort_order(value: object) -> int:
+    """Reject non-int sort orders (bools included) before the DAO."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"sort_order must be an integer, got {value!r}")
+    return value
+
+
+async def _warn_on_duplicate_name(
+    database: NestQuestDatabase,
+    display_name: str,
+    *,
+    exclude_child_id: int | None = None,
+) -> None:
+    """Log a warning when ``display_name`` duplicates an existing child.
+
+    Case-insensitive and whitespace-normalised on both sides.  The
+    duplicate is still allowed — this only makes it visible; the log
+    line names both rows so an accidental double-add can be found.
+    ``exclude_child_id`` removes the child being edited from the scan,
+    so an edit only warns when the result collides with a DIFFERENT
+    child's name — including case-only edits, where the self-comparison
+    would otherwise hide a real duplicate.
+    """
+    lowered = display_name.casefold()
+    for child in await ChildrenDao(database).list_all():
+        if exclude_child_id is not None and child.id == exclude_child_id:
+            continue
+        if child.display_name.strip().casefold() == lowered:
+            LOGGER.warning(
+                "Child display name %r duplicates existing child %d "
+                "(%r); allowed, but check this was intentional",
+                display_name,
+                child.id,
+                child.display_name,
+            )
+            return
+
+
+async def create_child(
+    database: NestQuestDatabase,
+    display_name: str,
+    *,
+    colour: str | None = None,
+    avatar_ref: str | None = None,
+    sort_order: int = 0,
+) -> ChildRecord:
+    """Create a child profile and return the stored record.
+
+    The display name is required, trimmed, and rejected when empty.
+    A duplicate display name (case-insensitive) is allowed but logged
+    as a warning.  The scan and the INSERT run inside the
+    connection-scoped lock shared with the DAOs, so a concurrent
+    create of the same normalised name queues behind the first insert
+    and sees it — the duplicate still warns; it can never slip through
+    unlogged.
+    """
+    name = _validate_text(display_name, "display_name", required=True)
+    assert name is not None
+    colour_value = _validate_text(colour, "colour")
+    avatar_value = _validate_text(avatar_ref, "avatar_ref")
+    order = _validate_sort_order(sort_order)
+    async with _connection_lock(database):
+        await _warn_on_duplicate_name(database, name)
+        created = await ChildrenDao(database).create(
+            name,
+            _now_stamp(),
+            colour=colour_value,
+            avatar_ref=avatar_value,
+            sort_order=order,
+        )
+    return created
+
+
+async def edit_child(
+    database: NestQuestDatabase,
+    child_id: int,
+    *,
+    display_name: str | object = _UNSET,
+    colour: str | None | object = _UNSET,
+    avatar_ref: str | None | object = _UNSET,
+    sort_order: int | object = _UNSET,
+) -> ChildRecord:
+    """Edit a child profile and return the updated record.
+
+    Arguments default to the module sentinel ``_UNSET`` meaning "leave
+    this field alone".  An explicitly passed ``None`` for an optional
+    field raises: the DAO's update writes only provided fields and
+    cannot express NULL, so a silent no-op would mask the caller's
+    intent (see the module docstring).  A provided display name is
+    validated like create's; the duplicate scan excludes only the child
+    being edited, so a case-only edit colliding with another child's
+    normalised name still warns.  Raises ValueError when the child does
+    not exist.
+
+    Atomicity: the existence check, the duplicate scan, the UPDATE and
+    the readback all run inside the connection-scoped lock shared with
+    the DAOs, so another edit cannot slip between this call's UPDATE
+    and its readback — the returned record is exactly the one this
+    edit produced, not a later writer's.
+    """
+    async with _connection_lock(database):
+        dao = ChildrenDao(database)
+        existing = await dao.get(child_id)
+        if existing is None:
+            raise ValueError(f"child {child_id} does not exist")
+        updates: dict[str, object] = {}
+        if display_name is not _UNSET:
+            name = _validate_text(
+                display_name, "display_name", required=True
+            )
+            assert name is not None
+            await _warn_on_duplicate_name(
+                database, name, exclude_child_id=child_id
+            )
+            updates["display_name"] = name
+        if colour is not _UNSET:
+            value = _validate_text(colour, "colour")
+            if value is None:
+                raise ValueError(
+                    "clearing colour is not supported; pass a replacement "
+                    "value or omit the field"
+                )
+            updates["colour"] = value
+        if avatar_ref is not _UNSET:
+            value = _validate_text(avatar_ref, "avatar_ref")
+            if value is None:
+                raise ValueError(
+                    "clearing avatar_ref is not supported; pass a "
+                    "replacement value or omit the field"
+                )
+            updates["avatar_ref"] = value
+        if sort_order is not _UNSET:
+            updates["sort_order"] = _validate_sort_order(sort_order)
+        if updates:
+            await dao.update(child_id, **updates)
+        edited = await dao.get(child_id)
+    assert edited is not None
+    return edited
+
+
+async def set_child_active(
+    database: NestQuestDatabase, child_id: int, is_active: bool
+) -> ChildRecord:
+    """Deactivate or reactivate a child; returns the updated record.
+
+    Deactivation hides the child from the panel without deleting their
+    history (feature guardrail); this is the only removal path.
+    ``is_active`` must be a real bool: the DAO's ``int()`` would
+    otherwise silently coerce strings and numerics ("1", 0) into a
+    state the caller never asked for.  Raises ValueError when the
+    child does not exist.
+
+    Atomicity: existence check, UPDATE and readback run inside the
+    connection-scoped lock, so a concurrent opposite transition cannot
+    make this call report the other caller's value.
+    """
+    async with _connection_lock(database):
+        dao = ChildrenDao(database)
+        if await dao.get(child_id) is None:
+            raise ValueError(f"child {child_id} does not exist")
+        if not isinstance(is_active, bool):
+            raise ValueError(
+                f"is_active must be a real bool, got {is_active!r}"
+            )
+        await dao.set_active(child_id, is_active)
+        updated = await dao.get(child_id)
+    assert updated is not None
+    return updated
+
+
+async def list_children(
+    database: NestQuestDatabase, *, active_only: bool = False
+) -> list[ChildRecord]:
+    """Return children in display order (sort_order, then id)."""
+    dao = ChildrenDao(database)
+    if active_only:
+        return await dao.list_active()
+    return await dao.list_all()

custom_components/nestquest/dao_instances.py

@@ -106,6 +106,7 @@ class CompletionEventRecord:
     event_type: str
     actor_source: str
     actor_user_id: str | None
+    actor_child_id: int | None
     occurred_at: str
     was_on_time: bool | None
 
@@ -115,7 +116,7 @@ _INSTANCE_COLUMNS = (
 )
 _EVENT_COLUMNS = (
     "id, instance_id, child_id, event_type, actor_source, actor_user_id, "
-    "occurred_at, was_on_time"
+    "actor_child_id, occurred_at, was_on_time"
 )
 
 #: The only event types the append path accepts; the schema CHECK is
@@ -144,8 +145,9 @@ def _event_from_row(row: tuple) -> CompletionEventRecord:
         event_type=row[3],
         actor_source=row[4],
         actor_user_id=row[5],
-        occurred_at=row[6],
-        was_on_time=None if row[7] is None else bool(row[7]),
+        actor_child_id=row[6],
+        occurred_at=row[7],
+        was_on_time=None if row[8] is None else bool(row[8]),
     )
 
 
@@ -396,19 +398,21 @@ class CompletionEventsDao:
         was_on_time: bool | None,
         *,
         actor_user_id: str | None = None,
+        actor_child_id: int | None = None,
     ) -> CompletionEventRecord:
         """Append one completion or un-completion event.
 
-        ``actor_source`` is 'user' (with ``actor_user_id`` set) or
-        'panel' (no user id — the tapped child is ``child_id``); the
-        actor-pair CHECK enforces this shape at the schema level.
-        ``occurred_at`` must be the module's one strict UTC shape
-        (YYYY-MM-DDTHH:MM:SS+00:00) so lexicographic range queries
-        stay exact.  ``was_on_time`` is required (True/False) for
-        completions; for un-completions it may be None or the reversed
-        completion's flag (Feature 08 still deciding).  The instance
-        and child must exist — validated atomically under the
-        connection lock.
+        ``actor_source`` is 'user' (an admin action: ``actor_user_id``
+        set, ``actor_child_id`` NULL) or 'panel' (a panel tap: no user
+        id, and ``actor_child_id`` carrying the tapped child profile,
+        D-008); the actor-pair CHECKs enforce this shape at the schema
+        level.  ``occurred_at`` must be the module's one strict UTC
+        shape (YYYY-MM-DDTHH:MM:SS+00:00) so lexicographic range
+        queries stay exact.  ``was_on_time`` is required (True/False)
+        for completions; for un-completions it may be None or the
+        reversed completion's flag (Feature 08 still deciding).  The
+        instance, child and — for panel events — the tapped child must
+        exist — validated atomically under the connection lock.
         """
         if event_type not in (EVENT_COMPLETED, EVENT_UNCOMPLETED):
             raise ValueError(
@@ -428,6 +432,15 @@ class CompletionEventsDao:
             raise ValueError(
                 "actor_source 'panel' must not carry actor_user_id"
             )
+        if actor_source == "panel" and actor_child_id is None:
+            raise ValueError(
+                "actor_source 'panel' requires actor_child_id "
+                "(the tapped child profile)"
+            )
+        if actor_source == "user" and actor_child_id is not None:
+            raise ValueError(
+                "actor_source 'user' must not carry actor_child_id"
+            )
         if was_on_time is not None and type(was_on_time) is not bool:
             raise ValueError(
                 f"was_on_time must be True, False or None, got "
@@ -458,16 +471,28 @@ class CompletionEventsDao:
                 )
                 if child is None:
                     raise ValueError(f"child {child_id} does not exist")
+                if actor_child_id is not None:
+                    tapped = await self._database.fetch_one(
+                        "SELECT 1 FROM children WHERE id = ?",
+                        (actor_child_id,),
+                    )
+                    if tapped is None:
+                        raise ValueError(
+                            f"actor_child_id {actor_child_id} does not "
+                            "reference an existing child"
+                        )
                 result = await self._database.execute(
                     "INSERT INTO completion_events (instance_id, child_id, "
-                    "event_type, actor_source, actor_user_id, occurred_at, "
-                    "was_on_time) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    "event_type, actor_source, actor_user_id, "
+                    "actor_child_id, occurred_at, was_on_time) "
+                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                     (
                         instance_id,
                         child_id,
                         event_type,
                         actor_source,
                         actor_user_id,
+                        actor_child_id,
                         occurred_at,
                         None if was_on_time is None else int(was_on_time),
                     ),

custom_components/nestquest/migrations.py

@@ -25,11 +25,12 @@ pre-D-008 databases to the multi-assignee model: it creates
 ``quest_definition_assignees`` and, when ``quest_definitions`` still
 carries a single ``child_id`` column, rebuilds the table without it,
 copying each definition's assignee across.  Migration 4 adds the
-windows table and migration 5 rebuilds ``quest_instances`` onto the
-widened (definition_id, child_id, due_date, window) key.  Each callable
-migration runs inside its own transaction together with its version
-stamp, so a crash mid-step rolls the statements and the stamp back
-together.
+windows table, migration 5 rebuilds ``quest_instances`` onto the
+widened (definition_id, child_id, due_date, window) key, and migration
+6 rebuilds ``completion_events`` with the D-008 ``actor_child_id``
+column.  Each callable migration runs inside its own transaction
+together with its version stamp, so a crash mid-step rolls the
+statements and the stamp back together.
 
 Migration entries are either an ordered sequence of SQL statements or
 an async callable taking ``(database, target_version)``.  Callables
@@ -73,6 +74,7 @@ from typing import Union
 
 from .db import NestQuestDatabase
 from .schema import (
+    COMPLETION_EVENTS_TABLE_SQL,
     QUEST_DEFINITIONS_TABLE_SQL,
     QUEST_INSTANCES_TABLE_SQL,
     SCHEMA_V1_STATEMENTS,
@@ -326,6 +328,74 @@ async def _widen_instance_key(
         await database.execute("PRAGMA foreign_keys = ON")
 
 
+async def _add_actor_child_id(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Bring ``completion_events`` to the D-008 actor-child shape.
+
+    The ``actor_child_id`` column records which panel profile was
+    tapped; it cannot be added with ``ALTER TABLE ADD COLUMN`` alone
+    because the canonical shape also carries an actor-pair coherence
+    CHECK (panel events require it, user events forbid it), and CHECK
+    constraints cannot be added to an existing table.  Databases with
+    the legacy shape are therefore rebuilt via SQLite's canonical
+    procedure, backfilling each legacy panel row's ``actor_child_id``
+    to ``child_id`` — pre-D-008 panel completions were always recorded
+    against the tapped child themselves — while admin rows stay NULL.
+    Row ids are preserved so ``quest_instances`` references and the
+    append-only history stay intact.  Fresh databases already match
+    the canonical shape and only stamp.
+
+    The rebuild runs with ``foreign_keys = OFF`` and verifies
+    ``PRAGMA foreign_key_check`` inside the transaction before the
+    stamp.  The copy is an INSERT..SELECT, not an UPDATE: the
+    completion_events table keeps its append-only guarantee.
+    """
+    legacy_shape = not await _table_has_column(
+        database, "completion_events", "actor_child_id"
+    )
+    await database.execute("PRAGMA foreign_keys = OFF")
+    try:
+        async with database.transaction():
+            if legacy_shape:
+                await database.execute(
+                    "CREATE TABLE completion_events_rebuilt "
+                    f"{COMPLETION_EVENTS_TABLE_SQL}"
+                )
+                await database.execute(
+                    "INSERT INTO completion_events_rebuilt "
+                    "(id, instance_id, child_id, event_type, actor_source, "
+                    "actor_user_id, actor_child_id, occurred_at, "
+                    "was_on_time) "
+                    "SELECT id, instance_id, child_id, event_type, "
+                    "actor_source, actor_user_id, "
+                    "CASE WHEN actor_source = 'panel' THEN child_id "
+                    "ELSE NULL END, "
+                    "occurred_at, was_on_time "
+                    "FROM completion_events ORDER BY id"
+                )
+                await database.execute("DROP TABLE completion_events")
+                await database.execute(
+                    "ALTER TABLE completion_events_rebuilt "
+                    "RENAME TO completion_events"
+                )
+                violations = await database.fetch_all(
+                    "PRAGMA foreign_key_check"
+                )
+                if violations:
+                    raise RuntimeError(
+                        "completion_events rebuild produced foreign-key "
+                        f"violations: {violations!r}"
+                    )
+                LOGGER.info(
+                    "Added actor_child_id to completion_events; legacy "
+                    "panel rows backfilled to their own child_id"
+                )
+            await stamp_schema_version(database, target_version)
+    finally:
+        await database.execute("PRAGMA foreign_keys = ON")
+
+
 #: The ordered migration list.  Append-only: never edit an applied
 #: entry, add the next one instead.  (Migration 1's content was
 #: rewritten pre-release per D-007 — no version 1 database shipped.)
@@ -340,6 +410,10 @@ MIGRATIONS: Sequence[MigrationStep] = [
     # Migration 5 (D-008): rebuild quest_instances onto the widened
     # (definition_id, child_id, due_date, window) key.
     _widen_instance_key,
+    # Migration 6 (D-008): rebuild completion_events with the
+    # actor_child_id column and its actor-pair coherence CHECK,
+    # backfilling legacy panel rows to their own child_id.
+    _add_actor_child_id,
 ]
 
 

custom_components/nestquest/schema.py

@@ -159,11 +159,12 @@ 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.
+NULL — the tapped child profile is ``actor_child_id``, D-008).  Two
+coherence CHECKs make both shapes non-negotiable: ``'user'`` requires
+``actor_user_id`` and forbids ``actor_child_id``; ``'panel'`` requires
+``actor_child_id`` (the tapped profile) and forbids ``actor_user_id``.
+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
@@ -344,29 +345,42 @@ SCHEMA_V1_QUEST_INSTANCES_DDL: list[str] = [
     """,
 ]
 
-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 quest_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')
-        )
+#: Column body of ``completion_events``, shared between the v1 DDL
+#: below and the actor-child rebuild migration, which must recreate the
+#: table in exactly this shape (D-008: ``actor_child_id`` records which
+#: panel profile was tapped — set for panel events, NULL for admin
+#: ones; legacy panel rows backfill to ``child_id``, the child they
+#: always were).
+COMPLETION_EVENTS_TABLE_SQL = """(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    instance_id INTEGER NOT NULL REFERENCES quest_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,
+    actor_child_id INTEGER REFERENCES children(id),
+    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 (
+        (actor_source = 'user' AND actor_child_id IS NULL)
+        OR (actor_source = 'panel' AND actor_child_id IS NOT NULL)
+    ),
+    CHECK (
+        (event_type = 'completed' AND was_on_time IS NOT NULL)
+        OR (event_type = 'uncompleted')
     )
+)"""
+
+SCHEMA_V1_COMPLETION_EVENTS_DDL: list[str] = [
+    f"""
+    CREATE TABLE IF NOT EXISTS completion_events {COMPLETION_EVENTS_TABLE_SQL}
     """,
 ]
 

tests/test_children_registry.py

@@ -0,0 +1,706 @@
+"""Tests for children.py: the Feature 03 child-profile business layer."""
+from __future__ import annotations
+
+import asyncio
+import logging
+import re
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from custom_components.nestquest.children import (
+    create_child,
+    edit_child,
+    list_children,
+    set_child_active,
+)
+from custom_components.nestquest.dao_children import ChildRecord, ChildrenDao
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.migrations import apply_migrations
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_hass_mock():
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+def _with_db(tmp_path, name):
+    def _run_test(body):
+        async def _main():
+            database = NestQuestDatabase(_make_hass_mock())
+            await database.open(tmp_path / name)
+            try:
+                await apply_migrations(database)
+                return await body(database)
+            finally:
+                await database.close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# create_child: required, trimmed display name
+# ---------------------------------------------------------------------------
+
+
+def test_create_child_returns_typed_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(
+            database,
+            "  Declan  ",
+            colour="#2E7D32",
+            avatar_ref="avatars/declan.png",
+            sort_order=2,
+        )
+        assert isinstance(child, ChildRecord)
+        # The name is trimmed before storage.
+        assert child.display_name == "Declan"
+        assert child.colour == "#2E7D32"
+        assert child.avatar_ref == "avatars/declan.png"
+        assert child.sort_order == 2
+        assert child.is_active is True
+        # The business layer stamps created_at in strict UTC shape.
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00", child.created_at
+        )
+        return child
+
+    _with_db(tmp_path, "create-typed.db")(_body)
+
+
+def test_create_child_defaults(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Chloe")
+        assert child.display_name == "Chloe"
+        assert child.colour is None
+        assert child.avatar_ref is None
+        assert child.sort_order == 0
+        return child
+
+    _with_db(tmp_path, "create-defaults.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["", "   ", "\t\n", None])
+def test_create_child_missing_name_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="display_name is required"):
+            await create_child(database, bad)
+        assert await list_children(database) == []
+        return None
+
+    _with_db(tmp_path, "create-empty-name.db")(_body)
+
+
+@pytest.mark.parametrize("bad", [42, ["Ada"], 3.5])
+def test_create_child_non_string_name_rejected(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="must be a string"):
+            await create_child(database, bad)
+        return None
+
+    _with_db(tmp_path, "create-bad-name-type.db")(_body)
+
+
+def test_create_child_optional_text_fields_validated(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="colour must be a string"):
+            await create_child(database, "Ada", colour=7)
+        with pytest.raises(ValueError, match="avatar_ref must be a string"):
+            await create_child(database, "Ada", avatar_ref=1.5)
+        return None
+
+    _with_db(tmp_path, "create-bad-optional.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["2", 1.5, True, None])
+def test_create_child_sort_order_must_be_int(tmp_path, bad) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="sort_order must be an integer"):
+            await create_child(database, "Ada", sort_order=bad)
+        return None
+
+    _with_db(tmp_path, "create-bad-order.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# create_child: duplicate display name allowed but warned
+# ---------------------------------------------------------------------------
+
+
+def test_create_child_duplicate_name_allowed_and_warned(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        first = await create_child(database, "Ada")
+        with caplog.at_level(
+            logging.WARNING, logger="custom_components.nestquest.children"
+        ):
+            second = await create_child(database, "  ADA ")
+        # Whitespace/case-normalised duplicate: allowed, both rows exist.
+        assert first.display_name == "Ada"
+        assert second.display_name == "ADA"
+        names = [c.display_name for c in await list_children(database)]
+        assert names == ["Ada", "ADA"]
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1, (
+            "exactly one duplicate warning for the second create"
+        )
+        assert "Ada" in warnings[0].getMessage()
+        return second
+
+    _with_db(tmp_path, "create-duplicate.db")(_body)
+
+
+def test_create_child_distinct_name_logs_no_warning(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        await create_child(database, "Ada")
+        caplog.clear()
+        await create_child(database, "Jordyn")
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert warnings == []
+        return None
+
+    _with_db(tmp_path, "create-no-duplicate.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# edit_child: returns the updated record
+# ---------------------------------------------------------------------------
+
+
+def test_edit_child_updates_fields_and_returns_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        edited = await edit_child(
+            database,
+            child.id,
+            display_name="  Ada L. ",
+            colour="#222222",
+            avatar_ref="avatars/ada.png",
+            sort_order=3,
+        )
+        assert isinstance(edited, ChildRecord)
+        assert edited.id == child.id
+        assert edited.display_name == "Ada L."
+        assert edited.colour == "#222222"
+        assert edited.avatar_ref == "avatars/ada.png"
+        assert edited.sort_order == 3
+        # The stored row reflects every edit.
+        stored = await list_children(database)
+        assert len(stored) == 1
+        assert stored[0] == edited
+        return edited
+
+    _with_db(tmp_path, "edit-round-trip.db")(_body)
+
+
+def test_edit_child_omitted_fields_unchanged(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(
+            database, "Ada", colour="#111111", sort_order=5
+        )
+        edited = await edit_child(database, child.id, sort_order=9)
+        # Only sort_order moved; colour and name untouched.
+        assert edited.sort_order == 9
+        assert edited.colour == "#111111"
+        assert edited.display_name == "Ada"
+        return edited
+
+    _with_db(tmp_path, "edit-partial.db")(_body)
+
+
+def test_edit_child_no_fields_returns_current_record(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        edited = await edit_child(database, child.id)
+        assert edited == child
+        return edited
+
+    _with_db(tmp_path, "edit-noop.db")(_body)
+
+
+def test_edit_child_missing_raises(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="does not exist"):
+            await edit_child(database, 999, display_name="X")
+        return None
+
+    _with_db(tmp_path, "edit-missing.db")(_body)
+
+
+def test_edit_child_name_validation(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        for bad in ("", "   ", None, 42):
+            with pytest.raises(ValueError, match="display_name"):
+                await edit_child(database, child.id, display_name=bad)
+        # The failed edits left the record untouched.
+        assert (await list_children(database))[0].display_name == "Ada"
+        return None
+
+    _with_db(tmp_path, "edit-name-validation.db")(_body)
+
+
+def test_edit_child_sort_order_must_be_int(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="sort_order must be an integer"):
+            await edit_child(database, child.id, sort_order="3")
+        return None
+
+    _with_db(tmp_path, "edit-bad-order.db")(_body)
+
+
+def test_edit_child_explicit_none_clear_rejected(tmp_path) -> None:
+    """Clearing is refused loudly, not silently ignored: the DAO cannot
+    express NULL writes, so an explicit None must never masquerade as a
+    no-op."""
+    async def _body(database):
+        child = await create_child(database, "Ada", colour="#111111")
+        with pytest.raises(ValueError, match="clearing colour"):
+            await edit_child(database, child.id, colour=None)
+        with pytest.raises(ValueError, match="clearing avatar_ref"):
+            await edit_child(database, child.id, avatar_ref=None)
+        # The refused clears left the colour intact.
+        assert (await list_children(database))[0].colour == "#111111"
+        return None
+
+    _with_db(tmp_path, "edit-clear-unsupported.db")(_body)
+
+
+def test_edit_child_rename_to_duplicate_warns_but_different_name_not(
+    tmp_path, caplog
+) -> None:
+    async def _body(database):
+        ada = await create_child(database, "Ada")
+        bo = await create_child(database, "Bo")
+        caplog.clear()
+        # Renaming Bo onto Ada's name warns (different child's name).
+        edited = await edit_child(database, bo.id, display_name="ada")
+        assert edited.display_name == "ada"
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        assert "Ada" in warnings[0].getMessage()
+        return edited
+
+    _with_db(tmp_path, "edit-rename-duplicate.db")(_body)
+
+
+def test_edit_child_rename_to_own_name_casing_alone_warns_not(
+    tmp_path, caplog
+) -> None:
+    """Renaming a child to their own name (casing aside) is not a
+    duplicate when NO OTHER child carries the normalised name."""
+    async def _body(database):
+        ada = await create_child(database, "Ada")
+        caplog.clear()
+        edited = await edit_child(database, ada.id, display_name="ADA")
+        assert edited.display_name == "ADA"
+        assert not any(
+            "duplicates existing child" in r.getMessage()
+            for r in caplog.records
+        )
+        return edited
+
+    _with_db(tmp_path, "edit-rename-self.db")(_body)
+
+
+def test_create_child_concurrent_same_name_still_warns(tmp_path) -> None:
+    """Racing creates of the same normalised name cannot slip through
+    unlogged: the scan and INSERT are serialized on the connection
+    lock, so the second create sees the first's committed row and
+    warns.  Deterministic gate on task A's INSERT while task B queues
+    behind the lock.  Without the lock, B's scan would run before A's
+    INSERT commits and no warning would fire."""
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "create-race.db")
+        try:
+            await apply_migrations(database)
+            logger = logging.getLogger(
+                "custom_components.nestquest.children"
+            )
+            seen: list[logging.LogRecord] = []
+
+            class _Capture(logging.Handler):
+                def emit(self, record: logging.LogRecord) -> None:
+                    seen.append(record)
+
+            handler = _Capture()
+            logger.addHandler(handler)
+            try:
+                armed["active"] = True
+                task_a = asyncio.ensure_future(
+                    create_child(database, "Ada")
+                )
+                await a_started.wait()
+                armed["active"] = False
+                task_b = asyncio.ensure_future(
+                    create_child(database, " ada ")
+                )
+                await asyncio.sleep(0)
+                gate_open.set()
+                first, second = await asyncio.gather(task_a, task_b)
+            finally:
+                logger.removeHandler(handler)
+
+            # Both creates land; the duplicate is allowed.
+            assert first.display_name == "Ada"
+            assert second.display_name == "ada"
+            names = [c.display_name for c in await list_children(database)]
+            assert names == ["Ada", "ada"]
+            # Exactly one warning: B's scan saw A's committed row.
+            warnings = [
+                r for r in seen if "duplicates existing child" in r.getMessage()
+            ]
+            assert len(warnings) == 1, (
+                "the queued duplicate create must warn exactly once"
+            )
+            assert "Ada" in warnings[0].getMessage()
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_edit_child_case_only_edit_colliding_with_other_child_warns(
+    tmp_path, caplog
+) -> None:
+    """A case-only edit must not hide a real duplicate: with two Ada
+    profiles already present, editing either one's name to another
+    casing still warns (the scan excludes only the edited child)."""
+    async def _body(database):
+        first = await create_child(database, "Ada")
+        await create_child(database, "ada")
+        caplog.clear()
+        edited = await edit_child(database, first.id, display_name="ADA")
+        assert edited.display_name == "ADA"
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        # Editing the name to a value that STAYS duplicated (no change,
+        # other child still carries it) warns too: the scan is on the
+        # resulting state, not on whether the word moved.
+        caplog.clear()
+        await edit_child(database, first.id, display_name=" ada ")
+        warnings = [
+            r for r in caplog.records if "duplicates existing child" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        return None
+
+    _with_db(tmp_path, "edit-case-only-duplicate.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# set_child_active / list_children
+# ---------------------------------------------------------------------------
+
+
+def test_set_child_active_round_trip(tmp_path) -> None:
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        deactivated = await set_child_active(database, child.id, False)
+        assert deactivated.is_active is False
+        assert [c.id for c in await list_children(database)] == [child.id]
+        assert await list_children(database, active_only=True) == []
+        reactivated = await set_child_active(database, child.id, True)
+        assert reactivated.is_active is True
+        assert len(await list_children(database, active_only=True)) == 1
+        return reactivated
+
+    _with_db(tmp_path, "set-active.db")(_body)
+
+
+def test_set_child_active_missing_raises(tmp_path) -> None:
+    async def _body(database):
+        with pytest.raises(ValueError, match="does not exist"):
+            await set_child_active(database, 999, False)
+        return None
+
+    _with_db(tmp_path, "set-active-missing.db")(_body)
+
+
+@pytest.mark.parametrize("bad", ["1", "0", 1, 0, 1.0, None, "true"])
+def test_set_child_active_rejects_non_bool(tmp_path, bad) -> None:
+    """The DAO's int() would silently coerce "1"/0/1.0 into a state the
+    caller never asked for; the business layer refuses non-bools."""
+    async def _body(database):
+        child = await create_child(database, "Ada")
+        with pytest.raises(ValueError, match="is_active must be a real bool"):
+            await set_child_active(database, child.id, bad)
+        # The refused transition left the child untouched.
+        assert (await list_children(database))[0].is_active is True
+        return None
+
+    _with_db(tmp_path, "set-active-non-bool.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# Concurrency: edit/set_active are atomic under the connection lock
+# ---------------------------------------------------------------------------
+
+
+def _gated_hass(armed: dict, gate_open: asyncio.Event, a_started: asyncio.Event):
+    """A hass mock whose executor gates the FIRST `_execute` job after
+    arming — the mutate statement of whichever business call starts
+    first — proving the caller holds the connection lock at that
+    moment and the second caller must queue behind it."""
+    from unittest.mock import MagicMock
+
+    hass = MagicMock()
+
+    async def _gated_executor(fn, *args):
+        if armed["active"]:
+            if getattr(fn, "__name__", "") == "_execute" and not armed["gated"]:
+                armed["gated"] = True
+                a_started.set()
+                await gate_open.wait()
+        return fn(*args)
+
+    hass.async_add_executor_job = _gated_executor
+    return hass
+
+
+def test_edit_child_concurrent_opposite_edits_return_own_values(
+    tmp_path,
+) -> None:
+    """Two racing edits cannot cross-contaminate each other's readback.
+
+    Deterministic gate: task A is paused at its UPDATE (it provably
+    holds the connection lock), task B is then created and must queue;
+    A's readback runs before B's edit, so A returns A's value and B
+    returns B's.  Without the lock, A's readback would see B's write.
+    """
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "edit-race.db")
+        try:
+            await apply_migrations(database)
+            child = await create_child(
+                database, "Ada", colour="#111111", sort_order=0
+            )
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                edit_child(database, child.id, colour="#AAAAAA", sort_order=1)
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                edit_child(
+                    database, child.id, colour="#BBBBBB", sort_order=2
+                )
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+
+            # Each caller gets back exactly what it wrote.
+            assert record_a.colour == "#AAAAAA"
+            assert record_a.sort_order == 1
+            assert record_b.colour == "#BBBBBB"
+            assert record_b.sort_order == 2
+            # B landed last: the stored row is B's (read via the DAO —
+            # the children leak guard forbids raw probes here).
+            stored = await ChildrenDao(database).get(child.id)
+            assert stored is not None
+            assert (stored.colour, stored.sort_order) == ("#BBBBBB", 2)
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_set_child_active_concurrent_opposite_transitions_return_own(
+    tmp_path,
+) -> None:
+    """Racing activate/deactivate calls each report their own request.
+
+    Same deterministic gate on task A's UPDATE; task B queues behind
+    the connection lock and lands the opposite transition after A.
+    """
+    async def _main():
+        armed = {"active": False, "gated": False}
+        gate_open = asyncio.Event()
+        a_started = asyncio.Event()
+        hass = _gated_hass(armed, gate_open, a_started)
+        database = NestQuestDatabase(hass)
+        await database.open(tmp_path / "active-race.db")
+        try:
+            await apply_migrations(database)
+            child = await create_child(database, "Ada")
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                set_child_active(database, child.id, False)
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                set_child_active(database, child.id, True)
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            record_a, record_b = await asyncio.gather(task_a, task_b)
+
+            assert record_a.is_active is False
+            assert record_b.is_active is True
+            stored = await ChildrenDao(database).get(child.id)
+            assert stored is not None
+            assert stored.is_active is True
+        finally:
+            await database.close()
+
+    _run(_main())
+
+
+def test_list_children_orders_by_sort_order_then_id(tmp_path) -> None:
+    async def _body(database):
+        second = await create_child(database, "Bo", sort_order=1)
+        first = await create_child(database, "Ada", sort_order=0)
+        third = await create_child(database, "Cleo", sort_order=1)
+        listed = await list_children(database)
+        # sort_order 0 first, then the two sort_order-1 rows by id.
+        assert [c.id for c in listed] == [first.id, second.id, third.id]
+        return listed
+
+    _with_db(tmp_path, "list-order.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# Deactivation keeps every referencing row intact (Feature 03)
+# ---------------------------------------------------------------------------
+
+
+def test_deactivating_child_preserves_definitions_instances_events(
+    tmp_path,
+) -> None:
+    async def _body(database):
+        from datetime import date, timedelta
+
+        from custom_components.nestquest.dao_children import ChildrenDao
+        from custom_components.nestquest.dao_instances import (
+            CompletionEventsDao,
+            QuestInstancesDao,
+        )
+        from custom_components.nestquest.dao_rules import (
+            QuestDefinitionsDao,
+            ScheduleRulesDao,
+        )
+
+        children = ChildrenDao(database)
+        rules = ScheduleRulesDao(database)
+        definitions = QuestDefinitionsDao(database)
+        instances = QuestInstancesDao(database)
+        events = CompletionEventsDao(database)
+
+        child = await create_child(database, "Ada")
+        rule = await rules.create("daily", "2026-01-01")
+        definition = await definitions.create(
+            "Brush teeth",
+            rule.id,
+            "2026-09-15T12:00:00+00:00",
+            assignee_child_ids=[child.id],
+        )
+        due = (date.today() + timedelta(days=1)).isoformat()
+        stamp = f"{due}T08:00:00+00:00"
+        instance = await instances.upsert(
+            definition.id, child.id, due, stamp, window="morning"
+        )
+        event = await events.append(
+            instance.id,
+            child.id,
+            "completed",
+            "panel",
+            f"{due}T09:00:00+00:00",
+            True,
+            actor_child_id=child.id,
+        )
+
+        # Deactivate: the referencing rows must not move.
+        deactivated = await set_child_active(database, child.id, False)
+        assert deactivated.is_active is False
+        assert await children.get(child.id) == deactivated
+        assert (await definitions.get(definition.id)) == definition
+        assert (await instances.get_by_id(instance.id)) == instance
+        assert await events.list_by_instance(instance.id) == [event]
+        # Excluded from the active list, still present in the full list.
+        assert await list_children(database, active_only=True) == []
+        assert [c.id for c in await list_children(database)] == [child.id]
+
+        # Reactivate: fully restored.
+        reactivated = await set_child_active(database, child.id, True)
+        assert reactivated.is_active is True
+        assert [c.id for c in await list_children(database, active_only=True)] == [
+            child.id
+        ]
+        assert (await instances.get_by_id(instance.id)) == instance
+        return reactivated
+
+    _with_db(tmp_path, "deactivate-data-intact.db")(_body)
+
+
+def test_no_code_path_deletes_a_child_row() -> None:
+    """Structural guardrail: children are deactivated, never deleted.
+
+    The DAO exposes no delete method at all, and no module in the
+    package (nor any test outside the sanctioned DDL/migration layers,
+    which create the tables) carries a statement that removes child
+    rows — the leak guard in test_dao_children enforces the file
+    allowlist, this asserts the no-delete property itself across every
+    file that may legitimately name the table.
+    """
+    import re
+    from pathlib import Path
+
+    package = Path(
+        __import__(
+            "custom_components.nestquest", fromlist=["__file__"]
+        ).__file__
+    ).parent
+
+    delete_pattern = re.compile(
+        r"DELETE\s+(FROM\s+)?((main|temp)\s*\.\s*)?[`'\"]*(\[)?children\b",
+        re.IGNORECASE,
+    )
+    for py in sorted(package.rglob("*.py")):
+        assert delete_pattern.search(py.read_text()) is None, (
+            f"a code path in {py.name} deletes child rows"
+        )
+
+    import inspect
+
+    from custom_components.nestquest.dao_children import ChildrenDao
+
+    methods = {
+        name
+        for name, _ in inspect.getmembers(ChildrenDao, inspect.isfunction)
+    }
+    assert not any(
+        name.startswith("delete") or name == "remove" for name in methods
+    ), f"ChildrenDao exposed a delete-shaped method: {methods}"

tests/test_dao_instances.py

@@ -664,6 +664,7 @@ def test_append_panel_event_round_trips(tmp_path) -> None:
             definition.id, child.id, D1, _now_stamp(), window="morning")
         event = await events.append(
             instance.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id,
             actor_user_id=None,
         )
         assert event.actor_source == "panel"
@@ -718,17 +719,31 @@ def test_append_validation_rejects_bad_shapes(tmp_path) -> None:
         with pytest.raises(ValueError, match="must not carry"):
             await events.append(
                 instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id,
                 actor_user_id="user-1",
             )
+        with pytest.raises(ValueError, match="requires actor_child_id"):
+            await events.append(
+                instance.id, child.id, "completed", "panel",
+                _now_stamp(), True,
+            )
+        with pytest.raises(ValueError, match="must not carry actor_child_id"):
+            await events.append(
+                instance.id, child.id, "completed", "user", _now_stamp(), True,
+                actor_user_id="user-1",
+                actor_child_id=child.id,
+            )
         with pytest.raises(ValueError, match="requires was_on_time"):
             await events.append(
-                instance.id, child.id, "completed", "panel", _now_stamp(), None
+                instance.id, child.id, "completed", "panel", _now_stamp(), None,
+                actor_child_id=child.id
             )
         # was_on_time must be a real bool: int-convertible junk that
         # would silently coerce (0.5 -> False) is rejected outright.
         with pytest.raises(ValueError, match="must be True, False or None"):
             await events.append(
-                instance.id, child.id, "completed", "panel", _now_stamp(), 0.5
+                instance.id, child.id, "completed", "panel",
+                _now_stamp(), 0.5, actor_child_id=child.id,
             )
         with pytest.raises(ValueError, match="must be True, False or None"):
             await events.append(
@@ -758,7 +773,7 @@ def test_append_rejects_non_utc_timestamps(tmp_path) -> None:
             with pytest.raises(ValueError, match="UTC"):
                 await events.append(
                     instance.id, child.id, "completed", "panel", bad,
-                    True,
+                    True, actor_child_id=child.id,
                 )
         return None
 
@@ -815,14 +830,16 @@ def test_completed_uncompleted_recompleted_three_ordered_events(
         instance = await instances.upsert(
             definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), True
+            instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), False
+            instance.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id
         )
         history = await events.list_by_instance(instance.id)
         assert [e.event_type for e in history] == [
@@ -855,7 +872,8 @@ def test_get_latest_returns_uncompleted_when_reversed(tmp_path) -> None:
         instance = await instances.upsert(
             definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
-            instance.id, child.id, "completed", "panel", _now_stamp(), True
+            instance.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), True,
@@ -878,17 +896,20 @@ def test_list_by_child_and_date_range_scopes_by_due_date(tmp_path) -> None:
         late = await instances.upsert(
             definition.id, child.id, D20, _now_stamp(), window="morning")
         await events.append(
-            early.id, child.id, "completed", "panel", _now_stamp(), True
+            early.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         await events.append(
-            inside.id, child.id, "completed", "panel", _now_stamp(), False
+            inside.id, child.id, "completed", "panel", _now_stamp(), False,
+                actor_child_id=child.id
         )
         await events.append(
             inside.id, child.id, "uncompleted", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         await events.append(
-            late.id, child.id, "completed", "panel", _now_stamp(), True
+            late.id, child.id, "completed", "panel", _now_stamp(), True,
+                actor_child_id=child.id
         )
         found = await events.list_by_child_and_date_range(
             child.id, D2, D2
@@ -922,7 +943,7 @@ def test_list_events_occurred_at_scope_variant(tmp_path) -> None:
         # queried on today's date finds it too.
         event = await events.append(
             late_due.id, child.id, "completed", "panel",
-            _now_stamp(), False,
+            _now_stamp(), False, actor_child_id=child.id,
         )
         by_due = await events.list_by_child_and_date_range(
             child.id, D20, D20
@@ -1010,7 +1031,12 @@ def test_no_mutation_sql_for_completion_events_anywhere() -> None:
     # UPDATE/DELETE/non-DAO-INSERT paths hit the pattern.
     insert_exempt = {
         "custom_components/nestquest/dao_instances.py",
+        "custom_components/nestquest/migrations.py",  # the rebuild
+        # migrations re-create the table via INSERT..SELECT (append-
+        # order preserved); UPDATE/DELETE stays forbidden everywhere
         "tests/test_schema.py",
+        "tests/test_migrations.py",  # seeds frozen legacy shapes to
+        # exercise the rebuild migrations
         "tests/test_dao_instances.py",  # this guard file; scanned by
         # its own guard's regex spans below instead
     }

tests/test_dao_matrix.py

@@ -372,14 +372,16 @@ def test_matrix_completion_events_append_read_and_no_mutation(
     async def _body(w: World):
         # insert: append a completion then a reversal, then re-complete
         await w.events.append(
-            w.instance.id, w.child.id, "completed", "panel", _stamp(), True
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), True,
+            actor_child_id=w.child.id,
         )
         await w.events.append(
             w.instance.id, w.child.id, "uncompleted", "user", _stamp(), None,
             actor_user_id="user-1",
         )
         await w.events.append(
-            w.instance.id, w.child.id, "completed", "panel", _stamp(), False
+            w.instance.id, w.child.id, "completed", "panel", _stamp(), False,
+            actor_child_id=w.child.id,
         )
         # read: full ordered history + latest
         history = await w.events.list_by_instance(w.instance.id)

tests/test_migrations.py

@@ -149,16 +149,17 @@ def test_migration_list_shape() -> None:
     (D-007) to create the quest_* tables directly; migration 2 exists
     for dev databases stamped version 1 by the pre-rename runner,
     migration 3 brings pre-D-008 definitions to the multi-assignee
-    model, migration 4 adds the windows table, and migration 5 rebuilds
+    model, migration 4 adds the windows table, migration 5 rebuilds
     quest_instances onto the widened (definition, child, date, window)
-    key.
+    key, and migration 6 adds actor_child_id to completion_events.
     """
     assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
     assert callable(MIGRATIONS[1])
     assert callable(MIGRATIONS[2])
     assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
     assert callable(MIGRATIONS[4])
-    assert len(MIGRATIONS) == 5
+    assert callable(MIGRATIONS[5])
+    assert len(MIGRATIONS) == 6
 
 
 # ---------------------------------------------------------------------------
@@ -442,6 +443,21 @@ _V2_INSTANCES_DDL = [
     """,
 ]
 
+_V2_EVENTS_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS completion_events (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        instance_id INTEGER NOT NULL REFERENCES quest_instances(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        event_type TEXT NOT NULL,
+        actor_source TEXT NOT NULL,
+        actor_user_id TEXT,
+        occurred_at TEXT NOT NULL,
+        was_on_time INTEGER
+    )
+    """,
+]
+
 
 def _seed_v2_multi_assignee_upgrade(database) -> None:
     """Create a database shaped like a post-rename, pre-D-008 dev file.
@@ -458,12 +474,13 @@ def _seed_v2_multi_assignee_upgrade(database) -> None:
         "quest_definition_assignees ",
         "quest_definition_windows ",
         "quest_instances ",
+        "completion_events ",
     )
     for sql in SCHEMA_V1_STATEMENTS:
         if any(name in sql for name in _v2_replaced):
             continue
         _run(database.execute(sql))
-    for sql in _V2_DEFINITIONS_DDL + _V2_INSTANCES_DDL:
+    for sql in _V2_DEFINITIONS_DDL + _V2_INSTANCES_DDL + _V2_EVENTS_DDL:
         _run(database.execute(sql))
     _run(database.execute(VERSION_TABLE_DDL))
     _run(
@@ -503,6 +520,26 @@ def _seed_v2_multi_assignee_upgrade(database) -> None:
             "VALUES (1, 2, '2026-09-15', '2026-09-14T00:00:00+00:00')"
         )
     )
+    # One legacy panel event and one legacy admin event: the panel one
+    # must backfill actor_child_id to its own child; the admin one must
+    # stay NULL.
+    _run(
+        database.execute(
+            "INSERT INTO completion_events (instance_id, child_id, "
+            "event_type, actor_source, occurred_at, was_on_time) "
+            "VALUES (1, 2, 'completed', 'panel', "
+            "'2026-09-15T10:00:00+00:00', 1)"
+        )
+    )
+    _run(
+        database.execute(
+            "INSERT INTO completion_events (instance_id, child_id, "
+            "event_type, actor_source, actor_user_id, occurred_at, "
+            "was_on_time) "
+            "VALUES (1, 2, 'uncompleted', 'user', 'parent-1', "
+            "'2026-09-15T11:00:00+00:00', 1)"
+        )
+    )
 
 
 def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
@@ -559,6 +596,23 @@ def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
         assert [
             "definition_id", "child_id", "due_date", "window"
         ] in unique_columns
+        # The completion events survived the actor_child_id rebuild:
+        # the panel row backfilled to its own child, the admin row
+        # stayed NULL, and ids are preserved.
+        events = _run(
+            database.fetch_all(
+                "SELECT id, event_type, actor_source, actor_child_id "
+                "FROM completion_events ORDER BY id"
+            )
+        )
+        assert events == [
+            (1, "completed", "panel", 2),
+            (2, "uncompleted", "user", None),
+        ]
+        columns = _run(
+            database.fetch_all("PRAGMA table_info(completion_events)")
+        )
+        assert "actor_child_id" in {c[1] for c in columns}
         # No dangling references anywhere.
         assert _run(database.fetch_all("PRAGMA foreign_key_check")) == []
         # The rebuilt table is live for the new model: a definition

tests/test_schema.py

@@ -1851,6 +1851,7 @@ def test_completion_events_columns_types_and_constraints(tmp_path) -> None:
             ("event_type", "TEXT", 1, None, 0),
             ("actor_source", "TEXT", 1, None, 0),
             ("actor_user_id", "TEXT", 0, None, 0),
+            ("actor_child_id", "INTEGER", 0, None, 0),
             ("occurred_at", "TEXT", 1, None, 0),
             ("was_on_time", "INTEGER", 0, None, 0),
         ]
@@ -1881,6 +1882,7 @@ def test_completion_events_foreign_keys_declared(tmp_path) -> None:
         assert declared == {
             ("quest_instances", "instance_id", "id"),
             ("children", "child_id", "id"),
+            ("children", "actor_child_id", "id"),
         }
     finally:
         _run(database.close())
@@ -2182,16 +2184,71 @@ def test_completion_events_valid_panel_completion_round_trips(tmp_path) -> None:
             instance_id=instance_id,
             actor_source="panel",
             actor_user_id=None,
+            actor_child_id=1,
             was_on_time=0,
         )
         row = _run(
             database.fetch_one(
-                "SELECT actor_source, actor_user_id, was_on_time "
-                "FROM completion_events WHERE id = ?",
+                "SELECT actor_source, actor_user_id, actor_child_id, "
+                "was_on_time FROM completion_events WHERE id = ?",
                 (event_id,),
             )
         )
-        assert row == ("panel", None, 0)
+        assert row == ("panel", None, 1, 0)
+    finally:
+        _run(database.close())
+
+
+@pytest.mark.parametrize(
+    ("actor_source", "actor_user_id", "actor_child_id"),
+    [
+        ("user", "user-1", 1),  # admin event must not carry a profile
+        ("panel", None, None),  # panel event requires the profile
+        ("panel", "user-1", 1),  # panel must not carry a user id either
+        ("panel", "user-1", None),
+        ("user", None, None),
+    ],
+)
+def test_completion_events_actor_child_coherence_enforced(
+    tmp_path, actor_source, actor_user_id, actor_child_id
+) -> None:
+    """D-008 actor-pair shape: 'user' events carry actor_user_id and no
+    actor_child_id; 'panel' events carry actor_child_id and no
+    actor_user_id.  The CHECKs make both halves non-negotiable."""
+    database = _open_db(tmp_path / "completion-events-coherence.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",
+                actor_source=actor_source,
+                actor_user_id=actor_user_id,
+                actor_child_id=actor_child_id,
+                was_on_time=1,
+            )
+    finally:
+        _run(database.close())
+
+
+def test_completion_events_actor_child_unknown_child_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "completion-events-bad-actor-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,
+                actor_source="panel",
+                actor_user_id=None,
+                actor_child_id=999,
+                was_on_time=1,
+            )
     finally:
         _run(database.close())