diff --git a/custom_components/nestquest/dao_instances.py b/custom_components/nestquest/dao_instances.py
index 43c8997..2e9aaa3 100644
--- a/custom_components/nestquest/dao_instances.py
+++ b/custom_components/nestquest/dao_instances.py
@@ -15,13 +15,13 @@ event.  A package-wide guard test (bottom of the test module) keeps
 mutation SQL for this table out of every other module.
 
 Upsert semantics: ``QuestInstancesDao.upsert`` is idempotent on
-(definition_id, due_date) — re-running materialization for a date
-whose instance already exists updates nothing that matters
-(child_id/due_time are refreshed only in the degenerate re-generation
-case where they changed) and never duplicates a row.  The schema's
-UNIQUE(definition_id, due_date) is the last line of defense.  The
-upsert refuses a child that is not an assignee of the definition
-(D-008 multi-assignee model).
+(definition_id, child_id, due_date, window) — re-running
+materialization for a tuple whose instance already exists refreshes
+only the generation-time snapshot columns and never duplicates a row.
+The schema's UNIQUE(definition_id, child_id, due_date, window) is the
+last line of defense.  The upsert refuses a child that is not an
+assignee of the definition and a window outside ``const.QUEST_WINDOWS``
+(D-008 multi-assignee, multi-window model).
 
 ``delete_future_uncompleted`` implements the Feature 06/07 rule that
 reassignment and schedule edits regenerate future instances: it
@@ -37,7 +37,7 @@ import datetime
 from dataclasses import dataclass
 
 from .dao_children import _connection_lock
-from .dao_rules import _validate_date
+from .dao_rules import _validate_date, _validate_window
 from .db import NestQuestDatabase
 
 #: Timestamp policy for completion_events.occurred_at: strict UTC
@@ -90,6 +90,7 @@ class QuestInstanceRecord:
     id: int
     definition_id: int
     child_id: int
+    window: str
     due_date: str
     due_time: str | None
     generated_at: str
@@ -110,7 +111,7 @@ class CompletionEventRecord:
 
 
 _INSTANCE_COLUMNS = (
-    "id, definition_id, child_id, due_date, due_time, generated_at"
+    "id, definition_id, child_id, window, due_date, due_time, generated_at"
 )
 _EVENT_COLUMNS = (
     "id, instance_id, child_id, event_type, actor_source, actor_user_id, "
@@ -128,9 +129,10 @@ def _instance_from_row(row: tuple) -> QuestInstanceRecord:
         id=row[0],
         definition_id=row[1],
         child_id=row[2],
-        due_date=row[3],
-        due_time=row[4],
-        generated_at=row[5],
+        window=row[3],
+        due_date=row[4],
+        due_time=row[5],
+        generated_at=row[6],
     )
 
 
@@ -160,31 +162,40 @@ class QuestInstancesDao:
         due_date: str,
         generated_at: str,
         *,
+        window: str,
         due_time: str | None = None,
     ) -> QuestInstanceRecord:
-        """Create the dated instance for (definition_id, due_date) idempotently.
+        """Create the dated instance for the widened key idempotently.
 
-        Running materialization twice for the same definition/date
-        yields ONE row: the ON CONFLICT path refreshes the
-        generation-time snapshot columns (child_id, due_time) and
-        regenerates the stamp, but can never create a duplicate — the
-        schema UNIQUE(definition_id, due_date) backs this up.
+        The instance key is (definition_id, child_id, due_date, window)
+        (D-008): running materialization twice for the same tuple
+        yields ONE row — the ON CONFLICT path refreshes the
+        generation-time snapshot columns (due_time) and regenerates the
+        stamp, but can never create a duplicate.  The schema's
+        UNIQUE(definition_id, child_id, due_date, window) backs this
+        up, so the twice-daily case (two windows, one child, one date)
+        yields two rows and the shared case (three assignees, one
+        window) yields three.
 
         Guardrails enforced here:
 
         - ``due_date`` must be TODAY or later: instances are never
           generated in the past (Feature 07).  The caller's
           materialization horizon is future-dated by definition.
+        - ``window`` must be one of ``const.QUEST_WINDOWS``.
         - An existing instance WITH a completion event is IMMUTABLE:
           the conflict path must not rewrite its snapshot columns
           (Feature 07/08).  The upsert refuses with ValueError in that
           case — silently no-oping would mask a materialization bug.
