diff --git a/docs/observability/relay-shared-metrics.md b/docs/observability/relay-shared-metrics.md
index f5af4e8e80..b538011675 100644
--- a/docs/observability/relay-shared-metrics.md
+++ b/docs/observability/relay-shared-metrics.md
@@ -301,10 +301,20 @@ telemetry:
 - Like `enabled`, `send` is profile-owned and is not overridden by
   managed-scope configuration.
 
-**Only packages for periods on or after the opt-in day are ever sent.** The
-opt-in day (UTC) is recorded when `send` first becomes true, and any package
-whose `period_start` predates it is permanently excluded, however late it was
-created.
+**A package is only sent when its whole period falls inside a recorded
+consent window.** Consent is stored as explicit intervals in the shared-
+metrics SQLite store (`send_consent_windows`): a window opens when `send:
+true` is first observed, is confirmed forward by every later observation,
+and closes — at the last *confirmed* moment, never at the wall clock — when
+`send: false` is observed. A single reconciler derives this table from the
+config on every process start, so wizard changes, hand-edits to
+`config.yaml`, and mid-pass revocations all take the same path, and no
+transition can be missed by any of them.
+
+Any package whose period predates the first window, falls between windows,
+or runs past the newest confirmed moment is excluded — the gate fails
+closed. A fresh package therefore waits at most one process start after its
+period completes before becoming eligible.
 
 The gate is on the **period**, not on the package's creation time. One period
 is split across several packages created on different days: a day's first
@@ -389,11 +399,15 @@ before every package, so a pass already in flight stops after the package it
 is currently sending rather than draining its whole batch. It does not delete
 previously transmitted packages, and it does not stop local collection.
 
-Turning sending off also **closes the consent window**. Packages collected
-while it was off are never transmitted, even if sending is later re-enabled —
-re-enabling starts a new window from that day. Without this, a write-once
-opt-in date would have retroactively released the entire refused period the
-next time the user changed their mind.
+Turning sending off also **closes the consent window** — at the last moment
+consent was actually observed, not at the wall clock. Packages whose periods
+fall between one window and the next are never transmitted, even if sending
+is later re-enabled, and this holds for any number of on/off cycles, across
+hand-edits with no process running, and under a clock that jumps backwards
+(window opens are clamped above every timestamp already in the store).
+Unlike the earlier single moving opt-in date, closing and reopening does NOT
+discard the still-undelivered backlog from a previous consented window —
+those packages stay inside their own interval and remain eligible.
 
 ### A.5 Retention
 
diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py
index 5303e020df..804a48ea38 100644
--- a/hermes_cli/config_defaults.py
+++ b/hermes_cli/config_defaults.py
@@ -3334,9 +3334,10 @@ DEFAULT_CONFIG = {
             # Transmit exported packages to the Nous telemetry service.
             # Requires ``enabled``: it never switches collection on by itself,
             # and ``send`` without ``enabled`` is logged as an error rather
-            # than silently doing nothing. Only packages whose period starts
-            # on or after the opt-in day are ever sent, so data collected
-            # before consent stays local.
+            # than silently doing nothing. A package is only sent when its
+            # whole period falls inside a recorded consent window, so data
+            # collected before consent — or while it was withdrawn — stays
+            # local.
             "send": False,
             # Ingest endpoint. Production by default; override for staging or
             # a local test server. Deliberately NOT overridable by an
diff --git a/hermes_cli/observability/relay_shared_metrics.py b/hermes_cli/observability/relay_shared_metrics.py
index 2d7ec0583a..978497945d 100644
--- a/hermes_cli/observability/relay_shared_metrics.py
+++ b/hermes_cli/observability/relay_shared_metrics.py
@@ -1084,60 +1084,26 @@ class _Runtime:
             self._safe(self._send_exported_packages)
 
     def _observe_send_consent(self, send_enabled: bool) -> None:
-        """Close the consent window on a true->false transition.
+        """Reconcile consent windows with the observed config state.
 
-        Persists the last-seen send state so a change is detected even though
-        this runs in a fresh process each time. Only the falling edge matters:
-        opening a new window is the sender's job, on the next enabled pass.
+        Thin wrapper over the SINGLE consent writer. The old edge-detection
+        body (last-seen key, rising/falling branches) is gone: reconciliation
+        derives the correct window state from what it observes, so there is
+        no transition to miss and no ordering between callers to get wrong.
 
-        Failures here must never break the export hook, but they are logged at
+        Failures must never break the export hook, but they are logged at
         warning rather than debug: silently failing to close a consent window
         is a privacy-relevant event, not routine bookkeeping.
         """
         try:
             from hermes_cli.observability.shared_metrics_sender import (
-                LAST_SEEN_SEND_KEY,
-                opt_in_period,
-                record_revoked,
+                reconcile_send_consent,
             )
             from hermes_cli.sqlite_util import write_txn
 
-            current = "1" if send_enabled else "0"
             with self.subscriber.store._connection() as connection:
                 with write_txn(connection):
-                    row = connection.execute(
-                        "SELECT value FROM telemetry_state WHERE key = ?",
-                        (LAST_SEEN_SEND_KEY,),
-                    ).fetchone()
-                    previous = str(row[0]) if row is not None else None
-
-                    if send_enabled:
-                        # Open the window HERE, on the rising edge, rather than
-                        # leaving it to the sender's first claim. The sender
-                        # only runs when there is something to send, so a user
-                        # who opts in and then opts out before any package
-                        # exists would otherwise have no window to close, and
-                        # record_revoked (which requires one) would no-op.
-                        opt_in_period(connection)
-                    elif previous == "1":
-                        # `previous == "1"` is the true falling edge. Widening
-                        # this to an unconditional else would be behaviourally
-                        # equivalent today — record_revoked is idempotent and
-                        # no-ops without an open window — so no test can tell
-                        # the two apart. It is written as an edge anyway
-                        # because that is the property intended, and a future
-                        # change to record_revoked should not silently turn
-                        # every disabled pass into a revocation.
-                        record_revoked(connection)
-
-                    if previous != current:
-                        connection.execute(
-                            """
-                            INSERT INTO telemetry_state(key, value) VALUES (?, ?)
-                            ON CONFLICT(key) DO UPDATE SET value = excluded.value
-                            """,
-                            (LAST_SEEN_SEND_KEY, current),
-                        )
+                    reconcile_send_consent(connection, send_enabled)
         except Exception:
             logger.warning(
                 "Unable to record a shared-metrics consent transition",
@@ -1259,8 +1225,54 @@ def handles_hook(hook_name: str) -> bool:
     return hook_name in HANDLED_HOOKS and enabled()
 
 
+_consent_reconcile_done = False
+
+
+def _reconcile_send_consent_once() -> None:
+    """Reconcile consent windows with config, once per process.
+
+    Runs BEFORE and INDEPENDENT of the collection gate — that placement is
+    the fix for the round-5 D1 leak, where the only idle-path consent
+    observer sat behind ``handles_hook()`` and became dead code the moment
+    ``enabled: false`` was set. A user with collection off still gets their
+    send-consent windows reconciled here.
+
+    Skipped only when there is no store on disk AND consent is off: with no
+    store there are no packages, so there is nothing a window could protect,
+    and creating ``~/.hermes/telemetry`` for every fully-disabled user would
+    be a behaviour change in the wrong direction.
+    """
+    global _consent_reconcile_done
+    if _consent_reconcile_done:
+        return
+    _consent_reconcile_done = True
+    try:
+        from hermes_cli.config import read_raw_config_readonly
+        from hermes_cli.observability.shared_metrics import SharedMetricsStore
+        from hermes_cli.observability.shared_metrics_send_config import (
+            resolve_send_config,
+        )
+        from hermes_cli.observability.shared_metrics_sender import (
+            reconcile_send_consent,
+        )
+        from hermes_cli.sqlite_util import write_txn
+
+        resolved = resolve_send_config(read_raw_config_readonly() or {})
+        store = SharedMetricsStore()
+        if not resolved.send and not store.database_path.exists():
+            return
+        with store._connection() as connection:
+            with write_txn(connection):
+                reconcile_send_consent(connection, resolved.send)
+    except Exception:
+        logger.warning(
+            "Unable to reconcile shared-metrics send consent", exc_info=True
+        )
+
+
 def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
     """Project one Hermes lifecycle event into the core Relay integration."""
+    _reconcile_send_consent_once()
     if not handles_hook(hook_name):
         return
     if not relay_runtime.relay_instrumentation_enabled():
diff --git a/hermes_cli/observability/shared_metrics.py b/hermes_cli/observability/shared_metrics.py
index bf5c1fb0bf..32dfc4fff6 100644
--- a/hermes_cli/observability/shared_metrics.py
+++ b/hermes_cli/observability/shared_metrics.py
@@ -338,6 +338,7 @@ class SharedMetricsStore:
             """
         )
         SharedMetricsStore._add_send_columns(connection)
