fix: atomic complete_instance check+append

custom_components/nestquest/completion.py

@@ -0,0 +1,316 @@
+"""Business layer for the completion event log (Feature 08).
+
+Sits between the Feature 09 service gate and the typed
+:class:`~.dao_instances.CompletionEventsDao`: callers get actor-shape
+validation and the UTC occurred-at stamp here, and never talk to the
+DAO or SQL directly.  This module is the only Feature-08 caller of
+:meth:`~.dao_instances.CompletionEventsDao.append`.  Completing an
+instance (no-op if already done) is :func:`complete_instance`;
+un-completing arrives in a later task.  Instance current state is
+derived from the latest event via :func:`instance_state`
+(``open`` / ``done``); there is no status column.
+
+Actor policy matches the schema CHECKs (D-008): ``actor_source`` is
+``'user'`` or ``'panel'`` only — never ``'service'``.  ``'user'``
+requires ``actor_user_id`` and forbids ``actor_child_id``; ``'panel'``
+requires ``actor_child_id`` (the tapped profile) and forbids
+``actor_user_id``.  Inverted shapes raise a field-naming ValueError
+before any write.
+
+``occurred_at`` is stamped UTC ISO-8601 at second precision with an
+explicit ``+00:00`` offset, matching the DAO's one accepted shape so
+lexicographic range queries stay exact.  ``was_on_time`` is required
+(True/False) for completed events; un-completions may pass None.  There
+is no permission check here (that is Feature 09).
+"""
+from __future__ import annotations
+
+import asyncio
+import datetime
+
+from .dao_children import _CONNECTION_LOCKS, _connection_lock
+from .dao_instances import (
+    EVENT_COMPLETED,
+    EVENT_UNCOMPLETED,
+    CompletionEventRecord,
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
+from .db import NestQuestDatabase
+
+#: Timestamp policy for completion_events.occurred_at: strict UTC
+#: ISO-8601 with explicit '+00:00' offset (module timestamp policy).
+_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
+
+
+def _now_stamp() -> str:
+    """Return the module's strict UTC second-precision stamp."""
+    return datetime.datetime.now(datetime.timezone.utc).strftime(
+        _UTC_TIMESTAMP_FORMAT
+    )
+
+
+def _stamp_occurred_at(now: datetime.datetime | None) -> str:
+    """Return a DAO-accepted UTC stamp for ``now``, or the current time."""
+    if now is None:
+        return _now_stamp()
+    if not isinstance(now, datetime.datetime):
+        raise ValueError(f"now must be a datetime, got {now!r}")
+    if now.tzinfo is None:
+        raise ValueError("now must be timezone-aware")
+    return now.astimezone(datetime.timezone.utc).strftime(
+        _UTC_TIMESTAMP_FORMAT
+    )
+
+
+class _TaskReentrantLock:
+    """Shared connection lock the owning task may re-enter."""
+
+    def __init__(self, inner: asyncio.Lock) -> None:
+        self._inner = inner
+        self._task: asyncio.Task | None = None
+        self._depth = 0
+
+    def locked(self) -> bool:
+        return self._inner.locked()
+
+    async def acquire(self) -> bool:
+        task = asyncio.current_task()
+        if self._task is task:
+            self._depth += 1
+            return True
+        await self._inner.acquire()
+        self._task = task
+        self._depth = 1
+        return True
+
+    def release(self) -> None:
+        if self._task is not asyncio.current_task():
+            raise RuntimeError("lock released by a non-owner")
+        self._depth -= 1
+        if self._depth == 0:
+            self._task = None
+            self._inner.release()
+
+    async def __aenter__(self) -> _TaskReentrantLock:
+        await self.acquire()
+        return self
+
+    async def __aexit__(self, *exc: object) -> None:
+        self.release()
+
+
+def _reentrant_connection_lock(database) -> _TaskReentrantLock:
+    """Return the shared connection lock, made task-reentrant.
+
+    ``CompletionEventsDao.append`` also acquires this lock, so
+    check+append in :func:`complete_instance` must re-enter on the
+    same task.  The wrapper is stored in the shared dict so the DAO
+    sees the same object.
+    """
+    loop = asyncio.get_running_loop()
+    key = (id(database), id(loop))
+    lock = _CONNECTION_LOCKS.get(key)
+    if isinstance(lock, _TaskReentrantLock):
+        return lock
+    wrapped = _TaskReentrantLock(
+        lock if lock is not None else asyncio.Lock()
+    )
+    _CONNECTION_LOCKS[key] = wrapped
+    return wrapped
+
+
+def _validate_int_id(value: object, field: str) -> int:
+    """Reject non-int ids (bools included) before any lookup."""
+    if isinstance(value, bool) or not isinstance(value, int):
+        raise ValueError(f"{field} must be an integer, got {value!r}")
+    return value
+
+
+def _validate_event_type(value: object) -> str:
+    """Return a schema-legal event type, or raise naming the field."""
+    if value not in (EVENT_COMPLETED, EVENT_UNCOMPLETED):
+        raise ValueError(
+            f"event_type must be '{EVENT_COMPLETED}' or "
+            f"'{EVENT_UNCOMPLETED}', got {value!r}"
+        )
+    return value
+
+
+def _validate_actor(
+    actor_source: object,
+    actor_user_id: object,
+    actor_child_id: object,
+) -> tuple[str, str | None, int | None]:
+    """Return a schema-legal actor triple, or raise naming the field.
+
+    ``'user'`` requires a non-empty ``actor_user_id`` string and forbids
+    ``actor_child_id``.  ``'panel'`` requires an integer
+    ``actor_child_id`` and forbids ``actor_user_id``.  Any other
+    ``actor_source`` (including ``'service'``) is rejected.
+    """
+    if actor_source not in ("user", "panel"):
+        raise ValueError(
+            f"actor_source must be 'user' or 'panel', got {actor_source!r}"
+        )
+    if actor_source == "user":
+        if not isinstance(actor_user_id, str) or not actor_user_id.strip():
+            raise ValueError("actor_source 'user' requires actor_user_id")
+        if actor_child_id is not None:
+            raise ValueError(
+                "actor_source 'user' must not carry actor_child_id"
+            )
+        return "user", actor_user_id.strip(), None
+    if actor_user_id is not None:
+        raise ValueError("actor_source 'panel' must not carry actor_user_id")
+    if actor_child_id is None:
+        raise ValueError(
+            "actor_source 'panel' requires actor_child_id"
+        )
+    return (
+        "panel",
+        None,
+        _validate_int_id(actor_child_id, "actor_child_id"),
+    )
+
+
+def _validate_was_on_time(
+    event_type: str, was_on_time: object
+) -> bool | None:
+    """Return a real bool or None; completed events require True/False."""
+    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 {was_on_time!r}"
+        )
+    if event_type == EVENT_COMPLETED and was_on_time is None:
+        raise ValueError(
+            "a completed event requires was_on_time (True/False)"
+        )
+    return was_on_time
+
+
+async def append_event(
+    database: NestQuestDatabase,
+    instance_id: int,
+    child_id: int,
+    event_type: str,
+    *,
+    actor_source: str,
+    actor_user_id: str | None = None,
+    actor_child_id: int | None = None,
+    was_on_time: bool | None = None,
+    now: datetime.datetime | None = None,
+) -> CompletionEventRecord:
+    """Append one completion or un-completion event and return the row.
+
+    Validates actor shapes and the completed-event on-time requirement
+    before calling :meth:`CompletionEventsDao.append`.  ``occurred_at``
+    is stamped here (UTC ``+00:00``); pass ``now`` to pin the clock.
+    """
+    instance_id = _validate_int_id(instance_id, "instance_id")
+    child_id = _validate_int_id(child_id, "child_id")
+    event_type = _validate_event_type(event_type)
+    actor_source, actor_user_id, actor_child_id = _validate_actor(
+        actor_source, actor_user_id, actor_child_id
+    )
+    was_on_time = _validate_was_on_time(event_type, was_on_time)
+    occurred_at = _stamp_occurred_at(now)
+    return await CompletionEventsDao(database).append(
+        instance_id,
+        child_id,
+        event_type,
+        actor_source,
+        occurred_at,
+        was_on_time,
+        actor_user_id=actor_user_id,
+        actor_child_id=actor_child_id,
+    )
+
+
+async def instance_state(
+    database: NestQuestDatabase,
+    instance_id: int,
+) -> str:
+    """Return ``open`` or ``done`` from the instance's latest event.
+
+    No events, or a latest event of uncompleted, is ``open``.  A latest
+    event of completed is ``done``.  Unknown ``instance_id`` raises
+    ValueError naming the field.  Existence and the latest event are
+    read inside the connection lock so the answer is one snapshot.
+    """
+    instance_id = _validate_int_id(instance_id, "instance_id")
+    async with _connection_lock(database):
+        instance = await QuestInstancesDao(database).get_by_id(instance_id)
+        if instance is None:
+            raise ValueError(
+                f"instance_id: quest instance {instance_id} does not exist"
+            )
+        latest = await CompletionEventsDao(database).get_latest_for_instance(
+            instance_id
+        )
+    if latest is None or latest.event_type == EVENT_UNCOMPLETED:
+        return "open"
+    if latest.event_type == EVENT_COMPLETED:
+        return "done"
+    raise ValueError(
+        f"event_type must be '{EVENT_COMPLETED}' or "
+        f"'{EVENT_UNCOMPLETED}', got {latest.event_type!r}"
+    )
+
+
+async def complete_instance(
+    database: NestQuestDatabase,
+    instance_id: int,
+    *,
+    actor_source: str,
+    actor_user_id: str | None = None,
+    actor_child_id: int | None = None,
+    now: datetime.datetime | None = None,
+    today: datetime.date | None = None,
+) -> str:
+    """Append a completed event unless the instance is already done.
+
+    Returns the derived state ``done``.  An already-done instance is a
+    no-op: nothing is appended.  Unknown ``instance_id`` raises
+    ValueError naming the field.  Existence, latest-event, and the
+    append run under one connection lock so a concurrent complete
+    cannot double-append.
+
+    ``was_on_time`` is recorded as True — a placeholder until the
+    on-time task computes it from ``now`` / ``today`` against the
+    instance due_date and optional due_time.  ``today`` is accepted so
+    callers can thread the HA-local date through without a signature
+    change; it is unused here.
+    """
+    instance_id = _validate_int_id(instance_id, "instance_id")
+    actor_source, actor_user_id, actor_child_id = _validate_actor(
+        actor_source, actor_user_id, actor_child_id
+    )
+    async with _reentrant_connection_lock(database):
+        instance = await QuestInstancesDao(database).get_by_id(instance_id)
+        if instance is None:
+            raise ValueError(
+                f"instance_id: quest instance {instance_id} does not exist"
+            )
+        latest = await CompletionEventsDao(database).get_latest_for_instance(
+            instance_id
+        )
+        if latest is not None and latest.event_type == EVENT_COMPLETED:
+            return "done"
+        if latest is not None and latest.event_type != EVENT_UNCOMPLETED:
+            raise ValueError(
+                f"event_type must be '{EVENT_COMPLETED}' or "
+                f"'{EVENT_UNCOMPLETED}', got {latest.event_type!r}"
+            )
+        await append_event(
+            database,
+            instance.id,
+            instance.child_id,
+            EVENT_COMPLETED,
+            actor_source=actor_source,
+            actor_user_id=actor_user_id,
+            actor_child_id=actor_child_id,
+            was_on_time=True,
+            now=now,
+        )
+    return "done"