+        - The child must be an assignee of the definition (D-008
+          multi-assignee model).
 
         The definition must exist (validated atomically under the
         connection lock so a concurrent definition deactivation or
         child deletion cannot slip between check and insert).
         """
         _validate_date(due_date, "due_date")
+        _validate_window(window)
         # Fail fast on obviously-past dates BEFORE queuing on the
         # connection lock (a nice early error); the AUTHORITATIVE
         # check runs after the lock is acquired, because a caller can
@@ -227,51 +238,56 @@ class QuestInstancesDao:
                         f"assigned to child {child_id}; instance must "
                         "carry an assigned child"
                     )
-                child = await self._database.fetch_one(
-                    "SELECT 1 FROM children WHERE id = ?", (child_id,)
-                )
-                if child is None:
-                    raise ValueError(f"child {child_id} does not exist")
                 existing = await self._database.fetch_one(
                     "SELECT 1 FROM completion_events "
                     "WHERE instance_id = (SELECT id FROM quest_instances "
-                    "WHERE definition_id = ? AND due_date = ?)",
-                    (definition_id, due_date),
+                    "WHERE definition_id = ? AND child_id = ? "
+                    "AND due_date = ? AND window = ?)",
+                    (definition_id, child_id, due_date, window),
                 )
                 if existing is not None:
                     raise ValueError(
                         f"instance for (definition {definition_id}, "
-                        f"due_date {due_date!r}) already has a completion "
+                        f"child {child_id}, due_date {due_date!r}, "
+                        f"window {window!r}) already has a completion "
                         "event and is immutable; it cannot be regenerated"
                     )
                 await self._database.execute(
                     "INSERT INTO quest_instances (definition_id, child_id, "
-                    "due_date, due_time, generated_at) "
-                    "VALUES (?, ?, ?, ?, ?) "
-                    "ON CONFLICT (definition_id, due_date) DO UPDATE SET "
-                    "child_id = excluded.child_id, "
+                    "window, due_date, due_time, generated_at) "
+                    "VALUES (?, ?, ?, ?, ?, ?) "
+                    "ON CONFLICT (definition_id, child_id, due_date, "
+                    "window) DO UPDATE SET "
                     "due_time = excluded.due_time, "
                     "generated_at = excluded.generated_at",
                     (
                         definition_id,
                         child_id,
+                        window,
                         due_date,
                         due_time,
                         generated_at,
                     ),
                 )
-                instance = await self.get(definition_id, due_date)
+                instance = await self.get(
+                    definition_id, child_id, due_date, window
+                )
         assert instance is not None
         return instance
 
     async def get(
-        self, definition_id: int, due_date: str
+        self,
+        definition_id: int,
+        child_id: int,
+        due_date: str,
+        window: str,
     ) -> QuestInstanceRecord | None:
-        """Return the instance for (definition_id, due_date), or None."""
+        """Return the instance for the widened key tuple, or None."""
         row = await self._database.fetch_one(
             f"SELECT {_INSTANCE_COLUMNS} FROM quest_instances "
-            "WHERE definition_id = ? AND due_date = ?",
-            (definition_id, due_date),
+            "WHERE definition_id = ? AND child_id = ? AND due_date = ? "
+            "AND window = ?",
+            (definition_id, child_id, due_date, window),
         )
         return _instance_from_row(row) if row is not None else None
 
diff --git a/custom_components/nestquest/migrations.py b/custom_components/nestquest/migrations.py
index 29d5d20..9901505 100644
--- a/custom_components/nestquest/migrations.py
+++ b/custom_components/nestquest/migrations.py
@@ -24,9 +24,12 @@ names before anything else assumes the new model.  Migration 3 brings
 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.  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.
+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.
 
 Migration entries are either an ordered sequence of SQL statements or
 an async callable taking ``(database, target_version)``.  Callables
@@ -71,6 +74,7 @@ from typing import Union
 from .db import NestQuestDatabase
 from .schema import (
     QUEST_DEFINITIONS_TABLE_SQL,
+    QUEST_INSTANCES_TABLE_SQL,
     SCHEMA_V1_STATEMENTS,
     SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
     SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
@@ -147,6 +151,14 @@ async def _table_has_column(
     return any(row[1] == column for row in rows)
 
 
+#: Window assigned to instances migrated from the pre-window instance
+#: shape (D-008).  Pre-window rows were generated for whole-day
+#: obligations; the morning window is the documented default landing
+#: spot, and future dates are regenerated with real window declarations
+#: by materialization (delete_future_uncompleted + re-run).
+LEGACY_MIGRATED_WINDOW = "morning"
+
+
 async def _rename_legacy_task_tables(
     database: NestQuestDatabase, target_version: int
 ) -> None:
@@ -250,6 +262,70 @@ async def _add_definition_assignees(
         await database.execute("PRAGMA foreign_keys = ON")
 
 
+async def _widen_instance_key(
+    database: NestQuestDatabase, target_version: int
+) -> None:
+    """Bring ``quest_instances`` to the D-008 instance key.
+
+    Databases whose instance table still carries the pre-window shape
+    (no ``window`` column, UNIQUE on (definition_id, due_date)) are
+    rebuilt via SQLite's canonical procedure into the widened shape —
+    UNIQUE (definition_id, child_id, due_date, window) — with every
+    migrated row landing in :data:`LEGACY_MIGRATED_WINDOW` so existing
+    ids (and the completion events referencing them) survive intact.
+    The old rows were unique per (definition, due_date), so pinning one
+    window cannot collide with the widened key.  Fresh databases whose
+    table already matches the canonical shape only stamp.
+
+    The rebuild runs with ``foreign_keys = OFF`` (dropping the old
+    table under enforcement would fire the implicit DELETE and violate
+    the ``completion_events`` foreign key) and verifies
+    ``PRAGMA foreign_key_check`` inside the transaction before the
+    stamp, so a corrupt copy aborts the whole migration.
+    """
+    legacy_shape = not await _table_has_column(
+        database, "quest_instances", "window"
+    )
+    await database.execute("PRAGMA foreign_keys = OFF")
+    try:
+        async with database.transaction():
+            if legacy_shape:
+                await database.execute(
+                    "CREATE TABLE quest_instances_rebuilt "
+                    f"{QUEST_INSTANCES_TABLE_SQL}"
+                )
+                await database.execute(
+                    "INSERT INTO quest_instances_rebuilt "
+                    "(id, definition_id, child_id, window, due_date, "
+                    "due_time, generated_at) "
+                    "SELECT id, definition_id, child_id, ?, due_date, "
+                    "due_time, generated_at "
+                    "FROM quest_instances ORDER BY id",
+                    (LEGACY_MIGRATED_WINDOW,),
+                )
+                await database.execute("DROP TABLE quest_instances")
+                await database.execute(
+                    "ALTER TABLE quest_instances_rebuilt "
+                    "RENAME TO quest_instances"
+                )
+                violations = await database.fetch_all(
+                    "PRAGMA foreign_key_check"
+                )
+                if violations:
+                    raise RuntimeError(
+                        "instance rebuild produced foreign-key "
+                        f"violations: {violations!r}"
+                    )
+                LOGGER.info(
+                    "Widened quest_instances key; migrated rows pinned "
+                    "to the %r window",
+                    LEGACY_MIGRATED_WINDOW,
+                )
+            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.)
@@ -261,6 +337,9 @@ MIGRATIONS: Sequence[MigrationStep] = [
     # CREATE IF NOT EXISTS suffices for both fresh databases (which
     # already made it in migration 1) and pre-window databases.
     list(SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL),
+    # Migration 5 (D-008): rebuild quest_instances onto the widened
+    # (definition_id, child_id, due_date, window) key.
+    _widen_instance_key,
 ]
 
 
diff --git a/custom_components/nestquest/schema.py b/custom_components/nestquest/schema.py
index 7a73340..46447e6 100644
--- a/custom_components/nestquest/schema.py
+++ b/custom_components/nestquest/schema.py
@@ -109,25 +109,31 @@ status column.  An instance's current state derives from the latest row
 in ``completion_events`` (Feature 08), so materialization (Feature 07)
 can insert instances without knowing anything about completion and the
 append-only log stays the single source of truth.  ``definition_id``,
-``due_date`` and ``child_id`` snapshot the materialization-time facts
-("why this instance exists"): reassignment and rule edits change future
-instances only and never rewrite these rows.  ``due_time`` snapshots
-the definition's optional due time at generation for the same reason —
-the instance keeps its generation-time value even if the definition is
-edited later.  ``generated_at`` is a caller-set UTC ISO-8601 timestamp
-per the module timestamp policy (no DB default, so a generation batch
-stamps one coherent time).
+``child_id``, ``due_date`` and ``window`` snapshot the
+materialization-time facts ("why this instance exists"): assignment
+edits and rule edits change future instances only and never rewrite
+these rows.  ``window`` records which declared day window produced the
+instance (D-008), CHECK-constrained to the same three spellings
+``const`` owns.  ``due_time`` snapshots the window's optional due time
+at generation for the same reason — the instance keeps its
+generation-time value even if the definition is edited later.
+``generated_at`` is a caller-set UTC ISO-8601 timestamp per the module
+timestamp policy (no DB default, so a generation batch stamps one
+coherent time).
 
-Uniqueness: the (definition_id, due_date) pair is declared as a
-table-level ``UNIQUE`` constraint rather than a separate ``CREATE
-UNIQUE INDEX`` statement.  SQLite materializes it as an implicit unique
-index (``sqlite_autoindex_quest_instances_1``) — the required "unique
-index prevents duplicate instances" — and the constraint form cannot be
-silently dropped the way a standalone index can, so the duplicate door
-stays shut.  Idempotent materialization keys on exactly this pair.  No
-secondary indexes beyond it are declared yet: history and board queries
-(Features 08/10/13) should justify their indexes as a later migration
-once their query shapes exist.
+Uniqueness: the (definition_id, child_id, due_date, window) tuple is
+declared as a table-level ``UNIQUE`` constraint rather than a separate
+``CREATE UNIQUE INDEX`` statement.  SQLite materializes it as an
+implicit unique index (``sqlite_autoindex_quest_instances_1``) — the
+required "unique index prevents duplicate instances" — and the
+constraint form cannot be silently dropped the way a standalone index
+can, so the duplicate door stays shut.  Idempotent materialization keys
+on exactly this tuple: the twice-daily case (one definition, two
+windows, one child, one date) yields two rows and the shared case (one
+definition, three assignees, one window) yields three.  No secondary
+indexes beyond it are declared yet: history and board queries (Features
+08/10/13) should justify their indexes as a later migration once their
+query shapes exist.
 
 No ``ON DELETE`` clauses are declared on any foreign key in this
 module: children and definitions are deactivated, never deleted, so a
@@ -316,17 +322,25 @@ SCHEMA_V1_PRESENCE_OVERRIDES_DDL: list[str] = [
     """,
 ]
 