+        SharedMetricsStore._add_consent_tables(connection)
         connection.execute(
             """
             INSERT INTO telemetry_state(key, value)
@@ -383,6 +384,55 @@ class SharedMetricsStore:
                     f"ALTER TABLE package_outbox ADD COLUMN {column} {declaration}"
                 )
 
+    @staticmethod
+    def _add_consent_tables(connection: sqlite3.Connection) -> None:
+        """Create the consent-window tables, idempotently.
+
+        Additive like ``_add_send_columns`` — the schema version is
+        deliberately NOT bumped, and old readers never touch these tables.
+
+        ``send_consent_windows`` records consent as explicit intervals rather
+        than a moving day-stamp: a window is opened when send consent is
+        observed, heartbeat-confirmed on every later observation, and closed
+        at the LAST CONFIRMED moment (never "now") when consent is observed
+        withdrawn. Consent is asserted only for time that was actually
+        observed, so unobserved gaps — a hand-edited config with no process
+        running — fail closed by construction.
+
+        ``consent_marks`` holds two monotonic high-water marks with strictly
+        separated roles:
+
+        - ``obs``: the latest observation stamp ever seen. Advanced only by
+          the reconciler. Confirms consent and clamps window closes.
+        - ``data``: the latest package ``period_end`` ever stored. Advanced
+          only by the package writer. Clamps window OPENS, so a rolled-back
+          clock can never open a window underneath packages that already
+          exist on disk.
+
+        The separation is load-bearing: letting data stamps confirm consent
+        re-created a refused-window leak (packages stored during an off
+        window would vouch for it), and letting observation stamps clamp
+        opens is not enough on its own to stop a rollback sliding a window
+        under existing refused data.
+        """
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS send_consent_windows (
+                opened_at TEXT NOT NULL,
+                last_confirmed_at TEXT NOT NULL,
+                closed_at TEXT
+            )
+            """
+        )
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS consent_marks (
+                name TEXT PRIMARY KEY CHECK (name IN ('obs', 'data')),
+                stamp TEXT NOT NULL
+            )
+            """
+        )
+
     @staticmethod
     def _create_counter_aggregates_table(connection: sqlite3.Connection) -> None:
         connection.execute(
@@ -617,6 +667,16 @@ class SharedMetricsStore:
                 payload["generated_at"],
             ),
         )
+        # Advance the data high-water mark. This is the ONLY writer of the
+        # 'data' mark: it clamps consent-window opens so a rolled-back clock
+        # can never open a window underneath packages that already exist.
+        connection.execute(
+            """
+            INSERT INTO consent_marks(name, stamp) VALUES ('data', ?)
+            ON CONFLICT(name) DO UPDATE SET stamp = MAX(stamp, excluded.stamp)
+            """,
+            (payload["period_end"],),
+        )
         for row in rows:
             connection.execute(
                 """
diff --git a/hermes_cli/observability/shared_metrics_sender.py b/hermes_cli/observability/shared_metrics_sender.py
index 49c2397f6e..389611ed37 100644
--- a/hermes_cli/observability/shared_metrics_sender.py
+++ b/hermes_cli/observability/shared_metrics_sender.py
@@ -17,7 +17,10 @@ file. See Appendix A.7 of ``docs/observability/relay-shared-metrics.md``.
 **Consent is gated on the package's PERIOD, not its creation time.** One
 period is split across packages created on different days, so a created-at
 gate would send a period's tail while dropping its head and silently
-undercount the opt-in day.
+undercount the first consented day. The gate itself is interval containment:
+the period must fall entirely inside a recorded consent window
+(``send_consent_windows``), maintained by the single ``reconcile_send_consent``
+writer below.
 """
 
 from __future__ import annotations
@@ -88,18 +91,6 @@ _PERMANENT_STATUSES = frozenset({400, 413})
 #: doomed package at the head of the queue.
 MAX_SEND_ATTEMPTS = 25
 
-OPT_IN_PERIOD_KEY = "send_opt_in_period"
-
-#: Set when sending is turned off, cleared by the next enabled pass (which
-#: also advances OPT_IN_PERIOD_KEY). This is what makes consent revocation
-#: permanent for the packages collected while it was off.
-SEND_REVOKED_KEY = "send_revoked"
-
-#: Last send-consent state this machine observed ("1"/"0"). Persisted because
-#: each hook fires in a fresh process, so a true->false edge is only visible
-#: by comparing against what was recorded last time.
-LAST_SEEN_SEND_KEY = "send_last_seen"
-
 
 def _utc_now() -> datetime:
     return datetime.now(timezone.utc)
@@ -173,46 +164,86 @@ def _retry_after_seconds(value: str | None, default: int) -> int:
         return default
 
 
-def opt_in_period(connection: sqlite3.Connection, *, now: datetime | None = None) -> str:
-    """Return the day (UTC) from which packages may be sent.
+def reconcile_send_consent(
+    connection: sqlite3.Connection,
+    send_enabled: bool,
+    *,
+    now: datetime | None = None,
+) -> None:
+    """Reconcile the consent-window table with the observed config state.
 
-    Must run inside a write transaction.
+    THE ONLY writer of consent state. Must run inside a write transaction.
+    A pure function of (config, now, store): call it from anywhere, any
+    number of times, in any order — the resulting windows are the same. This
+    replaces the previous edge-detection design, whose three partial
+    observers (wizard, relay, mid-pass) each covered a different subset of
+    transitions and repeatedly leaked the transitions between the subsets.
 
-    This is the CURRENT consent window's start, not a permanent first-ever
-    opt-in date. If the user previously turned sending off, ``record_revoked``
-    stamps that; the next enabled pass advances the gate to the day sending
-    resumed, so packages collected during the opted-out window are never
-    transmitted. Without that advance, re-enabling would retroactively release
-    the entire period the user had explicitly refused.
+    Timestamp discipline (each rule is load-bearing; see the validation
+    harness in tests/hermes_cli/test_shared_metrics_consent_windows.py):
+
+    - The 'obs' mark advances to every observation stamp, monotonically.
+      An open window's ``last_confirmed_at`` follows it: consent is asserted
+      only for time that was actually observed.
+    - A close is stamped at ``last_confirmed_at`` — never "now" — so an
+      unobserved gap (hand-edited config, machine off for 90 days) is never
+      inside a window and fails closed.
+    - An open clamps to ``max(now, obs, data)``: a rolled-back clock cannot
+      open a window underneath refused packages already on disk, and cannot
+      make the new window adjacent to the previous close.
     """
