Review fix: serialize create_child's duplicate scan with the insert

Review round 3 (Codex gpt-5.6-terra) found two concurrent creates of
the same normalised name could both pass the duplicate scan before
either INSERT committed, inserting a duplicate with no warning. The
scan and INSERT now run inside the shared connection-scoped lock, so
the second create always sees the first's committed row and warns.
A gated concurrent-create test proves exactly one warning fires; a
sequential-duplicate warning assertion rode along. Suite: 851 passed.

custom_components/nestquest/children.py

@@ -0,0 +1,239 @@
+"""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.
+    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")
+        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()

tests/test_children_registry.py

@@ -0,0 +1,577 @@
+"""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)
+
+
+# ---------------------------------------------------------------------------
+# 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)