+#: Column body of ``quest_instances``, shared between the v1 DDL
+#: below and the instance-key rebuild migration, which must recreate
+#: the table in exactly this shape (D-008: the instance key is
+#: (definition_id, child_id, due_date, window) — one instance per
+#: assigned child per declared window per firing date).
+QUEST_INSTANCES_TABLE_SQL = """(
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+    child_id INTEGER NOT NULL REFERENCES children(id),
+    window TEXT NOT NULL CHECK (window IN ('morning', 'afternoon', 'evening')),
+    due_date TEXT NOT NULL,
+    due_time TEXT,
+    generated_at TEXT NOT NULL,
+    UNIQUE (definition_id, child_id, due_date, window)
+)"""
+
 SCHEMA_V1_QUEST_INSTANCES_DDL: list[str] = [
-    """
-    CREATE TABLE IF NOT EXISTS quest_instances (
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
-        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
-        child_id INTEGER NOT NULL REFERENCES children(id),
-        due_date TEXT NOT NULL,
-        due_time TEXT,
-        generated_at TEXT NOT NULL,
-        UNIQUE (definition_id, due_date)
-    )
+    f"""
+    CREATE TABLE IF NOT EXISTS quest_instances {QUEST_INSTANCES_TABLE_SQL}
     """,
 ]
 
diff --git a/tests/test_dao_instances.py b/tests/test_dao_instances.py
index 628752e..fe79027 100644
--- a/tests/test_dao_instances.py
+++ b/tests/test_dao_instances.py
@@ -119,12 +119,10 @@ def test_upsert_creates_then_is_idempotent(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         first = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         assert isinstance(first, QuestInstanceRecord)
         second = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         assert second.id == first.id
         rows = await database.fetch_one(
             "SELECT COUNT(*) FROM quest_instances"
@@ -138,27 +136,116 @@ def test_upsert_creates_then_is_idempotent(tmp_path) -> None:
 def test_upsert_different_dates_create_distinct_rows(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        a = await instances.upsert(definition.id, child.id, D1, _now_stamp())
-        b = await instances.upsert(definition.id, child.id, D2, _now_stamp())
+        a = await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
+        b = await instances.upsert(definition.id, child.id, D2, _now_stamp(), window="morning")
         assert a.id != b.id
         return (a, b)
 
     _with_db(tmp_path, "upsert-multi-date.db")(_body)
 
 
+def test_upsert_twice_daily_fanout_two_windows_two_rows(tmp_path) -> None:
+    """D-008 twice-daily case: one definition, one child, one date,
+    two windows — two rows, each idempotent on its own tuple."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        morning = await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="morning"
+        )
+        evening = await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="evening"
+        )
+        assert morning.id != evening.id
+        rows = await database.fetch_all(
+            "SELECT window, child_id, due_date FROM quest_instances "
+            "WHERE definition_id = ? AND due_date = ?",
+            (definition.id, D1),
+        )
+        assert sorted(rows) == [
+            ("evening", child.id, D1),
+            ("morning", child.id, D1),
+        ]
+        # Re-running both is still idempotent.
+        await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="morning"
+        )
+        await instances.upsert(
+            definition.id, child.id, D1, _now_stamp(), window="evening"
+        )
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        assert count == (2,)
+        return (morning, evening)
+
+    _with_db(tmp_path, "upsert-twice-daily.db")(_body)
+
+
+def test_upsert_shared_definition_three_assignees_three_rows(
+    tmp_path,
+) -> None:
+    """D-008 shared case: one definition, three assignees, one window
+    — three rows, one per child."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        second = await children.create("Bo", _now_stamp())
+        third = await children.create("Cleo", _now_stamp())
+        await definitions.add_assignee(definition.id, second.id)
+        await definitions.add_assignee(definition.id, third.id)
+        created = [
+            await instances.upsert(
+                definition.id, assignee.id, D1, _now_stamp(),
+                window="morning",
+            )
+            for assignee in (child, second, third)
+        ]
+        assert len({row.id for row in created}) == 3
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances "
+            "WHERE definition_id = ? AND due_date = ? AND window = ?",
+            (definition.id, D1, "morning"),
+        )
+        assert count == (3,)
+        # Re-running the fan-out changes nothing.
+        for assignee in (child, second, third):
+            await instances.upsert(
+                definition.id, assignee.id, D1, _now_stamp(),
+                window="morning",
+            )
+        count = await database.fetch_one(
+            "SELECT COUNT(*) FROM quest_instances"
+        )
+        assert count == (3,)
+        return created
+
+    _with_db(tmp_path, "upsert-shared-fanout.db")(_body)
+
+
+def test_upsert_window_outside_const_list_raises(tmp_path) -> None:
+    """The DAO rejects a window spelling the schema CHECK would reject
+    anyway, with the typed ValueError instead of raw IntegrityError."""
+    async def _body(database, children, rules, definitions, instances,
+                    events, child, definition):
+        with pytest.raises(ValueError, match="window must be one of"):
+            await instances.upsert(
+                definition.id, child.id, D1, _now_stamp(), window="noon"
+            )
+        return None
+
+    _with_db(tmp_path, "upsert-bad-window.db")(_body)
+
+
 def test_upsert_refreshes_snapshot_columns_on_conflict(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         first = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp(), due_time="08:00"
-        )
+            definition.id, child.id, D1, _now_stamp(), due_time="08:00", window="morning")
         second = await instances.upsert(
             definition.id,
             child.id,
             D1,
             f"{D1}T13:00:00+00:00",
-            due_time="08:30",
-        )
+            due_time="08:30", window="morning")
         assert second.id == first.id
         assert second.due_time == "08:30"
         assert second.generated_at == f"{D1}T13:00:00+00:00"
@@ -172,7 +259,7 @@ def test_upsert_rejects_wrong_child_for_definition(tmp_path) -> None:
                     events, child, definition):
         other = await children.create("Bo", _now_stamp())
         with pytest.raises(ValueError, match="assigned to"):
-            await instances.upsert(definition.id, other.id, D1, _now_stamp())
+            await instances.upsert(definition.id, other.id, D1, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-wrong-child.db")(_body)
@@ -182,7 +269,7 @@ def test_upsert_rejects_unknown_definition_and_child(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         with pytest.raises(ValueError, match="definition 999"):
-            await instances.upsert(999, child.id, D1, _now_stamp())
+            await instances.upsert(999, child.id, D1, _now_stamp(), window="morning")
         # Unknown child: the definition-assignee check fires first and
         # is the one that matters (an instance must carry an assigned
         # child).  Under D-008 the assignee check is a membership test
@@ -193,7 +280,7 @@ def test_upsert_rejects_unknown_definition_and_child(tmp_path) -> None:
             "X", rule.id, _now_stamp(), assignee_child_ids=[child.id]
         )
         with pytest.raises(ValueError, match="not assigned to child 999"):
-            await instances.upsert(other_definition.id, 999, D1, _now_stamp())
+            await instances.upsert(other_definition.id, 999, D1, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-unknown.db")(_body)
@@ -204,7 +291,7 @@ def test_upsert_malformed_due_date_raises(tmp_path) -> None:
                     events, child, definition):
         for bad in ("not-a-date", "2026-9-15", ""):
             with pytest.raises(ValueError, match="YYYY-MM-DD"):
-                await instances.upsert(definition.id, child.id, bad, _now_stamp())
+                await instances.upsert(definition.id, child.id, bad, _now_stamp(), window="morning")
         return None
 
     _with_db(tmp_path, "upsert-bad-date.db")(_body)
@@ -216,11 +303,11 @@ def test_upsert_rejects_past_due_dates(tmp_path) -> None:
                     events, child, definition):
         yesterday = _future_day(-1)
         with pytest.raises(ValueError, match="never generated in the past"):
-            await instances.upsert(definition.id, child.id, yesterday, _now_stamp())
+            await instances.upsert(definition.id, child.id, yesterday, _now_stamp(), window="morning")
         # Today itself is allowed (the rollover job generates today's
         # instances before the day is over).
         today = datetime.date.today().isoformat()
-        record = await instances.upsert(definition.id, child.id, today, _now_stamp())
+        record = await instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
         assert record.due_date == today
         return record
 
@@ -296,7 +383,7 @@ def test_upsert_no_past_check_uses_execution_date_not_call_date(
             await asyncio.wait_for(holder_inside.wait(), timeout=5)
 
             task = asyncio.ensure_future(
-                instances.upsert(definition.id, child.id, today, _now_stamp())
+                instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
             )
             # A is now queued on the connection lock (it cannot pass
             # its in-lock check while B holds the lock).  Shift the
@@ -349,17 +436,15 @@ def test_upsert_refuses_immutable_completed_instance(tmp_path) -> None:
                     events, child, definition):
         generated_at = _now_stamp()
         instance = await instances.upsert(
-            definition.id, child.id, D1, generated_at, due_time="08:00"
-        )
+            definition.id, child.id, D1, generated_at, due_time="08:00", window="morning")
         await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
         )
         with pytest.raises(ValueError, match="immutable"):
             await instances.upsert(
-                definition.id, child.id, D1, _now_stamp(), due_time="09:00"
-            )
-        fetched = await instances.get(definition.id, D1)
+                definition.id, child.id, D1, _now_stamp(), due_time="09:00", window="morning")
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
         assert fetched.due_time == "08:00", (
             "failed regeneration must not touch the completed instance"
         )
@@ -378,12 +463,11 @@ def test_get_by_definition_and_date_round_trip(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         created = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp(), due_time="17:30"
-        )
-        fetched = await instances.get(definition.id, D1)
+            definition.id, child.id, D1, _now_stamp(), due_time="17:30", window="morning")
+        fetched = await instances.get(definition.id, child.id, D1, "morning")
         assert fetched == created
         assert fetched.due_time == "17:30"
-        assert await instances.get(definition.id, D2) is None
+        assert await instances.get(definition.id, child.id, D2, "morning") is None
         return fetched
 
     _with_db(tmp_path, "get-round-trip.db")(_body)
@@ -392,7 +476,7 @@ def test_get_by_definition_and_date_round_trip(tmp_path) -> None:
 def test_get_by_id_round_trip(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        created = await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        created = await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
         fetched = await instances.get_by_id(created.id)
         assert fetched == created
         assert await instances.get_by_id(999) is None
@@ -409,11 +493,10 @@ def test_list_by_child_and_date_filters(tmp_path) -> None:
         other_definition = await definitions.create(
             "Make bed", rule.id, _now_stamp(), assignee_child_ids=[other.id]
         )
-        await instances.upsert(definition.id, child.id, D1, _now_stamp())
-        await instances.upsert(definition.id, child.id, D2, _now_stamp())
+        await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
+        await instances.upsert(definition.id, child.id, D2, _now_stamp(), window="morning")
         await instances.upsert(
-            other_definition.id, other.id, D1, _now_stamp()
-        )
+            other_definition.id, other.id, D1, _now_stamp(), window="morning")
         mine = await instances.list_by_child_and_date(child.id, D1)
         assert len(mine) == 1
         assert mine[0].child_id == child.id
@@ -427,7 +510,7 @@ def test_list_by_date_range_filters_and_orders(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         for day in (D1, D3, D5, D20):
-            await instances.upsert(definition.id, child.id, day, _now_stamp())
+            await instances.upsert(definition.id, child.id, day, _now_stamp(), window="morning")
         found = await instances.list_by_date_range(
             child.id, D2, D6
         )
@@ -461,18 +544,17 @@ def test_list_by_date_range_rejects_inverted_range(tmp_path) -> None:
 def test_delete_future_uncompleted_removes_only_eligible(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
-        await instances.upsert(definition.id, child.id, D1, _now_stamp())
+        await instances.upsert(definition.id, child.id, D1, _now_stamp(), window="morning")
         future_open = await instances.upsert(
-            definition.id, child.id, D5, _now_stamp()
-        )
-        await instances.upsert(definition.id, child.id, D6, _now_stamp())
+            definition.id, child.id, D5, _now_stamp(), window="morning")
+        await instances.upsert(definition.id, child.id, D6, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, D5
         )
         assert deleted == 2
-        assert await instances.get(definition.id, D1) is not None
-        assert await instances.get(definition.id, D5) is None
-        assert await instances.get(definition.id, D6) is None
+        assert await instances.get(definition.id, child.id, D1, "morning") is not None
+        assert await instances.get(definition.id, child.id, D5, "morning") is None
+        assert await instances.get(definition.id, child.id, D6, "morning") is None
         return deleted
 
     _with_db(tmp_path, "delete-future.db")(_body)
@@ -482,11 +564,9 @@ def test_delete_future_uncompleted_spares_completed(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         completed = await instances.upsert(
-            definition.id, child.id, D5, _now_stamp()
-        )
+            definition.id, child.id, D5, _now_stamp(), window="morning")
         open_one = await instances.upsert(
-            definition.id, child.id, D6, _now_stamp()
-        )
+            definition.id, child.id, D6, _now_stamp(), window="morning")
         await events.append(
             completed.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -521,12 +601,12 @@ def test_delete_future_uncompleted_cannot_backdate_cutoff(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         today = datetime.date.today().isoformat()
-        await instances.upsert(definition.id, child.id, today, _now_stamp())
+        await instances.upsert(definition.id, child.id, today, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, "2020-01-01"
         )
         assert deleted == 1
-        assert await instances.get(definition.id, today) is None
+        assert await instances.get(definition.id, child.id, today, "morning") is None
         return deleted
 
     _with_db(tmp_path, "delete-no-backdate.db")(_body)
@@ -541,13 +621,13 @@ def test_delete_future_uncompleted_other_definitions_untouched(
             "Make bed", (await rules.create("daily", D1)).id,
             _now_stamp(), assignee_child_ids=[child.id],
         )
-        await instances.upsert(definition.id, child.id, D20, _now_stamp())
-        await instances.upsert(other.id, child.id, D20, _now_stamp())
+        await instances.upsert(definition.id, child.id, D20, _now_stamp(), window="morning")
+        await instances.upsert(other.id, child.id, D20, _now_stamp(), window="morning")
         deleted = await instances.delete_future_uncompleted(
             definition.id, D20
         )
         assert deleted == 1
-        assert await instances.get(other.id, D20) is not None
+        assert await instances.get(other.id, child.id, D20, "morning") is not None
         return deleted
 
     _with_db(tmp_path, "delete-scoped.db")(_body)
@@ -562,8 +642,7 @@ def test_append_completion_round_trips(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         event = await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -582,8 +661,7 @@ def test_append_panel_event_round_trips(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         event = await events.append(
             instance.id, child.id, "completed", "panel", _now_stamp(), False,
             actor_user_id=None,
@@ -600,8 +678,7 @@ def test_append_uncompleted_with_optional_on_time(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         reversal = await events.append(
             instance.id, child.id, "uncompleted", "user", _now_stamp(), None,
             actor_user_id="user-1",
@@ -622,8 +699,7 @@ def test_append_validation_rejects_bad_shapes(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         with pytest.raises(ValueError, match="event_type"):
             await events.append(
                 instance.id, child.id, "missed", "user", _now_stamp(), True,
@@ -670,8 +746,7 @@ def test_append_rejects_non_utc_timestamps(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         for bad in (
             "2026-09-14 12:00:00+00:00",  # space separator
             "2026-09-14T12:00:00",  # no offset
@@ -695,8 +770,7 @@ def test_append_requires_was_on_time_strict_bool(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         for bad in (0.5, 1.5, 2, "true", 1):
             with pytest.raises(ValueError, match="must be True, False"):
                 await events.append(
@@ -717,8 +791,7 @@ def test_append_rejects_unknown_instance_or_child(tmp_path) -> None:
                 actor_user_id="user-1",
             )
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         with pytest.raises(ValueError, match="belongs to child"):
             await events.append(
                 instance.id, 999, "completed", "user", _now_stamp(), True,
@@ -740,8 +813,7 @@ def test_completed_uncompleted_recompleted_three_ordered_events(
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
             instance.id, child.id, "completed", "panel", _now_stamp(), True
         )
@@ -771,8 +843,7 @@ def test_get_latest_for_instance_none_when_untouched(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         return await events.get_latest_for_instance(instance.id)
 
     assert _with_db(tmp_path, "latest-none.db")(_body) is None
@@ -782,8 +853,7 @@ def test_get_latest_returns_uncompleted_when_reversed(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
             instance.id, child.id, "completed", "panel", _now_stamp(), True
         )
@@ -802,14 +872,11 @@ def test_list_by_child_and_date_range_scopes_by_due_date(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         early = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         inside = await instances.upsert(
-            definition.id, child.id, D2, _now_stamp()
-        )
+            definition.id, child.id, D2, _now_stamp(), window="morning")
         late = await instances.upsert(
-            definition.id, child.id, D20, _now_stamp()
-        )
+            definition.id, child.id, D20, _now_stamp(), window="morning")
         await events.append(
             early.id, child.id, "completed", "panel", _now_stamp(), True
         )
@@ -847,8 +914,7 @@ def test_list_events_occurred_at_scope_variant(tmp_path) -> None:
     async def _body(database, children, rules, definitions, instances,
                     events, child, definition):
         late_due = await instances.upsert(
-            definition.id, child.id, D20, _now_stamp()
-        )
+            definition.id, child.id, D20, _now_stamp(), window="morning")
         # The event's occurred_at is today at 08:00 UTC (_now_stamp);
         # the instance is due D20.  The due-date scope queried on D20
         # still finds the event (scoping follows the instance's due
@@ -1028,8 +1094,7 @@ def test_completion_events_dao_instance_has_no_mutation_capability(
 
         database.execute = _recording
         instance = await instances.upsert(
-            definition.id, child.id, D1, _now_stamp()
-        )
+            definition.id, child.id, D1, _now_stamp(), window="morning")
         await events.append(
             instance.id, child.id, "completed", "user", _now_stamp(), True,
             actor_user_id="user-1",
@@ -1059,8 +1124,7 @@ def test_append_concurrent_events_serialize(tmp_path) -> None:
                 prepared
             )
             instance = await instances.upsert(
-                definition.id, child.id, D1, _now_stamp()
-            )
+                definition.id, child.id, D1, _now_stamp(), window="morning")
             tasks = [
                 asyncio.ensure_future(
                     events.append(
diff --git a/tests/test_dao_matrix.py b/tests/test_dao_matrix.py
index 9a2b36a..2360ffe 100644
--- a/tests/test_dao_matrix.py
+++ b/tests/test_dao_matrix.py
@@ -112,7 +112,8 @@ class World:
             self.child.id, D1, D2, True, note="test"
         )
         self.instance = await self.instances.upsert(
-            self.definition.id, self.child.id, D1, _stamp()
+            self.definition.id, self.child.id, D1, _stamp(),
+            window="morning",
         )
 
 
@@ -291,12 +292,15 @@ def test_matrix_presence_overrides_crud_and_constraints(tmp_path) -> None:
 def test_matrix_quest_instances_crud_and_constraints(tmp_path) -> None:
     async def _body(w: World):
         # insert (seeded) + read
-        fetched = await w.instances.get(w.definition.id, D1)
+        fetched = await w.instances.get(
+            w.definition.id, w.child.id, D1, "morning"
+        )
         assert fetched == w.instance
         # update path: conflict refresh on the OPEN instance — same row
         # id, snapshot columns (due_time) actually rewritten.
         refreshed = await w.instances.upsert(
-            w.definition.id, w.child.id, D1, _stamp(), due_time="17:00"
+            w.definition.id, w.child.id, D1, _stamp(),
+            window="morning", due_time="17:00",
         )
         assert refreshed.id == fetched.id, (
             "the conflict path must refresh the existing row in place"
@@ -316,15 +320,15 @@ def test_matrix_quest_instances_constraints(tmp_path) -> None:
         # proving the schema backs the DAO's idempotency.
         await w.database.execute(
             "INSERT INTO quest_instances (definition_id, child_id, "
-            "due_date, due_time, generated_at) "
-            "VALUES (?, ?, ?, NULL, ?)",
+            "window, due_date, due_time, generated_at) "
+            "VALUES (?, ?, 'morning', ?, NULL, ?)",
             (w.definition.id, w.child.id, D20, _stamp()),
         )
         with pytest.raises(sqlite3.IntegrityError):
             await w.database.execute(
                 "INSERT INTO quest_instances (definition_id, child_id, "
-                "due_date, due_time, generated_at) "
-                "VALUES (?, ?, ?, NULL, ?)",
+                "window, due_date, due_time, generated_at) "
+                "VALUES (?, ?, 'morning', ?, NULL, ?)",
                 (w.definition.id, w.child.id, D20, _stamp()),
             )
         # delete future uncompleted removes D20's open instance
@@ -340,10 +344,11 @@ def test_matrix_quest_instances_constraints(tmp_path) -> None:
 def test_matrix_quest_instances_upsert_idempotency(tmp_path) -> None:
     async def _body(w: World):
         first = await w.instances.upsert(
-            w.definition.id, w.child.id, D2, _stamp()
+            w.definition.id, w.child.id, D2, _stamp(), window="morning"
         )
         second = await w.instances.upsert(
-            w.definition.id, w.child.id, D2, _stamp(), due_time="17:00"
+            w.definition.id, w.child.id, D2, _stamp(),
+            window="morning", due_time="17:00",
         )
         assert second.id == first.id
         count = await w.database.fetch_one(
diff --git a/tests/test_dao_rules.py b/tests/test_dao_rules.py
index 0ae00b6..8784e73 100644
--- a/tests/test_dao_rules.py
+++ b/tests/test_dao_rules.py
@@ -927,7 +927,7 @@ def test_definition_assignment_leaves_history_rows_alone(
         # this table, but its row shape proves assignment isolation).
         await database.execute(
             "INSERT INTO quest_instances (definition_id, child_id, "
-            "due_date, generated_at) VALUES (?, ?, ?, ?)",
+            "window, due_date, generated_at) VALUES (?, ?, 'morning', ?, ?)",
             (definition.id, child.id, "2026-09-15", NOW),
         )
         await definitions.add_assignee(definition.id, other.id)
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index ce81da3..8eee3ee 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -149,13 +149,16 @@ 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, and migration 4 adds the windows table.
+    model, migration 4 adds the windows table, and migration 5 rebuilds
+    quest_instances onto the widened (definition, child, date, window)
+    key.
     """
     assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
     assert callable(MIGRATIONS[1])
     assert callable(MIGRATIONS[2])
     assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