-    today = (now or _utc_now()).date().isoformat()
+    stamp = _isoformat(now or _utc_now())
+    connection.execute(
+        """
+        INSERT INTO consent_marks(name, stamp) VALUES ('obs', ?)
+        ON CONFLICT(name) DO UPDATE SET stamp = MAX(stamp, excluded.stamp)
+        """,
+        (stamp,),
+    )
+    marks = dict(
+        connection.execute("SELECT name, stamp FROM consent_marks").fetchall()
+    )
+    obs = marks["obs"]  # >= stamp; immune to clock rollback
+    data = marks.get("data")
 
-    revoked = _state_get(connection, SEND_REVOKED_KEY)
-    if revoked:
-        # Sending resumed after a revocation: the new window starts today.
-        _state_set(connection, OPT_IN_PERIOD_KEY, today)
+    open_row = connection.execute(
+        "SELECT rowid FROM send_consent_windows WHERE closed_at IS NULL"
+    ).fetchone()
+
+    if send_enabled:
+        if open_row is None:
+            opened = max(x for x in (obs, data) if x is not None)
+            connection.execute(
+                "INSERT INTO send_consent_windows(opened_at, last_confirmed_at)"
+                " VALUES (?, ?)",
+                (opened, opened),
+            )
+        else:
+            connection.execute(
+                "UPDATE send_consent_windows"
+                " SET last_confirmed_at = MAX(last_confirmed_at, ?)"
+                " WHERE rowid = ?",
+                (obs, open_row[0]),
+            )
+    elif open_row is not None:
         connection.execute(
-            "DELETE FROM telemetry_state WHERE key = ?", (SEND_REVOKED_KEY,)
+            "UPDATE send_consent_windows SET closed_at = last_confirmed_at"
+            " WHERE rowid = ?",
+            (open_row[0],),
         )
-        return today
-
-    existing = _state_get(connection, OPT_IN_PERIOD_KEY)
-    if existing:
-        return existing
-
-    _state_set(connection, OPT_IN_PERIOD_KEY, today)
-    return today
 
 
-def record_revoked(connection: sqlite3.Connection) -> None:
-    """Mark that sending was turned off, closing the current consent window.
-
-    Idempotent. The marker is only cleared by the next enabled pass, which
-    also advances the gate — so any package collected between the two events
-    stays local permanently.
-    """
-    if _state_get(connection, OPT_IN_PERIOD_KEY):
-        _state_set(connection, SEND_REVOKED_KEY, "1")
+#: Claim-time consent predicate: the package's period must fall entirely
+#: inside SOME recorded consent window. An open window vouches only up to its
+#: last confirmed moment, so a package whose period runs past it waits for
+#: the next reconcile heartbeat (fail-closed; released within one hook fire).
+CONSENT_GATE_SQL = """EXISTS (
+    SELECT 1 FROM send_consent_windows w
+    WHERE package_outbox.period_start >= w.opened_at
+      AND package_outbox.period_end <=
+          CASE WHEN w.closed_at IS NULL THEN w.last_confirmed_at
+               ELSE w.closed_at END
+)"""
 
 
 def _state_get(connection: sqlite3.Connection, key: str) -> str | None:
@@ -278,7 +309,6 @@ class SharedMetricsSender:
         """
         with self._store._connection() as connection:
             with write_txn(connection):
-                period = opt_in_period(connection, now=now)
                 stamp = _isoformat(now)
                 lease_until = now + timedelta(seconds=_CLAIM_LEASE_SECONDS)
 
@@ -286,6 +316,10 @@ class SharedMetricsSender:
                 exclusion = (
                     f" AND package_id NOT IN ({placeholders})" if seen else ""
                 )
+                # Consent is a READ here — the claim must never mutate the
+                # window table. The old design's opt_in_period() call at this
+                # exact spot meant selecting a row could rewrite what was
+                # permitted to be sent (and did, under a rolled-back clock).
                 row = connection.execute(
                     f"""
                     SELECT package_id, payload_json, sent_install_id
@@ -293,13 +327,13 @@ class SharedMetricsSender:
                     WHERE exported_at IS NOT NULL
                       AND (send_state IS NULL OR send_state = 'pending')
                       AND (next_attempt_at IS NULL OR next_attempt_at <= ?)
-                      AND substr(period_start, 1, 10) >= ?
+                      AND {CONSENT_GATE_SQL}
                       AND send_attempts < ?
                       {exclusion}
                     ORDER BY created_at, package_id
                     LIMIT 1
                     """,
-                    (stamp, period, MAX_SEND_ATTEMPTS, *sorted(seen)),
+                    (stamp, MAX_SEND_ATTEMPTS, *sorted(seen)),
                 ).fetchone()
                 if row is None:
                     return None
@@ -523,13 +557,12 @@ class SharedMetricsSender:
         for _ in range(MAX_PACKAGES_PER_PASS):
             if not self._still_consented():
                 # The user turned sending off while this pass was running.
-                # Stop without transmitting anything further, and close the
-                # consent window. This covers only the mid-pass case; a
-                # revocation made while no pass is running is caught by the
-                # relay's edge detector before it early-returns, because this
-                # loop would never run to observe it.
+                # Stop without transmitting anything further, and reconcile
+                # so the window closes at its last confirmed moment. This is
+                # the same single writer every other observation point uses —
+                # not a separate recording mechanism.
                 logger.info("Shared-metrics sending disabled mid-pass; stopping")
-                self._record_revocation()
+                self._reconcile(send_enabled=False)
                 break
             try:
                 package = self._claim_next(self._now(), seen)
@@ -561,14 +594,18 @@ class SharedMetricsSender:
                 outcome.deferred += 1
         return outcome
 
-    def _record_revocation(self) -> None:
-        """Close the consent window after an observed revocation."""
+    def _reconcile(self, *, send_enabled: bool) -> None:
+        """Run the single consent writer from within a pass."""
         try:
             with self._store._connection() as connection:
                 with write_txn(connection):
-                    record_revoked(connection)
+                    reconcile_send_consent(
+                        connection, send_enabled, now=self._now()
+                    )
         except Exception:
-            logger.debug("Unable to record consent revocation", exc_info=True)
+            logger.warning(
+                "Unable to reconcile shared-metrics consent", exc_info=True
+            )
 
     def _still_consented(self) -> bool:
         """Re-read profile-owned send consent.
diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py
index 5a000d0374..6af33a000d 100644
--- a/hermes_cli/setup.py
+++ b/hermes_cli/setup.py
@@ -2481,43 +2481,28 @@ def setup_telemetry(config: dict):
 
 
 def _record_send_consent_change(*, enabled: bool) -> None:
-    """Persist a consent transition at the moment the user makes it.
+    """Reconcile consent windows at the moment the user decides.
 
-    Enabling stamps the day so the gate excludes anything collected earlier.
-    Disabling stamps a revocation so that if the user ever re-enables, the
-    packages collected while sending was off are never released — the doc
-    promises `send: false` means no further packages leave the machine, and
-    that has to survive a later change of mind.
+    Same single writer as the relay and the sender — reconciliation derives
+    the window state from the observation, so wizard, relay, and mid-pass
+    callers cannot disagree. The relay's once-per-process reconcile would
+    catch this on the next hook fire anyway; running it here just makes the
+    wizard's effect immediate.
     """
     try:
         from hermes_cli.observability.shared_metrics import SharedMetricsStore
         from hermes_cli.observability.shared_metrics_sender import (
-            LAST_SEEN_SEND_KEY,
-            opt_in_period,
-            record_revoked,
+            reconcile_send_consent,
         )
         from hermes_cli.sqlite_util import write_txn
 
         store = SharedMetricsStore()
         with store._connection() as connection:
             with write_txn(connection):
-                if enabled:
-                    opt_in_period(connection)
-                else:
-                    record_revoked(connection)
-                # Keep the relay's edge detector in step. Without this the
-                # wizard's change looks like "no transition" on the next hook
-                # fire, and a later true->false edge could be missed.
-                connection.execute(
-                    """
-                    INSERT INTO telemetry_state(key, value) VALUES (?, ?)
-                    ON CONFLICT(key) DO UPDATE SET value = excluded.value
-                    """,
-                    (LAST_SEEN_SEND_KEY, "1" if enabled else "0"),
-                )
+                reconcile_send_consent(connection, enabled)
     except Exception:
-        # Never block the wizard on telemetry bookkeeping. The sender records
-        # the same transitions on its next pass.
+        # Never block the wizard on telemetry bookkeeping. The relay runs the
+        # same reconciliation on the next lifecycle hook.
         logger.debug("Unable to record shared-metrics consent change", exc_info=True)
 
 
diff --git a/scripts/e2e_shared_metrics_staging.py b/scripts/e2e_shared_metrics_staging.py
index 3e52e61a7a..e0c497a342 100644
--- a/scripts/e2e_shared_metrics_staging.py
+++ b/scripts/e2e_shared_metrics_staging.py
@@ -62,6 +62,31 @@ def main() -> int:
     )
 
     today = datetime.now(timezone.utc).date().isoformat()
+    # The generator only exports COMPLETED periods, so the realistic E2E
+    # package is yesterday's. It also has to be: the consent gate only
+    # releases a package once its whole period is confirmed consented, and
+    # today's period cannot be confirmed before it ends.
+    from datetime import timedelta
+
+    period_day = (
+        datetime.now(timezone.utc).date() - timedelta(days=1)
+    ).isoformat()
+
+    # Open the consent window before the period, confirm it after — exactly
+    # what the runtime reconciler does across two days of hook fires.
+    from hermes_cli.observability.shared_metrics_sender import (
+        reconcile_send_consent,
+    )
+    from hermes_cli.sqlite_util import write_txn
+
+    with store._connection() as connection:
+        with write_txn(connection):
+            reconcile_send_consent(
+                connection,
+                True,
+                now=datetime.now(timezone.utc) - timedelta(days=2),
+            )
+            reconcile_send_consent(connection, True)
     real_install_id = str(uuid.uuid4())
     packages = []
 
@@ -77,8 +102,8 @@ def main() -> int:
             "generated_at": datetime.now(timezone.utc).isoformat().replace(
                 "+00:00", "Z"
             ),
-            "period_start": f"{today}T00:00:00Z",
-            "period_end": f"{today}T23:59:59Z",
+            "period_start": f"{period_day}T00:00:00Z",
+            "period_end": f"{period_day}T23:59:59Z",
             "resource": {
                 "hermes_version": "e2e-test",
                 "os_family": "macos",
@@ -105,11 +130,11 @@ def main() -> int:
                 """,
                 (
                     package_id,
-                    f"{today}T00:00:00Z",
-                    f"{today}T23:59:59Z",
+                    f"{period_day}T00:00:00Z",
+                    f"{period_day}T23:59:59Z",
                     json.dumps(payload),
-                    f"{today}T0{index}:00:00Z",
-                    f"{today}T0{index}:00:01Z",
+                    f"{period_day}T0{index}:00:00Z",
+                    f"{period_day}T0{index}:00:01Z",
                 ),
             )
         packages.append((package_id, metric_count))
diff --git a/tests/hermes_cli/test_setup_telemetry.py b/tests/hermes_cli/test_setup_telemetry.py
index 4f66259eaa..2397524343 100644
--- a/tests/hermes_cli/test_setup_telemetry.py
+++ b/tests/hermes_cli/test_setup_telemetry.py
@@ -33,7 +33,10 @@ def test_disabling_collection_closes_the_send_consent_window(monkeypatch, tmp_pa
     package collected in between.
     """
     from hermes_cli.observability.shared_metrics import SharedMetricsStore
-    from hermes_cli.observability.shared_metrics_sender import SEND_REVOKED_KEY
+    from hermes_cli.observability.shared_metrics_sender import (
+        reconcile_send_consent,
+    )
+    from hermes_cli.sqlite_util import write_txn
 
     store = SharedMetricsStore(
         database_path=tmp_path / "m.db", outbox_directory=tmp_path / "o"
@@ -48,24 +51,21 @@ def test_disabling_collection_closes_the_send_consent_window(monkeypatch, tmp_pa
         "hermes_cli.setup.prompt_yes_no", lambda _question, default: False
     )
     config = {"telemetry": {"shared_metrics": {"enabled": True, "send": True}}}
-    # Consent was granted earlier, so a window is already open — that is
-    # precisely the state whose closure must be recorded.
-    from hermes_cli.sqlite_util import write_txn
-    from hermes_cli.observability.shared_metrics_sender import opt_in_period
-
+    # Consent was granted earlier, so a window is open — that is precisely
+    # the state whose closure must be recorded.
     with store._connection() as connection:
         with write_txn(connection):
-            opt_in_period(connection)
+            reconcile_send_consent(connection, True)
 
     setup_telemetry(config)
 
     assert config["telemetry"]["shared_metrics"]["enabled"] is False
     assert config["telemetry"]["shared_metrics"]["send"] is False
     with store._connection() as connection:
-        row = connection.execute(
-            "SELECT value FROM telemetry_state WHERE key = ?", (SEND_REVOKED_KEY,)
-        ).fetchone()
-    assert row is not None and row[0] == "1", (
+        open_windows = connection.execute(
+            "SELECT COUNT(*) FROM send_consent_windows WHERE closed_at IS NULL"
+        ).fetchone()[0]
+    assert open_windows == 0, (
         "disabling collection left the send consent window open"
     )
 
diff --git a/tests/hermes_cli/test_shared_metrics_consent_windows.py b/tests/hermes_cli/test_shared_metrics_consent_windows.py
new file mode 100644
index 0000000000..83da1a4749
--- /dev/null
+++ b/tests/hermes_cli/test_shared_metrics_consent_windows.py
@@ -0,0 +1,221 @@
+"""Property tests for the consent-interval model.
+
+Ported from the /tmp validation harness that gated the redesign: every
+scenario here is a defect that actually occurred (rounds 3-5) or a clock
+adversary the day-stamp model could not survive. The v1 and v2 drafts of the
+redesign each FAILED scenarios in this file before shipping — that is the
+harness working, and why these run against the real store and the real
+reconciler rather than a model of them.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from hermes_cli.observability.shared_metrics import SharedMetricsStore
+from hermes_cli.observability.shared_metrics_sender import (
+    CONSENT_GATE_SQL,
+    reconcile_send_consent,
+)
+from hermes_cli.sqlite_util import write_txn
+
+T0 = datetime(2026, 8, 1, tzinfo=timezone.utc)
+
+
+def ts(days=0, hours=0):
+    return (T0 + timedelta(days=days, hours=hours)).isoformat().replace(
+        "+00:00", "Z"
+    )
+
+
+def dt(days=0, hours=0):
+    return T0 + timedelta(days=days, hours=hours)
+
+
+@pytest.fixture
+def store(tmp_path):
+    return SharedMetricsStore(
+        database_path=tmp_path / "m.db", outbox_directory=tmp_path / "o"
+    )
+
+
+def _add(store, pid, start, end):
+    """Store a package the way the generator does: at period end."""
+    with store._connection() as connection:
+        with write_txn(connection):
+            connection.execute(
+                "INSERT INTO package_outbox(package_id, period_start, period_end,"
+                " payload_json, created_at, exported_at) VALUES (?, ?, ?, ?, ?, ?)",
+                (pid, start, end, json.dumps({"package_id": pid}), end, end),
+            )
+            connection.execute(
+                """INSERT INTO consent_marks(name, stamp) VALUES ('data', ?)
+                   ON CONFLICT(name) DO UPDATE SET stamp = MAX(stamp, excluded.stamp)""",
+                (end,),
+            )
+
+
+def _observe(store, send_enabled, when):
+    with store._connection() as connection:
+        with write_txn(connection):
+            reconcile_send_consent(connection, send_enabled, now=when)
+
+
+def _eligible(store):
+    with store._connection() as connection:
+        return sorted(
+            row[0]
+            for row in connection.execute(
+                f"SELECT package_id FROM package_outbox WHERE {CONSENT_GATE_SQL}"
+            )
+        )
+
+
+def _windows(store):
+    with store._connection() as connection:
+        return [
+            tuple(row)
+            for row in connection.execute(
+                "SELECT opened_at, last_confirmed_at, closed_at"
+                " FROM send_consent_windows ORDER BY opened_at"
+            )
+        ]
+
+
+class TestRefusedWindowIsNeverReleased:
+    def test_on_off_on_with_realistic_interleaving(self, store):
+        """Rounds 3 and 5: the refused middle must never transmit, and
+        neither consented era may be lost."""
+        _observe(store, True, dt(0))
+        for n in range(5):
+            _add(store, f"d{n:02d}", ts(days=n), ts(days=n + 1))
+            _observe(store, True, dt(days=n + 1))
+        _observe(store, False, dt(5))
+        for n in range(5, 10):
+            _add(store, f"d{n:02d}", ts(days=n), ts(days=n + 1))
+        _observe(store, True, dt(10))
+        for n in range(10, 15):
+            _add(store, f"d{n:02d}", ts(days=n), ts(days=n + 1))
+            _observe(store, True, dt(days=n + 1))
+
+        eligible = _eligible(store)
+        assert not [p for p in eligible if 5 <= int(p[1:]) < 10], eligible
+        assert [f"d{n:02d}" for n in range(5)] == eligible[:5], (
+            "pre-revocation consented backlog was destroyed"
+        )
+        assert [f"d{n:02d}" for n in range(10, 15)] == eligible[5:], eligible
+
+    def test_hand_edit_with_a_90_day_silent_gap(self, store):
+        """Round 5 D1, strongest form: NOTHING observes the off window.
+
+        The close back-dates to the last confirmed moment, so the unobserved
+        gap is outside every window and fails closed.
+        """
+        _observe(store, True, dt(0))
+        _add(store, "consented", ts(0, 1), ts(0, 2))
+        _observe(store, True, dt(0, 6))
+        for n in range(1, 90, 10):
+            _add(store, f"REFUSED-d{n}", ts(days=n), ts(days=n, hours=1))
+        _observe(store, False, dt(90))   # first observation: boot on day 90
+        _observe(store, True, dt(91))
+        _observe(store, True, dt(92))
+
+        eligible = _eligible(store)
+        assert not [p for p in eligible if p.startswith("REFUSED")], eligible
+        assert "consented" in eligible, (
+            "the confirmed-morning package must survive the reconciliation"
+        )
+
+
+class TestClockAdversaries:
+    def test_rollback_at_re_enable_releases_nothing(self, store):
+        """Round 5 D2: the data mark clamps opens above existing packages."""
+        _observe(store, True, dt(0))
+        _observe(store, True, dt(5))
+        _observe(store, False, dt(5))
+        for n in range(1, 4):
+            _add(store, f"REFUSED-{n}", ts(days=5, hours=n), ts(days=5, hours=n + 1))
+        _observe(store, True, dt(-12))   # 12-day rollback at re-enable
+        _observe(store, True, dt(-11))
+
+        during = [p for p in _eligible(store) if p.startswith("REFUSED")]
+        assert not during, f"rollback released refused packages: {during}"
+
+        _observe(store, True, dt(20))    # clock recovers
+        _observe(store, True, dt(21))
+        after = [p for p in _eligible(store) if p.startswith("REFUSED")]
+        assert not after, f"recovery released refused packages: {after}"
+
+    def test_recovery_does_not_wedge_future_sending(self, store):
+        _observe(store, True, dt(0))
+        _observe(store, False, dt(5))
+        _observe(store, True, dt(-12))
+        _observe(store, True, dt(20))
+        _add(store, "post-recovery", ts(21), ts(21, 4))
+        _observe(store, True, dt(22))
+        assert "post-recovery" in _eligible(store)
+
+
+class TestSubDayGranularity:
+    def test_intra_day_refusal_holds_back_the_whole_day_package(self, store):
+        """Round 5 D3: a day package spanning a refused stretch must wait."""
+        _observe(store, True, dt(0))
+        _observe(store, True, dt(10, 9))
+        _observe(store, False, dt(10, 9))
+        _observe(store, True, dt(10, 18))
+        _observe(store, True, dt(11, 2))
+        _add(store, "halfday", ts(10), ts(11))
+        assert "halfday" not in _eligible(store)
+
+
+class TestReconcilerProperties:
+    def test_idempotent_under_replay(self, store):
+        for _ in range(4):
+            _observe(store, True, dt(0))
+        _observe(store, False, dt(2))
+        for _ in range(5):
+            _observe(store, False, dt(3))
+        _observe(store, True, dt(4))
+        for _ in range(3):
+            _observe(store, True, dt(5))
+        assert len(_windows(store)) == 2
+
+    def test_the_observation_mark_is_monotonic(self, store):
+        """A rolled-back clock must never lower the observation high-water.
+
+        Every downstream guarantee leans on this: closes clamp to it via
+        last_confirmed_at, and opens clamp to max(obs, data). Found as a
+        surviving mutant (obs upsert rewritten from MAX to overwrite) —
+        the leak scenarios happen to be covered by the data mark whenever a
+        leakable package exists, but the property itself must hold on its
+        own, not by coincidence of the sibling mark.
+        """
+        _observe(store, True, dt(5))
+        _observe(store, True, dt(0))   # rollback
+        with store._connection() as connection:
+            stamp = connection.execute(
+                "SELECT stamp FROM consent_marks WHERE name = 'obs'"
+            ).fetchone()[0]
+        assert stamp == ts(5), f"obs mark moved backwards: {stamp}"
+
+    def test_the_gate_is_read_only(self, store):
+        _observe(store, True, dt(0))
+        before = _windows(store)
+        for _ in range(10):
+            _eligible(store)
+        assert _windows(store) == before
+
+    def test_no_window_fails_closed(self, store):
+        _add(store, "orphan", ts(0), ts(1))
+        assert _eligible(store) == []
+
+    def test_fresh_package_waits_one_heartbeat_then_releases(self, store):
+        """The documented latency cost of confirmation-based windows."""
+        _observe(store, True, dt(0))
+        _add(store, "fresh", ts(0, 1), ts(0, 2))
+        assert _eligible(store) == []
+        _observe(store, True, dt(0, 3))
+        assert _eligible(store) == ["fresh"]
diff --git a/tests/hermes_cli/test_shared_metrics_send_wiring.py b/tests/hermes_cli/test_shared_metrics_send_wiring.py
index 3900847955..8ed7724b1e 100644
--- a/tests/hermes_cli/test_shared_metrics_send_wiring.py
+++ b/tests/hermes_cli/test_shared_metrics_send_wiring.py
@@ -209,13 +209,13 @@ class TestInteractivePathIsNotBlocked:
         runtime._join_send_thread(timeout=5)
 
 
-class TestConsentRevocationWindow:
-    """The falling edge must close the window even with no pass running.
+class TestConsentWindows:
+    """Consent reconciliation must work from the relay, in any order.
 
-    Round 3 recorded revocation inside the send loop, which cannot fire for
-    the dominant case: the user turns sending off while idle, so the relay
-    early-returns and no sender is ever built. Re-enabling then released
-    every package collected during the refused window.
+    Round 4's edge detector missed the idle-revocation path; round 5 found it
+    was also dead code whenever collection was off (handles_hook gated it).
+    These tests drive the relay entry points against the single reconciler
+    and assert on the interval table — the only consent state that exists.
     """
 
     def _runtime(self, tmp_path):
@@ -223,20 +223,19 @@ class TestConsentRevocationWindow:
         runtime.subscriber.store = RealBackedStore(tmp_path)
         return runtime
 
-    def _state(self, runtime, key):
+    def _windows(self, runtime):
         with runtime.subscriber.store._connection() as connection:
-            row = connection.execute(
-                "SELECT value FROM telemetry_state WHERE key = ?", (key,)
-            ).fetchone()
-        return row[0] if row else None
+            return [
+                tuple(row)
+                for row in connection.execute(
+                    "SELECT opened_at, last_confirmed_at, closed_at"
+                    " FROM send_consent_windows ORDER BY opened_at"
+                )
+            ]
 
     def test_revoking_while_idle_closes_the_window(
         self, monkeypatch, tmp_path, capture_sender
     ):
-        from hermes_cli.observability.shared_metrics_sender import (
-            SEND_REVOKED_KEY,
-        )
-
         runtime = self._runtime(tmp_path)
 
         _set_config(monkeypatch, _config(enabled=True, send=True))
@@ -247,88 +246,101 @@ class TestConsentRevocationWindow:
         for _ in range(6):
             runtime._send_exported_packages()
 
-        assert self._state(runtime, SEND_REVOKED_KEY) == "1", (
-            "revoking while no pass was running left the consent window open"
+        windows = self._windows(runtime)
+        assert windows and all(w[2] is not None for w in windows), (
+            f"revoking while idle left a window open: {windows}"
         )
 
-    def test_no_spurious_revocation_when_nothing_changes(
+    def test_replayed_observations_create_no_junk_windows(
         self, monkeypatch, tmp_path, capture_sender
     ):
-        """The detector must key on an EDGE, not on every disabled pass.
-
-        A level trigger re-closes a window the user has since REOPENED: each
-        later disabled pass stamps revoked again, so the next enabled pass
-        advances the gate and silently drops packages the user did consent to.
-        Mutation-checked — an earlier version of this test used a
-        never-consented store, where record_revoked no-ops regardless, and so
-        could not tell an edge trigger from a level trigger.
-        """
-        from hermes_cli.observability.shared_metrics_sender import (
-            OPT_IN_PERIOD_KEY,
-            SEND_REVOKED_KEY,
-        )
-
+        """Reconciliation is idempotent — there is no edge to double-count."""
         runtime = self._runtime(tmp_path)
 
         _set_config(monkeypatch, _config(enabled=True, send=True))
-        runtime._send_exported_packages()
-
+        for _ in range(4):
+            runtime._send_exported_packages()
         _set_config(monkeypatch, _config(enabled=True, send=False))
-        runtime._send_exported_packages()
-        assert self._state(runtime, SEND_REVOKED_KEY) == "1"
-
-        # User changes their mind and re-enables.
+        for _ in range(4):
+            runtime._send_exported_packages()
         _set_config(monkeypatch, _config(enabled=True, send=True))
-        runtime._send_exported_packages()
-        assert self._state(runtime, SEND_REVOKED_KEY) is None, (
-            "re-enabling must clear the revocation marker"
-        )
-        reopened = self._state(runtime, OPT_IN_PERIOD_KEY)
-
-        # Further ENABLED passes must not disturb the reopened window.
         for _ in range(4):
             runtime._send_exported_packages()
 
-        assert self._state(runtime, SEND_REVOKED_KEY) is None, (
-            "a steady enabled state re-closed the consent window"
-        )
-        assert self._state(runtime, OPT_IN_PERIOD_KEY) == reopened
+        assert len(self._windows(runtime)) == 2
 
-    def test_a_never_consented_user_is_never_marked_revoked(
+    def test_a_never_consented_user_gets_no_window(
         self, monkeypatch, tmp_path, capture_sender
     ):
-        from hermes_cli.observability.shared_metrics_sender import (
-            SEND_REVOKED_KEY,
-        )
-
         runtime = self._runtime(tmp_path)
         _set_config(monkeypatch, _config(enabled=True, send=False))
         for _ in range(5):
             runtime._send_exported_packages()
 
-        assert self._state(runtime, SEND_REVOKED_KEY) is None
+        assert self._windows(runtime) == []
 
-    def test_re_enabling_after_an_idle_revocation_starts_a_new_window(
+    def test_re_enabling_opens_a_new_window_after_the_refusal(
         self, monkeypatch, tmp_path, capture_sender
     ):
-        from hermes_cli.observability.shared_metrics_sender import (
-            OPT_IN_PERIOD_KEY,
-            SEND_REVOKED_KEY,
-        )
-
+        """The refused gap must fall BETWEEN the two windows."""
         runtime = self._runtime(tmp_path)
         _set_config(monkeypatch, _config(enabled=True, send=True))
         runtime._send_exported_packages()
-        first_window = self._state(runtime, OPT_IN_PERIOD_KEY)
-
         _set_config(monkeypatch, _config(enabled=True, send=False))
         runtime._send_exported_packages()
-        assert self._state(runtime, SEND_REVOKED_KEY) == "1"
-
-        # Re-enabling must not simply resume the original window.
         _set_config(monkeypatch, _config(enabled=True, send=True))
         runtime._send_exported_packages()
-        assert first_window is not None
+
+        windows = self._windows(runtime)
+        assert len(windows) == 2
+        first, second = windows
+        assert first[2] is not None, "first window must be closed"
+        assert second[2] is None, "second window must be open"
+        assert second[0] >= first[2], (
+            f"new window may not overlap the refused gap: {windows}"
+        )
+
+    def test_reconcile_runs_even_when_collection_is_disabled(
+        self, monkeypatch, tmp_path
+    ):
+        """Round-5 D1: enabled:false must not make consent handling dead code.
+
+        The module-level once-per-process reconciler must close the window
+        regardless of handles_hook(). Drives the real observe_lifecycle gate
+        path: handles_hook is False throughout.
+        """
+        from hermes_cli.observability.shared_metrics import SharedMetricsStore
+        from hermes_cli.observability.shared_metrics_sender import (
+            reconcile_send_consent,
+        )
+        from hermes_cli.sqlite_util import write_txn
+
+        store = SharedMetricsStore(
+            database_path=tmp_path / "m.db", outbox_directory=tmp_path / "o"
+        )
+        # A consent window is open from an earlier consented era.
+        with store._connection() as connection:
+            with write_txn(connection):
+                reconcile_send_consent(connection, True)
+
+        monkeypatch.setattr(
+            "hermes_cli.observability.shared_metrics.SharedMetricsStore",
+            lambda *a, **k: store,
+        )
+        _set_config(monkeypatch, _config(enabled=False, send=False))
+        monkeypatch.setattr(mod, "_consent_reconcile_done", False)
+
+        # The full lifecycle entry point, with collection OFF.
+        mod.observe_lifecycle("finish_task")
+
+        with store._connection() as connection:
+            open_windows = connection.execute(
+                "SELECT COUNT(*) FROM send_consent_windows WHERE closed_at IS NULL"
+            ).fetchone()[0]
+        assert open_windows == 0, (
+            "enabled:false made the consent reconciler unreachable (D1)"
+        )
+
 
 
 class TestFailureIsolation:
diff --git a/tests/hermes_cli/test_shared_metrics_sender.py b/tests/hermes_cli/test_shared_metrics_sender.py
index de489a36cb..6b59d5d69d 100644
--- a/tests/hermes_cli/test_shared_metrics_sender.py
+++ b/tests/hermes_cli/test_shared_metrics_sender.py
@@ -19,12 +19,11 @@ from hermes_cli.observability.shared_metrics_sender import (
     MAX_ATTEMPTS,
     MAX_PACKAGES_PER_PASS,
     MAX_SEND_ATTEMPTS,
-    OPT_IN_PERIOD_KEY,
     REQUEST_TIMEOUT_SECONDS,
     SharedMetricsSender,
-    opt_in_period,
-    record_revoked,
+    reconcile_send_consent,
 )
+from hermes_cli.sqlite_util import write_txn
 
 INSTALL_ID = "12a73e97-4de9-4766-830d-9ca1192c0420"
 NOW = datetime(2026, 8, 26, 12, 0, tzinfo=timezone.utc)
@@ -61,10 +60,45 @@ class FakeTransport:
 
 @pytest.fixture
 def store(tmp_path):
-    return SharedMetricsStore(
+    """A store with a broad consent window already open.
+
+    Most tests exercise claiming/retry/transport, not the consent gate, and
+    the interval gate fails closed with no window. One window opened before
+    every test package and confirmed well past NOW keeps those tests about
+    what they are about. Gate tests clear it via _clear_consent.
+    """
+    built = SharedMetricsStore(
         database_path=tmp_path / "metrics.sqlite3",
         outbox_directory=tmp_path / "outbox",
     )
+    _grant_consent(built)
+    return built
+
+
+def _grant_consent(
+    store,
+    opened=datetime(2026, 8, 20, tzinfo=timezone.utc),
+    confirmed_through=datetime(2026, 10, 1, tzinfo=timezone.utc),
+):
+    """Open a consent window and heartbeat it forward, via the real writer."""
+    with store._connection() as connection:
+        with write_txn(connection):
+            reconcile_send_consent(connection, True, now=opened)
+            reconcile_send_consent(connection, True, now=confirmed_through)
+
+
+def _revoke_consent(store, at):
+    with store._connection() as connection:
+        with write_txn(connection):
+            reconcile_send_consent(connection, False, now=at)
+
+
+def _clear_consent(store):
+    """Remove all consent state, for tests of the fail-closed default."""
+    with store._connection() as connection:
+        with write_txn(connection):
+            connection.execute("DELETE FROM send_consent_windows")
+            connection.execute("DELETE FROM consent_marks")
 
 
 def _add_package(store, package_id, period_day, *, exported=True, install_id=INSTALL_ID):
@@ -231,6 +265,9 @@ class TestContractResponses:
 
 class TestConsentGate:
     def test_packages_from_before_opt_in_are_never_sent(self, store):
+        # Consent opens on Aug 24; the "old" package's period predates it.
+        _clear_consent(store)
+        _grant_consent(store, opened=datetime(2026, 8, 24, tzinfo=timezone.utc))
         _add_package(store, "old", "2026-08-20")
         _add_package(store, "new", "2026-08-26")
         transport = FakeTransport(FakeResponse(202))
@@ -245,19 +282,27 @@ class TestConsentGate:
         _sender(store, transport).send_pending()
         assert sorted(b["package_id"] for b in transport.bodies) == ["head", "tail"]
 
-    def test_opt_in_day_is_recorded_once_and_does_not_move(self, store):
+    def test_opt_in_is_immortalised_as_a_window_not_a_day(self, store):
+        """The window survives replayed observations without moving."""
         with store._connection() as connection:
-            first = opt_in_period(connection, now=NOW)
-            later = opt_in_period(connection, now=NOW + timedelta(days=10))
-        assert first == later == "2026-08-26"
-
-    def test_opt_in_day_is_persisted(self, store):
+            rows = connection.execute(
+                "SELECT opened_at, closed_at FROM send_consent_windows"
+            ).fetchall()
+        assert len(rows) == 1 and rows[0][1] is None
+        _grant_consent(store)  # replay: must not create a second window
         with store._connection() as connection:
-            opt_in_period(connection, now=NOW)
-            value = connection.execute(
-                "SELECT value FROM telemetry_state WHERE key = ?", (OPT_IN_PERIOD_KEY,)
+            count = connection.execute(
+                "SELECT COUNT(*) FROM send_consent_windows"
             ).fetchone()[0]
-        assert value == "2026-08-26"
+        assert count == 1
+
+    def test_no_consent_window_means_nothing_is_sent(self, store):
+        """The gate fails closed: absence of a window is absence of consent."""
+        _clear_consent(store)
+        _add_package(store, "pkg-1", "2026-08-26")
+        transport = FakeTransport(FakeResponse(202))
+        _sender(store, transport).send_pending()
+        assert transport.calls == []
 
     def test_unexported_packages_are_skipped(self, store):
         _add_package(store, "pending-export", "2026-08-26", exported=False)
@@ -266,32 +311,31 @@ class TestConsentGate:
         assert transport.calls == []
 
     def test_revoking_then_re_enabling_never_releases_the_off_window(self, store):
-        """Regression: re-opt-in retroactively transmitted the refused window.
+        """The R3/R5 leak: re-opt-in must not release the refused interval.
 
-        opt_in_period was write-once, so packages collected while the user had
-        send: false still had period_start >= the ORIGINAL opt-in day. Turning
-        sending back on released the entire opted-out window — contradicting
-        the documented promise that `send: false` means no further packages
-        leave the machine.
+        Under the interval model the refused days fall BETWEEN two windows;
+        no later observation can place them inside one, so the property holds
+        for any number of on/off cycles — not just the single cycle the old
+        moving day-stamp was patched to survive.
         """
-        _add_package(store, "consented", "2026-08-26")
-        with store._connection() as connection:
-            with __import__(
-                "hermes_cli.sqlite_util", fromlist=["write_txn"]
-            ).write_txn(connection):
-                opt_in_period(connection, now=NOW)
+        _clear_consent(store)
+        _grant_consent(store, opened=NOW - timedelta(days=2), confirmed_through=NOW)
+        _add_package(store, "consented", "2026-08-25")
 
-        # User turns sending off; packages keep being collected.
-        with store._connection() as connection:
-            with __import__(
-                "hermes_cli.sqlite_util", fromlist=["write_txn"]
-            ).write_txn(connection):
-                record_revoked(connection)
+        # User turns sending off; packages keep being collected for 3 days.
+        _revoke_consent(store, at=NOW)
         for day in ("2026-08-27", "2026-08-28", "2026-08-29"):
             _add_package(store, f"refused-{day}", day)
 
-        # User re-enables a few days later.
+        # User re-enables 5 days later; heartbeat confirms past the horizon.
         later = NOW + timedelta(days=5)
+        with store._connection() as connection:
+            with write_txn(connection):
+                reconcile_send_consent(connection, True, now=later)
+                reconcile_send_consent(
+                    connection, True, now=later + timedelta(days=30)
+                )
+
         transport = FakeTransport(*[FakeResponse(202)] * 10)
         SharedMetricsSender(
             store, ENDPOINT, post=transport, sleep=lambda _s: None, now=lambda: later
@@ -301,21 +345,30 @@ class TestConsentGate:
         assert not any("refused" in pid for pid in sent), (
             f"transmitted packages collected while sending was off: {sent}"
         )
+        # And the interval model's improvement over the day-stamp: the
+        # pre-revocation consented package is NOT collateral damage.
+        assert "consented" in sent, (
+            "the consented backlog was destroyed by the revoke/re-enable cycle"
+        )
 
     def test_a_package_from_after_re_enabling_is_sent(self, store):
-        """The revocation fix must not wedge sending off permanently."""
-        with store._connection() as connection:
-            with __import__(
-                "hermes_cli.sqlite_util", fromlist=["write_txn"]
-            ).write_txn(connection):
-                opt_in_period(connection, now=NOW)
-                record_revoked(connection)
+        """The revocation handling must not wedge sending off permanently."""
+        _clear_consent(store)
+        _grant_consent(store, opened=NOW - timedelta(days=2), confirmed_through=NOW)
+        _revoke_consent(store, at=NOW)
 
         later = NOW + timedelta(days=5)
-        _add_package(store, "after-re-optin", later.date().isoformat())
+        with store._connection() as connection:
+            with write_txn(connection):
+                reconcile_send_consent(connection, True, now=later)
+                reconcile_send_consent(
+                    connection, True, now=later + timedelta(days=10)
+                )
+        _add_package(store, "after-re-optin", (later + timedelta(days=1)).date().isoformat())
         transport = FakeTransport(FakeResponse(202))
         SharedMetricsSender(
-            store, ENDPOINT, post=transport, sleep=lambda _s: None, now=lambda: later
+            store, ENDPOINT, post=transport, sleep=lambda _s: None,
+            now=lambda: later + timedelta(days=2),
         ).send_pending()
         assert len(transport.calls) == 1
 
diff --git a/tests/hermes_cli/test_shared_metrics_sender_e2e.py b/tests/hermes_cli/test_shared_metrics_sender_e2e.py
index 552ae93552..9ff8bf9caf 100644
--- a/tests/hermes_cli/test_shared_metrics_sender_e2e.py
+++ b/tests/hermes_cli/test_shared_metrics_sender_e2e.py
@@ -77,10 +77,28 @@ def server():
 
 @pytest.fixture
 def store(tmp_path):
-    return SharedMetricsStore(
+    built = SharedMetricsStore(
         database_path=tmp_path / "metrics.sqlite3",
         outbox_directory=tmp_path / "outbox",
     )
+    # Open a consent window covering the fixture packages; the interval gate
+    # fails closed without one, and this file tests transport, not consent.
+    from datetime import datetime, timezone
+
+    from hermes_cli.observability.shared_metrics_sender import (
+        reconcile_send_consent,
+    )
+    from hermes_cli.sqlite_util import write_txn
+
+    with built._connection() as connection:
+        with write_txn(connection):
+            reconcile_send_consent(
+                connection, True, now=datetime(2026, 8, 20, tzinfo=timezone.utc)
+            )
+            reconcile_send_consent(
+                connection, True, now=datetime(2026, 10, 1, tzinfo=timezone.utc)
+            )
+    return built
 
 
 def _endpoint(server):