tests/test_completion.py

@@ -0,0 +1,686 @@
+"""Tests for completion.py: the Feature 08 append-only business layer."""
+from __future__ import annotations
+
+import asyncio
+import datetime
+import inspect
+import re
+
+import pytest
+
+from custom_components.nestquest.completion import (
+    append_event,
+    complete_instance,
+    instance_state,
+)
+from custom_components.nestquest.dao_children import ChildrenDao
+from custom_components.nestquest.dao_instances import (
+    EVENT_COMPLETED,
+    EVENT_UNCOMPLETED,
+    CompletionEventRecord,
+    CompletionEventsDao,
+    QuestInstancesDao,
+)
+from custom_components.nestquest.dao_rules import (
+    QuestDefinitionsDao,
+    ScheduleRulesDao,
+)
+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():
+    from unittest.mock import AsyncMock, MagicMock
+
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+def _today_iso() -> str:
+    return datetime.date.today().isoformat()
+
+
+def _now_stamp() -> str:
+    return f"{_today_iso()}T08:00:00+00:00"
+
+
+def _future_day(offset: int) -> str:
+    return (
+        datetime.date.today() + datetime.timedelta(days=offset)
+    ).isoformat()
+
+
+D1 = _future_day(1)
+
+
+async def _prepare(path) -> tuple:
+    database = NestQuestDatabase(_make_hass_mock())
+    await database.open(path)
+    await apply_migrations(database)
+    children = ChildrenDao(database)
+    rules = ScheduleRulesDao(database)
+    definitions = QuestDefinitionsDao(database)
+    instances = QuestInstancesDao(database)
+    child = await children.create("Ada", _now_stamp())
+    rule = await rules.create("daily", D1)
+    definition = await definitions.create(
+        "Brush teeth", rule.id, _now_stamp(), assignee_child_ids=[child.id]
+    )
+    instance = await instances.upsert(
+        definition.id, child.id, D1, _now_stamp(), window="morning"
+    )
+    return database, child, instance
+
+
+def _with_db(tmp_path, name):
+    def _run_test(body):
+        async def _main():
+            prepared = await _prepare(tmp_path / name)
+            try:
+                return await body(*prepared)
+            finally:
+                await prepared[0].close()
+
+        return _run(_main())
+
+    return _run_test
+
+
+# ---------------------------------------------------------------------------
+# inverted actor shapes: field-naming ValueErrors, no write
+# ---------------------------------------------------------------------------
+
+
+def test_user_without_actor_user_id_rejected() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                was_on_time=True,
+            )
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id=None,
+                was_on_time=True,
+            )
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id="",
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_user_must_not_carry_actor_child_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_child_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id="user-1",
+                actor_child_id=1,
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_panel_must_not_carry_actor_user_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_user_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="panel",
+                actor_user_id="user-1",
+                actor_child_id=1,
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_panel_requires_actor_child_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_child_id"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="panel",
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_actor_source_service_rejected() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="actor_source"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="service",
+                actor_user_id="user-1",
+                was_on_time=True,
+            )
+
+    _run(_body())
+
+
+def test_completed_requires_was_on_time() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="was_on_time"):
+            await append_event(
+                object(),
+                1,
+                1,
+                EVENT_COMPLETED,
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+
+    _run(_body())
+
+
+# ---------------------------------------------------------------------------
+# happy paths
+# ---------------------------------------------------------------------------
+
+
+def test_user_append_round_trips(tmp_path) -> None:
+    async def _body(database, child, instance):
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+        )
+        assert isinstance(event, CompletionEventRecord)
+        assert event.event_type == EVENT_COMPLETED
+        assert event.actor_source == "user"
+        assert event.actor_user_id == "user-1"
+        assert event.actor_child_id is None
+        assert event.was_on_time is True
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00", event.occurred_at
+        )
+        return event
+
+    _with_db(tmp_path, "user-append.db")(_body)
+
+
+def test_panel_append_round_trips(tmp_path) -> None:
+    async def _body(database, child, instance):
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="panel",
+            actor_child_id=child.id,
+            was_on_time=False,
+        )
+        assert event.actor_source == "panel"
+        assert event.actor_user_id is None
+        assert event.actor_child_id == child.id
+        assert event.was_on_time is False
+        assert re.fullmatch(
+            r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\+00:00", event.occurred_at
+        )
+        return event
+
+    _with_db(tmp_path, "panel-append.db")(_body)
+
+
+def test_uncompleted_allows_null_was_on_time(tmp_path) -> None:
+    async def _body(database, child, instance):
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=None,
+        )
+        assert event.event_type == EVENT_UNCOMPLETED
+        assert event.was_on_time is None
+        return event
+
+    _with_db(tmp_path, "uncompleted-null.db")(_body)
+
+
+def test_now_pins_occurred_at_utc(tmp_path) -> None:
+    async def _body(database, child, instance):
+        pinned = datetime.datetime(
+            2026, 9, 18, 8, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        event = await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+            now=pinned,
+        )
+        assert event.occurred_at == "2026-09-18T08:00:00+00:00"
+        return event
+
+    _with_db(tmp_path, "pinned-now.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# append-only: the module never exposes update/delete
+# ---------------------------------------------------------------------------
+
+
+def test_module_never_exposes_update_or_delete() -> None:
+    import custom_components.nestquest.completion as completion
+
+    public = {
+        name
+        for name, _ in inspect.getmembers(completion, inspect.isfunction)
+        if not name.startswith("_")
+    }
+    forbidden = {
+        name
+        for name in public
+        if name.startswith(
+            (
+                "update",
+                "delete",
+                "set",
+                "clear",
+                "remove",
+                "modify",
+                "edit",
+                "upsert",
+            )
+        )
+    }
+    assert forbidden == set(), (
+        f"completion.py exposed mutation methods: {forbidden}"
+    )
+    assert "append_event" in public
+    assert "instance_state" in public
+    assert "complete_instance" in public
+    assert "uncomplete_instance" not in public
+    source = inspect.getsource(completion)
+    assert "UPDATE" not in source
+    assert "DELETE" not in source
+    assert "INSERT" not in source
+
+
+# ---------------------------------------------------------------------------
+# instance_state: derived from the latest event
+# ---------------------------------------------------------------------------
+
+
+def test_instance_state_open_when_no_events(tmp_path) -> None:
+    async def _body(database, child, instance):
+        assert await instance_state(database, instance.id) == "open"
+
+    _with_db(tmp_path, "state-none.db")(_body)
+
+
+def test_instance_state_open_when_latest_uncompleted(tmp_path) -> None:
+    async def _body(database, child, instance):
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        assert await instance_state(database, instance.id) == "open"
+
+    _with_db(tmp_path, "state-uncompleted.db")(_body)
+
+
+def test_instance_state_unknown_instance_id_raises(tmp_path) -> None:
+    async def _body(database, child, instance):
+        with pytest.raises(ValueError, match="instance_id"):
+            await instance_state(database, instance.id + 999)
+
+    _with_db(tmp_path, "state-unknown.db")(_body)
+
+
+def test_instance_state_rejects_non_integer_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="instance_id"):
+            await instance_state(object(), "1")
+        with pytest.raises(ValueError, match="instance_id"):
+            await instance_state(object(), True)
+
+    _run(_body())
+
+
+def test_instance_state_three_events_returns_done_and_preserves_rows(
+    tmp_path,
+) -> None:
+    async def _body(database, child, instance):
+        base = datetime.datetime(
+            2026, 9, 18, 8, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+            now=base,
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            now=base + datetime.timedelta(seconds=1),
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=False,
+            now=base + datetime.timedelta(seconds=2),
+        )
+        events = CompletionEventsDao(database)
+        before = await events.list_by_instance(instance.id)
+        assert [e.event_type for e in before] == [
+            EVENT_COMPLETED,
+            EVENT_UNCOMPLETED,
+            EVENT_COMPLETED,
+        ]
+        assert await instance_state(database, instance.id) == "done"
+        after = await events.list_by_instance(instance.id)
+        assert after == before
+        assert len(after) == 3
+
+    _with_db(tmp_path, "state-three.db")(_body)
+
+
+# ---------------------------------------------------------------------------
+# complete_instance: append completed, no-op if already done
+# ---------------------------------------------------------------------------
+
+
+def test_complete_instance_appends_and_returns_done(tmp_path) -> None:
+    async def _body(database, child, instance):
+        events = CompletionEventsDao(database)
+        assert await events.list_by_instance(instance.id) == []
+        state = await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        assert state == "done"
+        assert await instance_state(database, instance.id) == "done"
+        rows = await events.list_by_instance(instance.id)
+        assert len(rows) == 1
+        assert rows[0].event_type == EVENT_COMPLETED
+        assert rows[0].actor_source == "user"
+        assert rows[0].actor_user_id == "user-1"
+        assert rows[0].was_on_time is True
+
+    _with_db(tmp_path, "complete-open.db")(_body)
+
+
+def test_complete_instance_already_done_is_noop(tmp_path) -> None:
+    async def _body(database, child, instance):
+        first = await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        events = CompletionEventsDao(database)
+        before = await events.list_by_instance(instance.id)
+        assert first == "done"
+        assert len(before) == 1
+        second = await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-2",
+        )
+        after = await events.list_by_instance(instance.id)
+        assert second == "done"
+        assert after == before
+        assert after[0].actor_user_id == "user-1"
+
+    _with_db(tmp_path, "complete-noop.db")(_body)
+
+
+def test_complete_instance_unknown_instance_id_raises(tmp_path) -> None:
+    async def _body(database, child, instance):
+        with pytest.raises(ValueError, match="instance_id"):
+            await complete_instance(
+                database,
+                instance.id + 999,
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+
+    _with_db(tmp_path, "complete-unknown.db")(_body)
+
+
+def test_complete_instance_rejects_non_integer_id() -> None:
+    async def _body():
+        with pytest.raises(ValueError, match="instance_id"):
+            await complete_instance(
+                object(),
+                "1",
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+        with pytest.raises(ValueError, match="instance_id"):
+            await complete_instance(
+                object(),
+                True,
+                actor_source="user",
+                actor_user_id="user-1",
+            )
+
+    _run(_body())
+
+
+def test_complete_instance_after_uncompleted_appends(tmp_path) -> None:
+    async def _body(database, child, instance):
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_COMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+            was_on_time=True,
+        )
+        await append_event(
+            database,
+            instance.id,
+            child.id,
+            EVENT_UNCOMPLETED,
+            actor_source="user",
+            actor_user_id="user-1",
+        )
+        events = CompletionEventsDao(database)
+        before = await events.list_by_instance(instance.id)
+        assert await instance_state(database, instance.id) == "open"
+        state = await complete_instance(
+            database,
+            instance.id,
+            actor_source="panel",
+            actor_child_id=child.id,
+        )
+        assert state == "done"
+        after = await events.list_by_instance(instance.id)
+        assert len(after) == len(before) + 1
+        assert after[-1].event_type == EVENT_COMPLETED
+        assert after[-1].actor_source == "panel"
+        assert after[-1].was_on_time is True
+
+    _with_db(tmp_path, "complete-reopen.db")(_body)
+
+
+def test_complete_instance_now_pins_occurred_at(tmp_path) -> None:
+    async def _body(database, child, instance):
+        pinned = datetime.datetime(
+            2026, 9, 18, 8, 0, 0, tzinfo=datetime.timezone.utc
+        )
+        await complete_instance(
+            database,
+            instance.id,
+            actor_source="user",
+            actor_user_id="user-1",
+            now=pinned,
+        )
+        rows = await CompletionEventsDao(database).list_by_instance(
+            instance.id
+        )
+        assert rows[0].occurred_at == "2026-09-18T08:00:00+00:00"
+
+    _with_db(tmp_path, "complete-pinned.db")(_body)
+
+
+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 INSERT of whichever ``complete_instance`` starts first
+    — proving that caller holds the connection lock 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_complete_instance_concurrent_second_call_is_noop(tmp_path) -> None:
+    """Two racing completes cannot both append.
+
+    Deterministic gate: task A is paused at its INSERT (it holds the
+    connection lock); task B is then started and must queue.  A
+    finishes, B sees done and appends nothing.
+    """
+    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 / "complete-race.db")
+        try:
+            await apply_migrations(database)
+            children = ChildrenDao(database)
+            rules = ScheduleRulesDao(database)
+            definitions = QuestDefinitionsDao(database)
+            instances = QuestInstancesDao(database)
+            child = await children.create("Ada", _now_stamp())
+            rule = await rules.create("daily", D1)
+            definition = await definitions.create(
+                "Brush teeth",
+                rule.id,
+                _now_stamp(),
+                assignee_child_ids=[child.id],
+            )
+            instance = await instances.upsert(
+                definition.id,
+                child.id,
+                D1,
+                _now_stamp(),
+                window="morning",
+            )
+
+            armed["active"] = True
+            task_a = asyncio.ensure_future(
+                complete_instance(
+                    database,
+                    instance.id,
+                    actor_source="user",
+                    actor_user_id="user-1",
+                )
+            )
+            await a_started.wait()
+            armed["active"] = False
+            task_b = asyncio.ensure_future(
+                complete_instance(
+                    database,
+                    instance.id,
+                    actor_source="user",
+                    actor_user_id="user-2",
+                )
+            )
+            await asyncio.sleep(0)
+            gate_open.set()
+            state_a, state_b = await asyncio.gather(task_a, task_b)
+            assert state_a == "done"
+            assert state_b == "done"
+            rows = await CompletionEventsDao(database).list_by_instance(
+                instance.id
+            )
+            assert len(rows) == 1
+            assert rows[0].actor_user_id == "user-1"
+        finally:
+            await database.close()
+
+    _run(_main())