-    assert len(MIGRATIONS) == 4
+    assert callable(MIGRATIONS[4])
+    assert len(MIGRATIONS) == 5
 
 
 # ---------------------------------------------------------------------------
@@ -425,6 +428,20 @@ _V2_DEFINITIONS_DDL = [
     """,
 ]
 
+_V2_INSTANCES_DDL = [
+    """
+    CREATE TABLE IF NOT EXISTS quest_instances (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        definition_id INTEGER NOT NULL REFERENCES quest_definitions(id),
+        child_id INTEGER NOT NULL REFERENCES children(id),
+        due_date TEXT NOT NULL,
+        due_time TEXT,
+        generated_at TEXT NOT NULL,
+        UNIQUE (definition_id, due_date)
+    )
+    """,
+]
+
 
 def _seed_v2_multi_assignee_upgrade(database) -> None:
     """Create a database shaped like a post-rename, pre-D-008 dev file.
@@ -433,11 +450,20 @@ def _seed_v2_multi_assignee_upgrade(database) -> None:
     single child_id column and no assignees table exists — stamped at
     version 2, with definition and instance rows that must survive.
     """
+    # Freeze the version-2 shape: definitions carry child_id, the
+    # instance table has no window column, and the assignees/windows
+    # tables do not exist yet — everything else matches the current DDL.
+    _v2_replaced = (
+        "quest_definitions ",
+        "quest_definition_assignees ",
+        "quest_definition_windows ",
+        "quest_instances ",
+    )
     for sql in SCHEMA_V1_STATEMENTS:
-        if "quest_definitions " in sql:
-            continue  # replaced by the legacy-shape DDL below
+        if any(name in sql for name in _v2_replaced):
+            continue
         _run(database.execute(sql))
-    for sql in _V2_DEFINITIONS_DDL:
+    for sql in _V2_DEFINITIONS_DDL + _V2_INSTANCES_DDL:
         _run(database.execute(sql))
     _run(database.execute(VERSION_TABLE_DDL))
     _run(
@@ -506,14 +532,33 @@ def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
             )
         )
         assert assignees == [(1, 2)]
-        # The instance survived, still referencing the definition.
+        # The instance survived the rebuild, pinned to the migrated
+        # window, still referencing the definition.
         instance = _run(
             database.fetch_one(
-                "SELECT definition_id, child_id, due_date "
+                "SELECT definition_id, child_id, window, due_date "
                 "FROM quest_instances WHERE id = 1"
             )
         )
-        assert instance == (1, 2, "2026-09-15")
+        assert instance == (1, 2, "morning", "2026-09-15")
+        # The rebuilt instance table carries the widened unique key.
+        index_rows = _run(
+            database.fetch_all(
+                "PRAGMA index_list(quest_instances)"
+            )
+        )
+        unique_columns = []
+        for index_row in index_rows:
+            if index_row[2]:  # unique flag
+                info = _run(
+                    database.fetch_all(
+                        f"PRAGMA index_info('{index_row[1]}')"
+                    )
+                )
+                unique_columns.append([col[2] for col in info])
+        assert [
+            "definition_id", "child_id", "due_date", "window"
+        ] in unique_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
diff --git a/tests/test_schema.py b/tests/test_schema.py
index bd18996..e27c774 100644
--- a/tests/test_schema.py
+++ b/tests/test_schema.py
@@ -1722,8 +1722,8 @@ def test_applying_presence_ddl_twice_keeps_rows(tmp_path) -> None:
 # ---------------------------------------------------------------------------
 
 
-def _insert_instance(database, definition_id=1, **overrides):
-    """Insert one valid task_instance, returning the instance id.
+def _insert_instance(database, definition_id=1, window="morning", **overrides):
+    """Insert one valid quest_instance, returning the instance id.
 
     Pass ``_OMIT`` as a value to leave that column out of the INSERT so
     its DDL DEFAULT applies.
@@ -1731,6 +1731,7 @@ def _insert_instance(database, definition_id=1, **overrides):
     values: dict[str, object] = {
         "definition_id": definition_id,
         "child_id": 1,
+        "window": window,
         "due_date": "2026-09-14",
         "generated_at": "2026-09-14T00:00:00+00:00",
     }
@@ -1829,6 +1830,7 @@ def test_quest_instances_columns_types_and_constraints(tmp_path) -> None:
             ("id", "INTEGER", 0, None, 1),
             ("definition_id", "INTEGER", 1, None, 0),
             ("child_id", "INTEGER", 1, None, 0),
+            ("window", "TEXT", 1, None, 0),
             ("due_date", "TEXT", 1, None, 0),
             ("due_time", "TEXT", 0, None, 0),
             ("generated_at", "TEXT", 1, None, 0),
@@ -1884,7 +1886,7 @@ def test_completion_events_foreign_keys_declared(tmp_path) -> None:
         _run(database.close())
 
 
-def test_quest_instances_unique_index_on_definition_and_date(tmp_path) -> None:
+def test_quest_instances_unique_index_on_widened_key(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-unique.db")
     try:
         _apply(database)
@@ -1897,15 +1899,20 @@ def test_quest_instances_unique_index_on_definition_and_date(tmp_path) -> None:
         )
         implicit = [(name, sql) for name, sql in rows if sql is None]
         assert len(implicit) == 1, (
-            "the (definition_id, due_date) UNIQUE constraint must be backed "
-            "by exactly one implicit unique index"
+            "the (definition_id, child_id, due_date, window) UNIQUE "
+            "constraint must be backed by exactly one implicit unique index"
         )
         columns = _run(
             database.fetch_all(
                 f"PRAGMA index_info('{implicit[0][0]}')"
             )
         )
-        assert [row[2] for row in columns] == ["definition_id", "due_date"]
+        assert [row[2] for row in columns] == [
+            "definition_id",
+            "child_id",
+            "due_date",
+            "window",
+        ]
         # The implicit index is genuinely unique: UNIQUE(...) is the only
         # constraint SQLite backs with a sql-less autoindex on this table.
         row = _run(
@@ -1932,12 +1939,16 @@ def test_quest_instances_valid_insert_round_trips(tmp_path) -> None:
         instance_id = _insert_instance(database, definition_id=1, child_id=1)
         row = _run(
             database.fetch_one(
-                "SELECT definition_id, child_id, due_date, due_time, generated_at "
+                "SELECT definition_id, child_id, window, due_date, "
+                "due_time, generated_at "
                 "FROM quest_instances WHERE id = ?",
                 (instance_id,),
             )
         )
-        assert row == (1, 1, "2026-09-14", None, "2026-09-14T00:00:00+00:00")
+        assert row == (
+            1, 1, "morning", "2026-09-14", None,
+            "2026-09-14T00:00:00+00:00",
+        )
     finally:
         _run(database.close())
 
@@ -1961,7 +1972,7 @@ def test_quest_instances_due_time_round_trips(tmp_path) -> None:
         _run(database.close())
 
 
-def test_quest_instances_duplicate_definition_date_fails(tmp_path) -> None:
+def test_quest_instances_duplicate_tuple_fails(tmp_path) -> None:
     database = _open_db(tmp_path / "task-instances-duplicate.db")
     try:
         _apply(database)
@@ -1973,6 +1984,78 @@ def test_quest_instances_duplicate_definition_date_fails(tmp_path) -> None:
         _run(database.close())
 
 
+def test_quest_instances_duplicate_tuple_different_child_allowed(
+    tmp_path,
+) -> None:
+    """D-008: same definition, date and window for two different
+    assignees of the same definition are two distinct instances."""
+    database = _open_db(tmp_path / "task-instances-shared.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Bo", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _run(
+            database.execute(
+                "INSERT INTO quest_definition_assignees "
+                "(definition_id, child_id) VALUES (1, 2)"
+            )
+        )
+        _insert_instance(database, definition_id=1, child_id=1)
+        _insert_instance(database, definition_id=1, child_id=2)
+        row = _run(database.fetch_one("SELECT COUNT(*) FROM quest_instances"))
+        assert row == (2,)
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_duplicate_tuple_different_window_allowed(
+    tmp_path,
+) -> None:
+    """D-008 twice-daily case: one definition, one child, one date,
+    two windows — two rows."""
+    database = _open_db(tmp_path / "task-instances-twice-daily.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        _insert_instance(database, definition_id=1, window="morning")
+        _insert_instance(database, definition_id=1, window="evening")
+        rows = _run(
+            database.fetch_all(
+                "SELECT window FROM quest_instances ORDER BY window"
+            )
+        )
+        assert [row[0] for row in rows] == ["evening", "morning"]
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_unknown_window_fails(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-bad-window.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _insert_instance(database, definition_id=1, window="noon")
+    finally:
+        _run(database.close())
+
+
+def test_quest_instances_window_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "task-instances-window-null.db")
+    try:
+        _apply(database)
+        _setup_definition(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _insert_instance(database, definition_id=1, window=_OMIT)
+    finally:
+        _run(database.close())
+
+
 def test_quest_instances_same_date_different_definitions_allowed(
     tmp_path,
 ) -> None: