From 0beef3d087cd3d9750175f9c35c3b149fc63507c Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 00:57:29 -0700
Subject: [PATCH 01/15] feat(rhythm): rhythm: config block (observe-stage spec)

---
 hueman/config.py            | 162 ++++++++++++++++++++++++++++++++++++
 tests/test_config_rhythm.py |  95 +++++++++++++++++++++
 2 files changed, 257 insertions(+)
 create mode 100644 tests/test_config_rhythm.py

diff --git a/hueman/config.py b/hueman/config.py
index 3783b7d..de9af4a 100644
--- a/hueman/config.py
+++ b/hueman/config.py
@@ -1028,6 +1028,163 @@ def parse(cls, value: Any, ctx: str = "security") -> "SecuritySpec":
         )
 
 
+def _clock_minute(value: Any, ctx: str) -> int:
+    """Parse a strict ``HH:MM`` clock time to minutes after midnight.
+
+    Unlike :func:`parse_time_ref` consumers that accept sun anchors, callers
+    of this helper need a fixed wall-clock minute; ``sunrise``/``sunset`` are
+    rejected with a :class:`ConfigError` naming ``ctx``.
+    """
+    ref = parse_time_ref(value, ctx=ctx)
+    if ref in ("sunrise", "sunset"):
+        raise ConfigError(f"{ctx} must be a clock time like '22:45'")
+    hh, mm = ref.split(":")
+    return int(hh) * 60 + int(mm)
+
+
+def _dur_min(value: Any, ctx: str) -> int:
+    """Parse a duration (``"90m"``, ``"2h"``) to whole minutes."""
+    return parse_duration(value, ctx=ctx) // 60_000
+
+
+@dataclass(frozen=True)
+class RhythmSignals:
+    """External phone-derived signal files for the rhythm engine.
+
+    Both are optional; the engine degrades to learned/default anchors when a
+    file is unset or absent. Files live on a shared mount written by an
+    external automation (e.g. Home Assistant), mirroring the TV bias
+    control-file channel.
+
+    Attributes:
+        next_alarm_file: Path to a file whose content is the epoch-seconds
+            timestamp of the phone's next alarm (empty/``0`` = no alarm set).
+            Re-read every tick; never consumed.
+        charging_file: Path whose *existence* means the phone is charging.
+            Never consumed.
+    """
+
+    next_alarm_file: str | None = None
+    charging_file: str | None = None
+
+    @classmethod
+    def parse(cls, value: Any, ctx: str) -> "RhythmSignals":
+        """Parse the ``rhythm.signals`` block (both paths optional)."""
+        d = _as_dict(value, ctx)
+        alarm = d.get("next_alarm_file")
+        charging = d.get("charging_file")
+        return cls(
+            next_alarm_file=str(alarm) if alarm else None,
+            charging_file=str(charging) if charging else None,
+        )
+
+
+@dataclass(frozen=True)
+class RhythmPresence:
+    """Presence-inference tunables (pet discounting and wake confirmation).
+
+    Attributes:
+        quiet_min: House-wide human quiet (minutes) required before the
+            sleep-onset vote can pass.
+        wake_confirm_events: Motion events within the confirm window needed
+            to call a wake "sustained" (a single pet blip cannot fake it).
+        wake_confirm_window_min: The sliding window (minutes) for the two
+            attributes above and for ``recent_rooms``.
+        pet_progression_min: Motion in one room counts as human when a
+            *different* room was active within this many minutes
+            (room-to-room progression); solo single-room motion is
+            discounted as a pet.
+    """
+
+    quiet_min: int = 30
+    wake_confirm_events: int = 3
+    wake_confirm_window_min: int = 10
+    pet_progression_min: int = 5
+
+    @classmethod
+    def parse(cls, value: Any, ctx: str) -> "RhythmPresence":
+        """Parse the ``rhythm.presence`` block (all fields defaulted)."""
+        d = _as_dict(value, ctx)
+        return cls(
+            quiet_min=_dur_min(d.get("quiet", "30m"), f"{ctx}.quiet"),
+            wake_confirm_events=int(d.get("wake_confirm_events", 3)),
+            wake_confirm_window_min=_dur_min(
+                d.get("wake_confirm_window", "10m"), f"{ctx}.wake_confirm_window"),
+            pet_progression_min=_dur_min(
+                d.get("pet_progression", "5m"), f"{ctx}.pet_progression"),
+        )
+
+
+@dataclass(frozen=True)
+class RhythmSpec:
+    """The rhythm engine: closed-loop day-phase inference (and, in later
+    stages, shepherding actuation).
+
+    Stage 1 ships ``stage: "observe"`` only — the engine infers and logs but
+    never writes to the bridge. See the design spec (ops repo,
+    ``docs/superpowers/specs/2026-07-04-rhythm-engine-design.md``).
+
+    Attributes:
+        stage: Rollout stage; only ``"observe"`` is implemented (parse accepts
+            the future ``"mornings"``/``"full"`` so a config can be staged, but
+            the daemon refuses to start them until those stages ship).
+        bedroom: Room name whose motion area anchors sleep/wake inference.
+        bed_target_min: Chosen bed-time target, minutes after midnight.
+        wake_default_min: Fallback wake anchor when no alarm and no history.
+        weekend_drift_cap_min: Max minutes weekend anchors may lag weekday's.
+        wind_down_lead_min: Wind-down phase starts this long before bed anchor.
+        dawn_lead_min: Dawn phase starts this long before the wake anchor.
+        dawn_max_advance_min: Cap on snooze-through compensation (unused in
+            observe; parsed now so staged configs validate).
+        morning_min: Duration of the ``morning`` phase after wake.
+        state_file: JSON file for learned anchors + the live snapshot.
+        signals: External phone signal files.
+        presence: Presence-inference tunables.
+    """
+
+    bedroom: str
+    stage: str = "observe"
+    bed_target_min: int = 23 * 60
+    wake_default_min: int = 7 * 60
+    weekend_drift_cap_min: int = 90
+    wind_down_lead_min: int = 90
+    dawn_lead_min: int = 25
+    dawn_max_advance_min: int = 20
+    morning_min: int = 60
+    state_file: str = "rhythm-state.json"
+    signals: RhythmSignals = RhythmSignals()
+    presence: RhythmPresence = RhythmPresence()
+
+    @classmethod
+    def parse(cls, value: Any, ctx: str = "rhythm") -> "RhythmSpec":
+        """Parse the ``rhythm:`` block. Only ``bedroom`` is required."""
+        d = _as_dict(value, ctx)
+        stage = str(d.get("stage", "observe"))
+        if stage not in ("observe", "mornings", "full"):
+            raise ConfigError(
+                f"{ctx}.stage must be 'observe', 'mornings' or 'full', got {stage!r}")
+        bedroom = d.get("bedroom")
+        if not bedroom or not isinstance(bedroom, str):
+            raise ConfigError(f"{ctx}.bedroom is required (the room name whose "
+                              "motion area anchors sleep/wake inference)")
+        return cls(
+            bedroom=bedroom,
+            stage=stage,
+            bed_target_min=_clock_minute(d.get("bed_target", "23:00"), f"{ctx}.bed_target"),
+            wake_default_min=_clock_minute(d.get("wake_default", "07:00"), f"{ctx}.wake_default"),
+            weekend_drift_cap_min=_dur_min(d.get("weekend_drift_cap", "90m"),
+                                           f"{ctx}.weekend_drift_cap"),
+            wind_down_lead_min=_dur_min(d.get("wind_down_lead", "90m"), f"{ctx}.wind_down_lead"),
+            dawn_lead_min=_dur_min(d.get("dawn_lead", "25m"), f"{ctx}.dawn_lead"),
+            dawn_max_advance_min=_dur_min(d.get("dawn_max_advance", "20m"),
+                                          f"{ctx}.dawn_max_advance"),
+            morning_min=_dur_min(d.get("morning", "60m"), f"{ctx}.morning"),
+            state_file=str(d.get("state_file", "rhythm-state.json")),
+            signals=RhythmSignals.parse(d.get("signals", {}), f"{ctx}.signals"),
+            presence=RhythmPresence.parse(d.get("presence", {}), f"{ctx}.presence"),
+        )
+
+
 @dataclass(frozen=True)
 class Config:
     """The fully validated desired state — the root of the config tree.
@@ -1053,6 +1210,7 @@ class Config:
     circadian_scene: CircadianSceneSpec | None = None
     circadian_daemon: CircadianDaemonSpec | None = None
     security: SecuritySpec | None = None
+    rhythm: RhythmSpec | None = None
     require_all_lights_assigned: bool = True
     marker: str = DEFAULT_MARKER
 
@@ -1101,6 +1259,9 @@ def parse(cls, doc: dict[str, Any]) -> "Config":
         security = (
             SecuritySpec.parse(doc["security"]) if doc.get("security") else None
         )
+        rhythm = (
+            RhythmSpec.parse(doc["rhythm"], "rhythm") if "rhythm" in doc else None
+        )
         return cls(
             bridge=bridge,
             location=location,
@@ -1113,6 +1274,7 @@ def parse(cls, doc: dict[str, Any]) -> "Config":
             circadian_scene=circadian_scene,
             circadian_daemon=circadian_daemon,
             security=security,
+            rhythm=rhythm,
             require_all_lights_assigned=bool(doc.get("require_all_lights_assigned", True)),
             marker=str(doc.get("marker", DEFAULT_MARKER)),
         )
diff --git a/tests/test_config_rhythm.py b/tests/test_config_rhythm.py
new file mode 100644
index 0000000..62036ca
--- /dev/null
+++ b/tests/test_config_rhythm.py
@@ -0,0 +1,95 @@
+"""Tests for the ``rhythm:`` config block."""
+import pytest
+
+from hueman.config import Config, RhythmSpec
+from hueman.errors import ConfigError
+
+BASE = {
+    "bridge": {"host": "bridge.local"},
+    "location": {"lat": 40.0, "lon": -75.0, "tz": "America/New_York"},
+    "circadian": {"day": {"brightness": 90, "kelvin": 5000},
+                  "evening": {"brightness": 40, "kelvin": 2700},
+                  "night": {"brightness": 15, "kelvin": 2200}},
+}
+
+
+def _cfg(rhythm):
+    doc = dict(BASE)
+    doc["rhythm"] = rhythm
+    return Config.parse(doc)
+
+
+def test_rhythm_absent_is_none():
+    assert Config.parse(dict(BASE)).rhythm is None
+
+
+def test_rhythm_minimal_defaults():
+    spec = _cfg({"bedroom": "Bedroom"}).rhythm
+    assert spec is not None
+    assert spec.stage == "observe"
+    assert spec.bedroom == "Bedroom"
+    assert spec.bed_target_min == 23 * 60
+    assert spec.wake_default_min == 7 * 60
+    assert spec.weekend_drift_cap_min == 90
+    assert spec.wind_down_lead_min == 90
+    assert spec.dawn_lead_min == 25
+    assert spec.morning_min == 60
+    assert spec.state_file == "rhythm-state.json"
+    assert spec.signals.next_alarm_file is None
+    assert spec.presence.quiet_min == 30
+    assert spec.presence.wake_confirm_events == 3
+    assert spec.presence.wake_confirm_window_min == 10
+    assert spec.presence.pet_progression_min == 5
+
+
+def test_rhythm_full_block():
+    spec = _cfg({
+        "stage": "observe",
+        "bedroom": "Bedroom",
+        "bed_target": "22:45",
+        "wake_default": "06:30",
+        "weekend_drift_cap": "2h",
+        "wind_down_lead": "75m",
+        "dawn_lead": "20m",
+        "dawn_max_advance": "15m",
+        "morning": "45m",
+        "state_file": "/data/rhythm.json",
+        "signals": {
+            "next_alarm_file": "/mnt/sig/.next-alarm",
+            "charging_file": "/mnt/sig/.phone-charging",
+        },
+        "presence": {
+            "quiet": "20m",
+            "wake_confirm_events": 4,
+            "wake_confirm_window": "8m",
+            "pet_progression": "3m",
+        },
+    }).rhythm
+    assert spec.bed_target_min == 22 * 60 + 45
+    assert spec.wake_default_min == 6 * 60 + 30
+    assert spec.weekend_drift_cap_min == 120
+    assert spec.wind_down_lead_min == 75
+    assert spec.dawn_lead_min == 20
+    assert spec.dawn_max_advance_min == 15
+    assert spec.morning_min == 45
+    assert spec.state_file == "/data/rhythm.json"
+    assert spec.signals.next_alarm_file == "/mnt/sig/.next-alarm"
+    assert spec.presence.quiet_min == 20
+    assert spec.presence.wake_confirm_events == 4
+    assert spec.presence.wake_confirm_window_min == 8
+    assert spec.presence.pet_progression_min == 3
+
+
+def test_rhythm_requires_bedroom():
+    with pytest.raises(ConfigError, match="rhythm.bedroom"):
+        _cfg({})
+
+
+def test_rhythm_rejects_unknown_stage():
+    with pytest.raises(ConfigError, match="rhythm.stage"):
+        _cfg({"bedroom": "Bedroom", "stage": "shepherd"})
+
+
+def test_rhythm_rejects_sun_anchored_bed_target():
+    with pytest.raises(ConfigError, match="rhythm.bed_target"):
+        _cfg({"bedroom": "Bedroom", "bed_target": "sunset"})

From 266b330a3da3afe3e3e8199832d519b445cfa9a1 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 01:02:59 -0700
Subject: [PATCH 02/15] refactor(config): CircadianDaemonSpec.parse uses
 _clock_minute

---
 hueman/config.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/hueman/config.py b/hueman/config.py
index de9af4a..924d63b 100644
--- a/hueman/config.py
+++ b/hueman/config.py
@@ -855,16 +855,12 @@ def parse(cls, value: Any, ctx: str = "circadian_daemon") -> "CircadianDaemonSpe
         mo = _as_dict(d.get("manual_override"), f"{ctx}.manual_override")
         retry = _as_dict(d.get("retry"), f"{ctx}.retry")
         log = _as_dict(d.get("log"), f"{ctx}.log")
-        hand_off = parse_time_ref(d.get("hand_off", "22:34"), ctx=f"{ctx}.hand_off")
-        if hand_off in ("sunrise", "sunset"):
-            raise ConfigError(f"{ctx}.hand_off must be a clock time like '22:34'")
-        hh, mm = hand_off.split(":")
         floor = d.get("brightness_floor")
         ceil = d.get("brightness_ceiling")
         return cls(
             zone=str(_require(d, "zone", ctx)),
             start=parse_anchor(d.get("start", "sunrise"), ctx=f"{ctx}.start"),
-            hand_off_min=int(hh) * 60 + int(mm),
+            hand_off_min=_clock_minute(d.get("hand_off", "22:34"), f"{ctx}.hand_off"),
             interval_ms=parse_duration(d.get("interval", "60s"), ctx=f"{ctx}.interval"),
             transition_ms=parse_duration(d.get("transition", "75s"), ctx=f"{ctx}.transition"),
             fade_off_ms=parse_duration(d.get("fade_off", "90s"), ctx=f"{ctx}.fade_off"),

From 3b1d59762063ace22ff30f05d79f0822c6664ce0 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 01:06:46 -0700
Subject: [PATCH 03/15] feat(rhythm): presence tracker with pet discounting and
 evidence rules

---
 hueman/presence.py     | 147 +++++++++++++++++++++++++++++++++++++++++
 tests/test_presence.py |  69 +++++++++++++++++++
 2 files changed, 216 insertions(+)
 create mode 100644 hueman/presence.py
 create mode 100644 tests/test_presence.py

diff --git a/hueman/presence.py b/hueman/presence.py
new file mode 100644
index 0000000..c69d349
--- /dev/null
+++ b/hueman/presence.py
@@ -0,0 +1,147 @@
+"""Per-room activity judging for the rhythm engine (pet discounting).
+
+The bridge's MotionAware areas report *presence*, not *identity*; in a
+one-human-plus-pets household the discriminator is movement *pattern*:
+
+* a light being changed is always a human (pets do not use switches);
+* motion is human when a *different* room was active within the progression
+  window (room-to-room movement) or a light change happened nearby;
+* solo single-room motion is discounted as a pet.
+
+Known blind spot, accepted by design: a human sitting nearly still in one
+room for a long time degrades to "pet" judgments — but the sleep-onset vote
+that consumes this signal also requires lights-out and TV-off, so a reader
+with the lamp on is never mistaken for an empty house.
+
+Every judgment carries the rule that produced it so the daemon can log an
+evidence trail (see the design spec's "explainable and measured" rule).
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from dataclasses import dataclass
+
+from .config import RhythmPresence
+
+
+@dataclass(frozen=True)
+class ActivityEvent:
+    """One activity observation.
+
+    Attributes:
+        room: Room name; ``""`` when unattributable (e.g. a manual override
+            detected on the driven zone).
+        kind: ``"motion"`` (a motion-area fired) or ``"light_change"`` (a
+            human manipulated lighting).
+        ts: Epoch seconds.
+    """
+
+    room: str
+    kind: str
+    ts: float
+
+
+@dataclass(frozen=True)
+class Judgment:
+    """The tracker's verdict on one event, with the rule that decided it.
+
+    Attributes:
+        event: The judged event.
+        human: Whether the event counts as human activity.
+        rule: ``"light-change"``, ``"progression"``, ``"light-change-context"``,
+            or ``"solo-motion"`` (the discount).
+    """
+
+    event: ActivityEvent
+    human: bool
+    rule: str
+
+
+@dataclass(frozen=True)
+class PresenceSummary:
+    """Aggregated view the rhythm engine consumes each tick.
+
+    Attributes:
+        quiet_s: Seconds since the last *human-judged* activity (pet blips do
+            not reset it). Very large when nothing human was ever seen.
+        last_active_room: Room of the most recent human activity, if any.
+        recent_rooms: Distinct rooms with motion inside the confirm window.
+        recent_motion_count: Motion events inside the confirm window
+            (human- and pet-judged alike; wake confirmation applies its own
+            bedroom/light-change requirement on top).
+        recent_light_change: Whether a light change occurred in the window.
+    """
+
+    quiet_s: float
+    last_active_room: str | None
+    recent_rooms: tuple[str, ...]
+    recent_motion_count: int
+    recent_light_change: bool
+
+
+class PresenceTracker:
+    """Judges activity events and answers "how quiet has the house been?".
+
+    Args:
+        spec: Presence tunables (windows and thresholds).
+    """
+
+    #: Retain at most this many recent events; windows are minutes long, so
+    #: this is generous while keeping memory flat over months of uptime.
+    _MAX_EVENTS = 512
+
+    def __init__(self, spec: RhythmPresence) -> None:
+        """Start with an empty event history and no human activity seen."""
+        self._spec = spec
+        self._events: deque[ActivityEvent] = deque(maxlen=self._MAX_EVENTS)
+        self._last_human_ts: float | None = None
+        self._last_human_room: str | None = None
+
+    def feed(self, event: ActivityEvent) -> Judgment:
+        """Judge one event, record it, and return the verdict with its rule."""
+        judgment = self._judge(event)
+        self._events.append(event)
+        if judgment.human:
+            self._last_human_ts = event.ts
+            if event.room:
+                self._last_human_room = event.room
+        return judgment
+
+    def _judge(self, event: ActivityEvent) -> Judgment:
+        """Apply the pet-discounting rules (see the module docstring)."""
+        if event.kind == "light_change":
+            return Judgment(event, human=True, rule="light-change")
+        window_start = event.ts - self._spec.pet_progression_min * 60.0
+        for prior in reversed(self._events):
+            if prior.ts < window_start:
+                break
+            if prior.kind == "light_change":
+                return Judgment(event, human=True, rule="light-change-context")
+            if prior.kind == "motion" and prior.room and prior.room != event.room:
+                return Judgment(event, human=True, rule="progression")
+        return Judgment(event, human=False, rule="solo-motion")
+
+    def summary(self, now: float) -> PresenceSummary:
+        """Aggregate the confirm-window state for the engine's tick."""
+        window_start = now - self._spec.wake_confirm_window_min * 60.0
+        rooms: list[str] = []
+        motion_count = 0
+        light_change = False
+        for ev in self._events:
+            if ev.ts < window_start:
+                continue
+            if ev.kind == "motion":
+                motion_count += 1
+                if ev.room and ev.room not in rooms:
+                    rooms.append(ev.room)
+            elif ev.kind == "light_change":
+                light_change = True
+        quiet_s = (now - self._last_human_ts) if self._last_human_ts is not None else 1e9
+        return PresenceSummary(
+            quiet_s=quiet_s,
+            last_active_room=self._last_human_room,
+            recent_rooms=tuple(rooms),
+            recent_motion_count=motion_count,
+            recent_light_change=light_change,
+        )
diff --git a/tests/test_presence.py b/tests/test_presence.py
new file mode 100644
index 0000000..1eca295
--- /dev/null
+++ b/tests/test_presence.py
@@ -0,0 +1,69 @@
+"""Tests for presence judging (pet discounting, quiet tracking)."""
+from hueman.config import RhythmPresence
+from hueman.presence import ActivityEvent, PresenceTracker
+
+SPEC = RhythmPresence()  # quiet 30m, confirm 3 in 10m, progression 5m
+T0 = 1_000_000.0
+
+
+def _motion(room, ts):
+    return ActivityEvent(room=room, kind="motion", ts=ts)
+
+
+def test_solo_single_room_motion_is_discounted_as_pet():
+    tr = PresenceTracker(SPEC)
+    j = tr.feed(_motion("Kitchen", T0))
+    assert j.human is False and j.rule == "solo-motion"
+    assert tr.summary(T0 + 60).quiet_s >= 60  # did not reset human-quiet
+
+
+def test_room_progression_is_human():
+    tr = PresenceTracker(SPEC)
+    tr.feed(_motion("Kitchen", T0))
+    j = tr.feed(_motion("Hallway", T0 + 120))  # different room within 5m
+    assert j.human is True and j.rule == "progression"
+    assert tr.summary(T0 + 130).quiet_s == 10.0
+    assert tr.summary(T0 + 130).last_active_room == "Hallway"
+
+
+def test_progression_window_expires():
+    tr = PresenceTracker(SPEC)
+    tr.feed(_motion("Kitchen", T0))
+    j = tr.feed(_motion("Hallway", T0 + 6 * 60))  # beyond 5m window
+    assert j.human is False and j.rule == "solo-motion"
+
+
+def test_light_change_is_always_human():
+    tr = PresenceTracker(SPEC)
+    j = tr.feed(ActivityEvent(room="", kind="light_change", ts=T0))
+    assert j.human is True and j.rule == "light-change"
+    assert tr.summary(T0 + 5).quiet_s == 5.0
+
+
+def test_motion_near_light_change_is_human():
+    tr = PresenceTracker(SPEC)
+    tr.feed(ActivityEvent(room="", kind="light_change", ts=T0))
+    j = tr.feed(_motion("Kitchen", T0 + 60))
+    assert j.human is True and j.rule == "light-change-context"
+
+
+def test_summary_windows():
+    tr = PresenceTracker(SPEC)
+    tr.feed(_motion("Bedroom", T0))
+    tr.feed(_motion("Hallway", T0 + 30))
+    tr.feed(_motion("Kitchen", T0 + 60))
+    s = tr.summary(T0 + 90)
+    assert s.recent_motion_count == 3
+    assert set(s.recent_rooms) == {"Bedroom", "Hallway", "Kitchen"}
+    assert s.recent_light_change is False
+    # 11 minutes later the confirm window (10m) has emptied
+    s2 = tr.summary(T0 + 60 + 11 * 60)
+    assert s2.recent_motion_count == 0 and s2.recent_rooms == ()
+
+
+def test_quiet_is_since_last_human_not_last_pet():
+    tr = PresenceTracker(SPEC)
+    tr.feed(_motion("Bedroom", T0))
+    tr.feed(_motion("Hallway", T0 + 10))          # human via progression
+    tr.feed(_motion("Kitchen", T0 + 20 * 60))      # solo — pet
+    assert tr.summary(T0 + 21 * 60).quiet_s == 21 * 60 - 10

From 3c7d27baa5cb3ae30a639f5c624f0203b6e414c1 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 01:12:36 -0700
Subject: [PATCH 04/15] fix(rhythm): progression rule requires both rooms
 attributed

---
 hueman/presence.py     | 2 +-
 tests/test_presence.py | 8 ++++++++
 2 files changed, 9 insertions(+), 1 deletion(-)

diff --git a/hueman/presence.py b/hueman/presence.py
index c69d349..f556754 100644
--- a/hueman/presence.py
+++ b/hueman/presence.py
@@ -118,7 +118,7 @@ def _judge(self, event: ActivityEvent) -> Judgment:
                 break
             if prior.kind == "light_change":
                 return Judgment(event, human=True, rule="light-change-context")
-            if prior.kind == "motion" and prior.room and prior.room != event.room:
+            if event.room and prior.kind == "motion" and prior.room and prior.room != event.room:
                 return Judgment(event, human=True, rule="progression")
         return Judgment(event, human=False, rule="solo-motion")
 
diff --git a/tests/test_presence.py b/tests/test_presence.py
index 1eca295..f16932b 100644
--- a/tests/test_presence.py
+++ b/tests/test_presence.py
@@ -61,6 +61,14 @@ def test_summary_windows():
     assert s2.recent_motion_count == 0 and s2.recent_rooms == ()
 
 
+def test_roomless_motion_cannot_be_progression():
+    """Motion with no room attribution never proves room-to-room movement."""
+    tr = PresenceTracker(SPEC)
+    tr.feed(_motion("Kitchen", T0))
+    j = tr.feed(_motion("", T0 + 60))
+    assert j.human is False and j.rule == "solo-motion"
+
+
 def test_quiet_is_since_last_human_not_last_pet():
     tr = PresenceTracker(SPEC)
     tr.feed(_motion("Bedroom", T0))

From d1234cda9b5975ca114a92445a6546969c8a25f0 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 01:15:43 -0700
Subject: [PATCH 05/15] feat(rhythm): learned anchor store (bounded, JSON
 round-trip)

---
 hueman/rhythm_control.py     | 80 ++++++++++++++++++++++++++++++++++++
 tests/test_rhythm_anchors.py | 43 +++++++++++++++++++
 2 files changed, 123 insertions(+)
 create mode 100644 hueman/rhythm_control.py
 create mode 100644 tests/test_rhythm_anchors.py

diff --git a/hueman/rhythm_control.py b/hueman/rhythm_control.py
new file mode 100644
index 0000000..35633bf
--- /dev/null
+++ b/hueman/rhythm_control.py
@@ -0,0 +1,80 @@
+"""Day-phase inference for the rhythm engine (pure logic, no I/O).
+
+Stage 1 (observe): given presence summaries and external signals, decide the
+current day phase, detect sleep onset and wake, and maintain *learned anchors*
+(recent observed wake / sleep-onset minutes per weekday/weekend class). The
+engine never performs I/O; the daemon supplies inputs and persists the store.
+
+Every decision carries a reason and an evidence mapping so the daemon can log
+an auditable trail — unexplainable automation in a home is indistinguishable
+from malfunction.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+from dataclasses import dataclass, field
+from statistics import median
+from typing import Any
+from zoneinfo import ZoneInfo
+
+from .config import RhythmSpec
+from .presence import PresenceSummary
+
+#: Newest observations kept per (kind, day-class); two weeks of weekdays.
+_MAX_SAMPLES = 14
+
+
+class AnchorStore:
+    """Recent observed wake / sleep-onset minutes, per weekday/weekend class.
+
+    One sample per calendar date (a re-observation replaces the earlier one);
+    at most 14 newest per (kind, class). Values are minutes after midnight.
+    Serialisable to a small JSON document the daemon persists; a corrupted or
+    missing document silently yields an empty store (learning restarts, the
+    engine falls back to config defaults — see the safety rails in the spec).
+    """
+
+    def __init__(self) -> None:
+        """Start empty; populate via :meth:`record` or :meth:`from_json`."""
+        # {day_class: {kind: [{"date": iso, "minute": int}, ...]}}
+        self._samples: dict[str, dict[str, list[dict[str, Any]]]] = {}
+
+    def record(self, kind: str, day_class: str, minute: int, date_iso: str) -> None:
+        """Record one observation, replacing the same date, keeping newest 14."""
+        bucket = self._samples.setdefault(day_class, {}).setdefault(kind, [])
+        bucket[:] = [s for s in bucket if s["date"] != date_iso]
+        bucket.append({"date": date_iso, "minute": int(minute)})
+        bucket.sort(key=lambda s: str(s["date"]))
+        del bucket[:-_MAX_SAMPLES]
+
+    def median(self, kind: str, day_class: str) -> int | None:
+        """Median observed minute for the pair, or ``None`` with no samples."""
+        bucket = self._samples.get(day_class, {}).get(kind, [])
+        if not bucket:
+            return None
+        return int(median(s["minute"] for s in bucket))
+
+    def to_json(self) -> dict[str, Any]:
+        """Serialise to the persisted document shape."""
+        return {"anchors": self._samples}
+
+    @classmethod
+    def from_json(cls, doc: dict[str, Any]) -> AnchorStore:
+        """Rebuild from :meth:`to_json` output; malformed input yields empty."""
+        store = cls()
+        anchors = doc.get("anchors")
+        if not isinstance(anchors, dict):
+            return store
+        for day_class, kinds in anchors.items():
+            if not isinstance(kinds, dict):
+                continue
+            for kind, samples in kinds.items():
+                if not isinstance(samples, list):
+                    continue
+                for s in samples:
+                    if (isinstance(s, dict)
+                            and isinstance(s.get("date"), str)
+                            and isinstance(s.get("minute"), int)):
+                        store.record(str(kind), str(day_class), s["minute"], s["date"])
+        return store
diff --git a/tests/test_rhythm_anchors.py b/tests/test_rhythm_anchors.py
new file mode 100644
index 0000000..9c748e4
--- /dev/null
+++ b/tests/test_rhythm_anchors.py
@@ -0,0 +1,43 @@
+"""Tests for the learned-anchor store."""
+from hueman.rhythm_control import AnchorStore
+
+
+def test_empty_store_has_no_median():
+    assert AnchorStore().median("wake", "weekday") is None
+
+
+def test_record_and_median():
+    st = AnchorStore()
+    for i, m in enumerate([420, 430, 440]):
+        st.record("wake", "weekday", m, f"2026-07-0{i + 1}")
+    assert st.median("wake", "weekday") == 430
+
+
+def test_one_sample_per_date_latest_wins():
+    st = AnchorStore()
+    st.record("wake", "weekday", 400, "2026-07-01")
+    st.record("wake", "weekday", 460, "2026-07-01")  # re-observation same day
+    assert st.median("wake", "weekday") == 460
+
+
+def test_capped_at_14_newest_kept():
+    st = AnchorStore()
+    for day in range(1, 21):  # 20 samples
+        st.record("wake", "weekday", 400 + day, f"2026-06-{day:02d}")
+    doc = st.to_json()
+    samples = doc["anchors"]["weekday"]["wake"]
+    assert len(samples) == 14
+    assert samples[0]["date"] == "2026-06-07"  # oldest surviving
+
+
+def test_json_roundtrip():
+    st = AnchorStore()
+    st.record("sleep_onset", "weekend", 1400, "2026-07-04")
+    st2 = AnchorStore.from_json(st.to_json())
+    assert st2.median("sleep_onset", "weekend") == 1400
+
+
+def test_from_json_tolerates_garbage():
+    assert AnchorStore.from_json({}).median("wake", "weekday") is None
+    assert AnchorStore.from_json({"anchors": {"weekday": {"wake": "nope"}}}
+                                 ).median("wake", "weekday") is None

From 24b54d881194ccfdd7d656ac5508214734e8fb0b Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 01:21:36 -0700
Subject: [PATCH 06/15] fix(rhythm): anchor median rounds instead of truncating

---
 hueman/rhythm_control.py     | 8 ++++++--
 tests/test_rhythm_anchors.py | 7 +++++++
 2 files changed, 13 insertions(+), 2 deletions(-)

diff --git a/hueman/rhythm_control.py b/hueman/rhythm_control.py
index 35633bf..4c504ad 100644
--- a/hueman/rhythm_control.py
+++ b/hueman/rhythm_control.py
@@ -49,11 +49,15 @@ def record(self, kind: str, day_class: str, minute: int, date_iso: str) -> None:
         del bucket[:-_MAX_SAMPLES]
 
     def median(self, kind: str, day_class: str) -> int | None:
-        """Median observed minute for the pair, or ``None`` with no samples."""
+        """Median observed minute for the pair, or ``None`` with no samples.
+
+        The midpoint of an even sample count rounds to the nearest minute.
+        """
         bucket = self._samples.get(day_class, {}).get(kind, [])
         if not bucket:
             return None
-        return int(median(s["minute"] for s in bucket))
+        minutes: list[int] = [int(s["minute"]) for s in bucket]
+        return round(median(minutes))
 
     def to_json(self) -> dict[str, Any]:
         """Serialise to the persisted document shape."""
diff --git a/tests/test_rhythm_anchors.py b/tests/test_rhythm_anchors.py
index 9c748e4..00dc843 100644
--- a/tests/test_rhythm_anchors.py
+++ b/tests/test_rhythm_anchors.py
@@ -30,6 +30,13 @@ def test_capped_at_14_newest_kept():
     assert samples[0]["date"] == "2026-06-07"  # oldest surviving
 
 
+def test_median_of_even_count_rounds_to_nearest_minute():
+    st = AnchorStore()
+    st.record("wake", "weekday", 420, "2026-07-01")
+    st.record("wake", "weekday", 425, "2026-07-02")
+    assert st.median("wake", "weekday") == 422  # 422.5 rounds half-to-even
+
+
 def test_json_roundtrip():
     st = AnchorStore()
     st.record("sleep_onset", "weekend", 1400, "2026-07-04")

From 7e7776499a2f771194dc810d7bba48e36c338783 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 01:35:36 -0700
Subject: [PATCH 07/15] feat(rhythm): day-phase engine with sleep/wake
 inference and evidence

---
 hueman/rhythm_control.py    | 297 ++++++++++++++++++++++++++++++++++++
 tests/test_rhythm_engine.py | 138 +++++++++++++++++
 2 files changed, 435 insertions(+)
 create mode 100644 tests/test_rhythm_engine.py

diff --git a/hueman/rhythm_control.py b/hueman/rhythm_control.py
index 4c504ad..12df01d 100644
--- a/hueman/rhythm_control.py
+++ b/hueman/rhythm_control.py
@@ -82,3 +82,300 @@ def from_json(cls, doc: dict[str, Any]) -> AnchorStore:
                             and isinstance(s.get("minute"), int)):
                         store.record(str(kind), str(day_class), s["minute"], s["date"])
         return store
+
+
+@dataclass(frozen=True)
+class SignalState:
+    """External signals the daemon hands the engine on each tick.
+
+    Attributes:
+        next_alarm_epoch: Epoch seconds of the phone's next alarm; ``None``
+            when unknown/none set.
+        phone_charging: Whether the phone is on the charger (bedtime proxy).
+        tv_on: Committed TV state from the bias aggregator (``False`` when
+            bias is not configured).
+        zone_on: Whether the daemon last commanded its driven zone on — a
+            lights-out proxy for the sleep vote.
+        sunset_min: Today's sunset as minutes after local midnight (evening
+            phase anchor); the daemon computes it from the solar calculator.
+    """
+
+    next_alarm_epoch: float | None
+    phone_charging: bool
+    tv_on: bool
+    zone_on: bool
+    sunset_min: float
+
+
+@dataclass(frozen=True)
+class PhaseDecision:
+    """One tick's verdict: the phase, whether it changed, and the evidence.
+
+    Attributes:
+        phase: The (possibly unchanged) current phase constant.
+        changed: True when this tick moved to a new phase.
+        reason: Short machine-greppable cause, e.g. ``"sleep-vote"``.
+        evidence: JSON-serialisable inputs behind the decision (anchors,
+            presence numbers, signal states) — the log's audit trail.
+    """
+
+    phase: str
+    changed: bool
+    reason: str
+    evidence: dict[str, Any]
+
+
+def _noon_shifted(minute: float) -> float:
+    """Map minute-of-day onto a noon-origin axis so evening>night comparisons
+    survive midnight (23:30 -> 690, 00:30 -> 750, 11:59 -> 1439)."""
+    return (minute - 720.0) % 1440.0
+
+
+class RhythmEngine:
+    """Anchored, negotiated day-phase state machine (observe stage).
+
+    Phases move forward only (one step per tick) through
+    ``dawn → morning → daylight → evening → wind_down → night → sleep`` with
+    two event-driven exceptions: the sleep vote can end an evening early
+    (``wind_down``/``night`` → ``sleep``) and confirmed wake evidence ends
+    ``dawn``/``sleep`` (→ ``morning``). All thresholds come from the spec;
+    learned anchors refine them but never move outside spec bounds.
+
+    Args:
+        spec: The parsed ``rhythm:`` block.
+        store: Learned anchors (the daemon owns persistence).
+        tz: IANA timezone for wall-clock and day-class decisions.
+    """
+
+    DAWN = "dawn"
+    MORNING = "morning"
+    DAYLIGHT = "daylight"
+    EVENING = "evening"
+    WIND_DOWN = "wind_down"
+    NIGHT = "night"
+    SLEEP = "sleep"
+
+    def __init__(self, spec: RhythmSpec, store: AnchorStore, *, tz: str) -> None:
+        """Wire the spec and store; phase is decided on the first tick."""
+        self._spec = spec
+        self._store = store
+        self._tz = ZoneInfo(tz)
+        self._phase: str | None = None
+        self._morning_started_min: float | None = None
+        self._last_evidence: dict[str, Any] = {}
+
+    @property
+    def phase(self) -> str:
+        """The current phase (``daylight`` before the first tick)."""
+        return self._phase or self.DAYLIGHT
+
+    @property
+    def store(self) -> AnchorStore:
+        """The learned-anchor store (the daemon persists it)."""
+        return self._store
+
+    # -- clock helpers ---------------------------------------------------- #
+    def _local(self, now: float) -> _dt.datetime:
+        """Local wall-clock datetime for epoch ``now``."""
+        return _dt.datetime.fromtimestamp(now, tz=self._tz)
+
+    @staticmethod
+    def _minute(local: _dt.datetime) -> float:
+        """Minutes after local midnight."""
+        return local.hour * 60.0 + local.minute + local.second / 60.0
+
+    def _day_class(self, local: _dt.datetime) -> str:
+        """``"weekend"`` for Sat/Sun mornings, else ``"weekday"``.
+
+        A *night* is classed by the morning it ends: an onset observed before
+        04:00 belongs to that same date's class, a late-evening onset to the
+        next day's.
+        """
+        date = local.date() if local.hour < 4 else local.date() + _dt.timedelta(days=1)
+        return "weekend" if date.weekday() >= 5 else "weekday"
+
+    def _wake_day_class(self, local: _dt.datetime) -> str:
+        """Day class of a wake observation (the date it happened)."""
+        return "weekend" if local.weekday() >= 5 else "weekday"
+
+    # -- anchors ----------------------------------------------------------- #
+    def _wake_anchor_min(self, now: float, signals: SignalState) -> tuple[float, str]:
+        """Tomorrow-or-today's wake anchor minute and its source label.
+
+        Preference order: a real alarm within the next 18 h; the learned
+        median for the applicable day class (weekend capped at the weekday
+        anchor + drift cap); the config default.
+        """
+        if signals.next_alarm_epoch is not None and 0 < signals.next_alarm_epoch - now <= 18 * 3600:
+            alarm_local = self._local(signals.next_alarm_epoch)
+            return self._minute(alarm_local), "alarm"
+        local = self._local(now)
+        day_class = self._day_class(local)
+        learned = self._store.median("wake", day_class)
+        if learned is not None:
+            if day_class == "weekend":
+                weekday_base = self._store.median("wake", "weekday")
+                base = float(weekday_base if weekday_base is not None
+                             else self._spec.wake_default_min)
+                learned = int(min(learned, base + self._spec.weekend_drift_cap_min))
+            return float(learned), f"learned-{day_class}"
+        return float(self._spec.wake_default_min), "default"
+
+    # -- votes ------------------------------------------------------------- #
+    def _sleep_vote(self, presence: PresenceSummary, signals: SignalState) -> tuple[bool, dict[str, Any]]:
+        """The sleep-onset confidence vote and its evidence."""
+        checks = {
+            "quiet": presence.quiet_s >= self._spec.presence.quiet_min * 60.0,
+            "tv_off": not signals.tv_on,
+            "zone_off": not signals.zone_on,
+            "bedroom_or_charging": (
+                presence.last_active_room == self._spec.bedroom or signals.phone_charging),
+        }
+        return all(checks.values()), {"sleep_vote": checks, "quiet_s": round(presence.quiet_s)}
+
+    def _wake_evidence(self, presence: PresenceSummary) -> tuple[bool, dict[str, Any]]:
+        """Sustained-wake test: enough motion AND (bedroom involved OR lights).
+
+        The bedroom/light requirement is pet rule 2 — a multi-room cat patrol
+        that never enters the bedroom and touches no light is not a wake.
+        """
+        sustained = (
+            presence.recent_motion_count >= self._spec.presence.wake_confirm_events
+            or len(presence.recent_rooms) >= 2
+        )
+        anchored = (self._spec.bedroom in presence.recent_rooms
+                    or presence.recent_light_change)
+        return sustained and anchored, {
+            "motion_count": presence.recent_motion_count,
+            "rooms": list(presence.recent_rooms),
+            "light_change": presence.recent_light_change,
+            "bedroom_involved": self._spec.bedroom in presence.recent_rooms,
+        }
+
+    # -- tick ---------------------------------------------------------------- #
+    def tick(self, now: float, presence: PresenceSummary, signals: SignalState) -> PhaseDecision:
+        """Advance at most one phase; return the decision with evidence."""
+        local = self._local(now)
+        minute = self._minute(local)
+        wake_anchor, wake_src = self._wake_anchor_min(now, signals)
+        bed_anchor = float(self._spec.bed_target_min)
+        evidence: dict[str, Any] = {
+            "minute": round(minute), "wake_anchor_min": int(wake_anchor),
+            "wake_anchor_src": wake_src, "bed_anchor_min": int(bed_anchor),
+            "sunset_min": round(signals.sunset_min),
+        }
+        prev = self._phase
+        phase, reason = self._decide(minute, now, local, wake_anchor, bed_anchor,
+                                     presence, signals, evidence)
+        changed = phase != prev
+        self._phase = phase
+        if changed:
+            self._last_evidence = evidence
+        return PhaseDecision(phase=phase, changed=changed, reason=reason, evidence=evidence)
+
+    def _decide(
+        self, minute: float, now: float, local: _dt.datetime,
+        wake_anchor: float, bed_anchor: float,
+        presence: PresenceSummary, signals: SignalState, evidence: dict[str, Any],
+    ) -> tuple[str, str]:
+        """The transition table; mutates ``evidence`` with vote details."""
+        spec = self._spec
+        wind_down_start = _noon_shifted(bed_anchor - spec.wind_down_lead_min)
+        night_start = _noon_shifted(bed_anchor)
+        m_shift = _noon_shifted(minute)
+
+        if self._phase is None:  # first tick: seed from wall clock
+            if m_shift >= night_start:
+                return self.NIGHT, "seed"
+            if m_shift >= wind_down_start:
+                return self.WIND_DOWN, "seed"
+            if m_shift >= _noon_shifted(signals.sunset_min):
+                return self.EVENING, "seed"
+            if minute < wake_anchor:
+                return self.NIGHT, "seed"          # pre-dawn restart: assume night
+            return self.DAYLIGHT, "seed"
+
+        if self._phase == self.SLEEP:
+            woke, wake_ev = self._wake_evidence(presence)
+            evidence.update(wake_ev)
+            in_dawn_window = wake_anchor - spec.dawn_lead_min <= minute < wake_anchor
+            if woke:
+                self._record_wake(local, minute)
+                self._morning_started_min = minute
+                return self.MORNING, "wake-detected"
+            if in_dawn_window:
+                return self.DAWN, "dawn-window"
+            return self.SLEEP, "hold"
+
+        if self._phase == self.DAWN:
+            woke, wake_ev = self._wake_evidence(presence)
+            evidence.update(wake_ev)
+            if woke:
+                self._record_wake(local, minute)
+                self._morning_started_min = minute
+                return self.MORNING, "wake-detected"
+            if minute >= wake_anchor + 120:  # missed detection failsafe
+                self._morning_started_min = minute
+                return self.MORNING, "wake-assumed-late"
+            if minute >= wake_anchor:
+                evidence["snoozed_through"] = True   # alarm passed, no motion yet
+            return self.DAWN, "hold"
+
+        if self._phase == self.MORNING:
+            started = self._morning_started_min if self._morning_started_min is not None else minute
+            if minute >= started + spec.morning_min:
+                return self.DAYLIGHT, "morning-elapsed"
+            return self.MORNING, "hold"
+
+        if self._phase == self.DAYLIGHT:
+            if m_shift >= wind_down_start:
+                return self.WIND_DOWN, "wind-down-lead"
+            if m_shift >= _noon_shifted(signals.sunset_min):
+                return self.EVENING, "sunset"
+            return self.DAYLIGHT, "hold"
+
+        if self._phase == self.EVENING:
+            if m_shift >= wind_down_start:
+                return self.WIND_DOWN, "wind-down-lead"
+            return self.EVENING, "hold"
+
+        if self._phase in (self.WIND_DOWN, self.NIGHT):
+            asleep, vote_ev = self._sleep_vote(presence, signals)
+            evidence.update(vote_ev)
+            if asleep:
+                self._record_sleep_onset(local, minute)
+                return self.SLEEP, "sleep-vote"
+            if self._phase == self.WIND_DOWN and m_shift >= night_start:
+                return self.NIGHT, "bed-anchor"
+            return self._phase, "hold"
+
+        return self.DAYLIGHT, "reset"  # unreachable; defensive
+
+    # -- records ------------------------------------------------------------- #
+    def _record_wake(self, local: _dt.datetime, minute: float) -> None:
+        """Log today's observed wake into the learned store."""
+        self._store.record("wake", self._wake_day_class(local), int(minute),
+                           local.date().isoformat())
+
+    def _record_sleep_onset(self, local: _dt.datetime, minute: float) -> None:
+        """Log tonight's observed sleep onset (classed by the morning it ends)."""
+        self._store.record("sleep_onset", self._day_class(local), int(minute),
+                           local.date().isoformat())
+
+    def snapshot(self, now: float) -> dict[str, Any]:
+        """JSON-serialisable state for persistence and ``hueman rhythm``."""
+        local = self._local(now)
+        return {
+            "phase": self.phase,
+            "as_of": local.isoformat(timespec="seconds"),
+            "bed_anchor_min": self._spec.bed_target_min,
+            "wake_anchor_min": int(self._wake_anchor_min(
+                now, SignalState(None, False, False, False, 0.0))[0]),
+            "learned": {
+                "wake_weekday": self._store.median("wake", "weekday"),
+                "wake_weekend": self._store.median("wake", "weekend"),
+                "sleep_onset_weekday": self._store.median("sleep_onset", "weekday"),
+                "sleep_onset_weekend": self._store.median("sleep_onset", "weekend"),
+            },
+            "last_change_evidence": self._last_evidence,
+        }
diff --git a/tests/test_rhythm_engine.py b/tests/test_rhythm_engine.py
new file mode 100644
index 0000000..075681a
--- /dev/null
+++ b/tests/test_rhythm_engine.py
@@ -0,0 +1,138 @@
+"""Simulated-day tests for the rhythm phase engine."""
+import datetime as dt
+from zoneinfo import ZoneInfo
+
+from hueman.config import RhythmSpec
+from hueman.presence import PresenceSummary
+from hueman.rhythm_control import AnchorStore, RhythmEngine, SignalState
+
+TZ = "America/New_York"
+
+
+def _spec(**kw):
+    base = {"bedroom": "Bedroom"}
+    base.update(kw)
+    return RhythmSpec.parse(base)
+
+
+def _epoch(day, hh, mm):
+    """Epoch seconds for 2026-07-<day> hh:mm in the test tz (a Wednesday=1st)."""
+    return dt.datetime(2026, 7, day, hh, mm, tzinfo=ZoneInfo(TZ)).timestamp()
+
+
+QUIET = PresenceSummary(quiet_s=3600, last_active_room="Bedroom",
+                        recent_rooms=(), recent_motion_count=0,
+                        recent_light_change=False)
+ACTIVE = PresenceSummary(quiet_s=5, last_active_room="Kitchen",
+                         recent_rooms=("Kitchen", "Hallway"),
+                         recent_motion_count=4, recent_light_change=False)
+BEDROOM_BURST = PresenceSummary(quiet_s=5, last_active_room="Bedroom",
+                                recent_rooms=("Bedroom",),
+                                recent_motion_count=3,
+                                recent_light_change=False)
+
+
+def _sig(alarm=None, charging=False, tv=False, zone_on=True, sunset=20 * 60 + 30):
+    return SignalState(next_alarm_epoch=alarm, phone_charging=charging,
+                       tv_on=tv, zone_on=zone_on, sunset_min=float(sunset))
+
+
+def test_starts_in_daylight_and_reports_no_change_on_steady_state():
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    d1 = eng.tick(_epoch(1, 12, 0), ACTIVE, _sig())
+    assert d1.phase == RhythmEngine.DAYLIGHT and d1.changed is True  # first tick sets phase
+    d2 = eng.tick(_epoch(1, 12, 1), ACTIVE, _sig())
+    assert d2.changed is False
+
+
+def test_evening_at_sunset_and_wind_down_before_bed_target():
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    eng.tick(_epoch(1, 12, 0), ACTIVE, _sig())
+    d = eng.tick(_epoch(1, 20, 31), ACTIVE, _sig())            # past sunset 20:30
+    assert d.phase == RhythmEngine.EVENING and d.changed
+    d = eng.tick(_epoch(1, 21, 31), ACTIVE, _sig())            # 23:00 - 90m = 21:30
+    assert d.phase == RhythmEngine.WIND_DOWN and d.changed
+    assert d.evidence["bed_anchor_min"] == 23 * 60
+
+
+def test_night_at_bed_anchor_then_sleep_when_vote_passes():
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    eng.tick(_epoch(1, 22, 0), ACTIVE, _sig())
+    d = eng.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    assert d.phase == RhythmEngine.NIGHT
+    # vote fails while the TV is on or the zone is lit or the house is active
+    d = eng.tick(_epoch(1, 23, 30), QUIET, _sig(tv=True, zone_on=False))
+    assert d.phase == RhythmEngine.NIGHT
+    d = eng.tick(_epoch(1, 23, 40), QUIET, _sig(zone_on=True))
+    assert d.phase == RhythmEngine.NIGHT
+    # quiet + tv off + zone off + last room bedroom -> SLEEP, onset recorded
+    store = AnchorStore()
+    eng2 = RhythmEngine(_spec(), store, tz=TZ)
+    eng2.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    d = eng2.tick(_epoch(1, 23, 50), QUIET, _sig(zone_on=False))
+    assert d.phase == RhythmEngine.SLEEP and d.changed
+    assert d.reason == "sleep-vote"
+    assert store.median("sleep_onset", "weekday") is not None
+
+
+def test_sleep_vote_needs_quiet():
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    eng.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    d = eng.tick(_epoch(1, 23, 50), ACTIVE, _sig(zone_on=False))
+    assert d.phase == RhythmEngine.NIGHT
+
+
+def test_dawn_from_alarm_then_wake_on_bedroom_burst_records_anchor():
+    store = AnchorStore()
+    eng = RhythmEngine(_spec(), store, tz=TZ)
+    eng.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    eng.tick(_epoch(1, 23, 50), QUIET, _sig(zone_on=False))     # -> SLEEP
+    alarm = _epoch(2, 6, 45)
+    d = eng.tick(_epoch(2, 6, 25), QUIET, _sig(alarm=alarm, zone_on=False))
+    assert d.phase == RhythmEngine.DAWN                          # 6:45 - 25m = 6:20
+    assert d.evidence["wake_anchor_min"] == 6 * 60 + 45
+    d = eng.tick(_epoch(2, 6, 50), BEDROOM_BURST, _sig(alarm=alarm, zone_on=False))
+    assert d.phase == RhythmEngine.MORNING and d.reason == "wake-detected"
+    assert store.median("wake", "weekday") == 6 * 60 + 50
+
+
+def test_wake_without_bedroom_or_light_change_is_not_wake():
+    """Rule 2: a cat patrol (multi-room, no bedroom, no lights) stays SLEEP."""
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    eng.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    eng.tick(_epoch(1, 23, 50), QUIET, _sig(zone_on=False))     # -> SLEEP
+    cat = PresenceSummary(quiet_s=5, last_active_room="Hallway",
+                          recent_rooms=("Hallway", "Kitchen"),
+                          recent_motion_count=5, recent_light_change=False)
+    d = eng.tick(_epoch(2, 3, 0), cat, _sig(zone_on=False))
+    assert d.phase == RhythmEngine.SLEEP
+
+
+def test_morning_lasts_morning_min_then_daylight():
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    eng.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    eng.tick(_epoch(1, 23, 50), QUIET, _sig(zone_on=False))
+    eng.tick(_epoch(2, 6, 50), BEDROOM_BURST, _sig(zone_on=False))  # MORNING
+    d = eng.tick(_epoch(2, 7, 55), ACTIVE, _sig())
+    assert d.phase == RhythmEngine.DAYLIGHT
+
+
+def test_weekend_day_class_used_for_records():
+    store = AnchorStore()
+    eng = RhythmEngine(_spec(), store, tz=TZ)
+    # 2026-07-03 is a Friday; sleep onset recorded late Friday belongs to
+    # Saturday's (weekend) class — the night is classed by the morning it ends.
+    eng.tick(_epoch(3, 23, 1), ACTIVE, _sig())
+    eng.tick(_epoch(3, 23, 50), QUIET, _sig(zone_on=False))
+    assert store.median("sleep_onset", "weekend") is not None
+    assert store.median("sleep_onset", "weekday") is None
+
+
+def test_snapshot_is_json_serialisable_and_carries_anchors():
+    import json
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    eng.tick(_epoch(1, 12, 0), ACTIVE, _sig())
+    snap = eng.snapshot(_epoch(1, 12, 1))
+    json.dumps(snap)
+    assert snap["phase"] == RhythmEngine.DAYLIGHT
+    assert "bed_anchor_min" in snap and "wake_anchor_min" in snap

From 481dca1f2488118a3227790f6ba2c8cba98dae51 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 01:59:11 -0700
Subject: [PATCH 08/15] fix(rhythm): copy evidence dict; test weekend drift cap

---
 hueman/rhythm_control.py    |  2 +-
 tests/test_rhythm_engine.py | 13 +++++++++++++
 2 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/hueman/rhythm_control.py b/hueman/rhythm_control.py
index 12df01d..d40a4a4 100644
--- a/hueman/rhythm_control.py
+++ b/hueman/rhythm_control.py
@@ -270,7 +270,7 @@ def tick(self, now: float, presence: PresenceSummary, signals: SignalState) -> P
         changed = phase != prev
         self._phase = phase
         if changed:
-            self._last_evidence = evidence
+            self._last_evidence = dict(evidence)
         return PhaseDecision(phase=phase, changed=changed, reason=reason, evidence=evidence)
 
     def _decide(
diff --git a/tests/test_rhythm_engine.py b/tests/test_rhythm_engine.py
index 075681a..b25cb17 100644
--- a/tests/test_rhythm_engine.py
+++ b/tests/test_rhythm_engine.py
@@ -128,6 +128,19 @@ def test_weekend_day_class_used_for_records():
     assert store.median("sleep_onset", "weekday") is None
 
 
+def test_weekend_learned_wake_is_capped_by_drift_rule():
+    """A learned weekend wake later than weekday+cap is clamped to the cap."""
+    store = AnchorStore()
+    store.record("wake", "weekday", 420, "2026-07-01")   # 07:00
+    store.record("wake", "weekend", 600, "2026-07-04")   # 10:00 observed
+    eng = RhythmEngine(_spec(), store, tz=TZ)
+    # 2026-07-04 is a Saturday; day_class at Saturday noon is weekend
+    d = eng.tick(_epoch(4, 12, 0), ACTIVE, _sig())
+    # cap = weekday 420 + 90 drift = 510 (08:30), so 600 clamps to 510
+    assert d.evidence["wake_anchor_min"] == 510
+    assert d.evidence["wake_anchor_src"] == "learned-weekend"
+
+
 def test_snapshot_is_json_serialisable_and_carries_anchors():
     import json
     eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)

From b6d8f88ccb0fd0478cb2b30bdceff1efa8539efb Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 02:08:39 -0700
Subject: [PATCH 09/15] =?UTF-8?q?feat(rhythm):=20daemon=20observe-stage=20?=
 =?UTF-8?q?plumbing=20=E2=80=94=20SSE=20routing,=20signals,=20evidence=20l?=
 =?UTF-8?q?og,=20atomic=20state?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 hueman/circadian_daemon.py  | 142 +++++++++++++++++++++++++++++++++++-
 tests/test_daemon_rhythm.py | 105 ++++++++++++++++++++++++++
 2 files changed, 246 insertions(+), 1 deletion(-)
 create mode 100644 tests/test_daemon_rhythm.py

diff --git a/hueman/circadian_daemon.py b/hueman/circadian_daemon.py
index f4863ad..4aec7b9 100644
--- a/hueman/circadian_daemon.py
+++ b/hueman/circadian_daemon.py
@@ -30,7 +30,10 @@
 
 from __future__ import annotations
 
+import datetime as _dt
+import json
 import logging
+import math
 import os
 import random
 from concurrent.futures import ThreadPoolExecutor
@@ -41,6 +44,7 @@
 import time
 from collections.abc import Callable
 from typing import cast
+from zoneinfo import ZoneInfo
 
 import requests
 
@@ -54,10 +58,12 @@
 )
 from .circadian_control import CircadianController, DriveTo, FadeOff, Hold
 from .client import HueClient
-from .config import CircadianDaemonSpec, Config
+from .config import CircadianDaemonSpec, Config, RhythmSpec
 from .engine import TargetState
 from .errors import AuthError, BridgeError, ConfigError
 from .payload import GroupedLightCommand, LightCommand
+from .presence import ActivityEvent, PresenceTracker
+from .rhythm_control import AnchorStore, RhythmEngine, SignalState
 from .security_control import SecurityController, unknown_security_groups
 from .state import BridgeState
 from .sun import SolarCalculator
@@ -90,6 +96,7 @@ class CircadianDaemon:
     _security_sse_off_rid: str | None
     _security_rids: dict[str, str]
     _security_light_rids: tuple[str, ...]
+    _rhythm_motion_rooms: dict[str, str]
 
     def __init__(
         self,
@@ -156,6 +163,16 @@ def __init__(
             self._rebuild_security_controller()
             self._security_sse_on_rid = self._resolve_resume_trigger(state, sec.sse_on)
             self._security_sse_off_rid = self._resolve_resume_trigger(state, sec.sse_off)
+        if config.rhythm is not None:
+            self._rhythm_motion_rooms = {
+                srid: (area.room_name or area.name)
+                for area in state.motion_areas
+                for srid in area.service_rids
+            }
+            if not self._rhythm_motion_rooms:
+                _LOG.warning(
+                    "rhythm: no MotionAware areas found on the bridge — presence "
+                    "inference will see only manual-override evidence")
         self._configure_logging(spec)
 
     @classmethod
@@ -191,6 +208,7 @@ def _setup(
         spec = self._require_spec(config)
         loc = config.location
         solar = SolarCalculator(loc.lat, loc.lon, loc.tz_offset_hours, tz=loc.tz)
+        self._solar = solar
         self._client = client
         self._config = config
         self._spec = spec
@@ -255,6 +273,25 @@ def _setup(
         self._security_rids = {}                      # group name -> grouped_light rid
         self._security_light_rids = ()                # member lights (chaos units)
         self._security_thread: threading.Thread | None = None
+        # --- rhythm engine (optional, observe stage only) ---
+        self._rhythm: RhythmSpec | None = config.rhythm
+        self._presence: PresenceTracker | None = None
+        self._rhythm_engine: RhythmEngine | None = None
+        self._rhythm_tz: ZoneInfo | None = None
+        self._rhythm_motion_rooms = {}   # *_area_motion service rid -> room
+        if self._rhythm is not None:
+            if self._rhythm.stage != "observe":
+                raise ConfigError(
+                    f"rhythm.stage {self._rhythm.stage!r} is not implemented yet; "
+                    "only 'observe' ships in stage 1 — see the rhythm design spec")
+            if loc.tz is None:
+                raise ConfigError(
+                    "rhythm requires location.tz (an IANA timezone name) for "
+                    "wall-clock day-phase reasoning; tz_offset_hours alone is not enough")
+            self._presence = PresenceTracker(self._rhythm.presence)
+            self._rhythm_tz = ZoneInfo(loc.tz)
+            self._rhythm_engine = RhythmEngine(
+                self._rhythm, self._load_anchor_store(), tz=loc.tz)
 
     def _rebuild_security_controller(self) -> None:
         """(Re)build the controller with the resolved member lights, so chaos
@@ -472,6 +509,11 @@ def _tick_once(self, now: float) -> None:
             self._poll_control_file(now)
             if self._bias is not None:
                 self._apply_bias(now)
+            if self._rhythm is not None:
+                try:
+                    self._rhythm_tick(now)
+                except Exception as e:  # inference must never break the drive loop
+                    _LOG.warning("rhythm tick error: %s", e)
 
     def _apply_bias(self, now: float) -> None:
         """Drive each bias light per :func:`bias_actions` for this tick.
@@ -607,6 +649,93 @@ def _poll_bias_files(self, now: float) -> None:
         if bias.file_off:
             self._consume_flag(bias.file_off, lambda: self._bias_aggregator.off(now, "file"), "bias OFF")
 
+    # -- rhythm engine (observe stage) --------------------------------------- #
+    def _load_anchor_store(self) -> AnchorStore:
+        """Load learned anchors from the state file; missing/corrupt = empty.
+
+        An unreadable store must never stop the daemon: learning restarts and
+        the engine falls back to config defaults (the spec's safety rail).
+        """
+        assert self._rhythm is not None  # only called when rhythm is configured
+        try:
+            with open(self._rhythm.state_file) as fh:
+                return AnchorStore.from_json(json.load(fh))
+        except (OSError, ValueError):
+            return AnchorStore()
+
+    def _read_alarm_epoch(self) -> float | None:
+        """Read the next-alarm signal file; garbage/0/missing/unset -> None."""
+        rhythm = self._rhythm
+        if rhythm is None or not rhythm.signals.next_alarm_file:
+            return None
+        try:
+            raw = open(rhythm.signals.next_alarm_file).read().strip()
+            value = float(raw)
+            return value if value > 0 else None
+        except (OSError, ValueError):
+            return None
+
+    def _phone_charging(self) -> bool:
+        """The charging signal file's existence means the phone is charging."""
+        rhythm = self._rhythm
+        if rhythm is None or not rhythm.signals.charging_file:
+            return False
+        try:
+            return os.path.exists(rhythm.signals.charging_file)
+        except OSError:  # pragma: no cover - filesystem-dependent
+            return False
+
+    def _rhythm_note_manual(self, now: float) -> None:
+        """Feed a detected manual override into presence as human evidence."""
+        if self._presence is None:
+            return
+        judgment = self._presence.feed(
+            ActivityEvent(room="", kind="light_change", ts=now))
+        _LOG.info("rhythm: manual light change -> human evidence (%s)", judgment.rule)
+
+    def _rhythm_tick(self, now: float) -> None:
+        """One observe-stage inference tick: decide, log evidence, persist.
+
+        Never writes to the bridge; a failure here must not disturb the
+        circadian tick, so callers wrap it (see :meth:`_tick_once`).
+        """
+        engine, presence, rhythm = self._rhythm_engine, self._presence, self._rhythm
+        if engine is None or presence is None or rhythm is None:
+            return
+        assert self._rhythm_tz is not None  # set together with the engine in _setup
+        # Sunset minute via the solar calculator (same source the curve uses);
+        # a polar infinite sunset falls back to 20:00.
+        local_date = _dt.datetime.fromtimestamp(now, tz=self._rhythm_tz).date()
+        sun = self._solar.sun_times(local_date)
+        sunset_min = float(sun.sunset_min) if math.isfinite(sun.sunset_min) else 20 * 60.0
+        signals = SignalState(
+            next_alarm_epoch=self._read_alarm_epoch(),
+            phone_charging=self._phone_charging(),
+            tv_on=bool(self._bias_aggregator.tv_on(now)) if self._bias is not None else False,
+            zone_on=self._cmd_on,
+            sunset_min=sunset_min,
+        )
+        decision = engine.tick(now, presence.summary(now), signals)
+        if decision.changed:
+            _LOG.info("rhythm: phase -> %s (%s) evidence=%s",
+                      decision.phase, decision.reason, json.dumps(decision.evidence))
+            self._persist_rhythm_state(now)
+
+    def _persist_rhythm_state(self, now: float) -> None:
+        """Atomically write anchors + snapshot to the state file."""
+        engine, rhythm = self._rhythm_engine, self._rhythm
+        if engine is None or rhythm is None:
+            return
+        doc: dict[str, object] = {"version": 1, "snapshot": engine.snapshot(now)}
+        doc.update(engine.store.to_json())  # {"anchors": ...} via the engine's store property
+        try:
+            tmp = rhythm.state_file + ".tmp"
+            with open(tmp, "w") as fh:
+                json.dump(doc, fh, indent=2)
+            os.replace(tmp, rhythm.state_file)
+        except OSError as e:  # pragma: no cover - filesystem-dependent
+            _LOG.warning("rhythm state write failed for %s: %s", rhythm.state_file, e)
+
     def _consume_flag(self, path: str, action: Callable[[], None], label: str) -> None:
         """If ``path`` exists, run ``action`` and remove it (best-effort)."""
         try:
@@ -987,6 +1116,15 @@ def _handle_event(self, event: BridgeEvent, now: float) -> None:
                     _LOG.info("resume trigger %s -> resumed", event.rid)
                     self._controller.on_resume(now)
                 return
+            if (self._presence is not None
+                    and event.rtype in ("convenience_area_motion", "security_area_motion")):
+                room = self._rhythm_motion_rooms.get(event.rid)
+                if room is not None and event.data.get("motion", {}).get("motion"):
+                    judgment = self._presence.feed(
+                        ActivityEvent(room=room, kind="motion", ts=now))
+                    _LOG.debug("rhythm: motion in %r -> human=%s (%s)",
+                               room, judgment.human, judgment.rule)
+                return
             if event.rtype != "grouped_light" or event.rid != self._rid:
                 return
             on_state = event.data.get("on", {}).get("on")
@@ -1000,6 +1138,7 @@ def _handle_event(self, event: BridgeEvent, now: float) -> None:
                     if self._cmd_on and self._controller.mode != CircadianController.SUSPENDED:
                         _LOG.info("zone turned off -> suspended")
                         self._controller.on_external_change(now)
+                        self._rhythm_note_manual(now)
                     # else: cmd_on already False -> our own fade-off, ignore.
                 else:  # on_state is True
                     if (
@@ -1044,6 +1183,7 @@ def _handle_event(self, event: BridgeEvent, now: float) -> None:
                     self._cmd_brightness,
                 )
                 self._controller.on_external_change(now)
+                self._rhythm_note_manual(now)
             else:
                 target_str = "?" if self._cmd_brightness is None else f"{self._cmd_brightness:.1f}%"
                 _LOG.debug(
diff --git a/tests/test_daemon_rhythm.py b/tests/test_daemon_rhythm.py
new file mode 100644
index 0000000..c779d63
--- /dev/null
+++ b/tests/test_daemon_rhythm.py
@@ -0,0 +1,105 @@
+"""Daemon plumbing tests for the rhythm observe stage (no bridge writes)."""
+import json
+
+import pytest
+
+from hueman.circadian_daemon import CircadianDaemon
+from hueman.config import Config
+from hueman.watch import BridgeEvent
+
+
+def _config(tmp_path, **rhythm_extra):
+    rhythm = {
+        "bedroom": "Bedroom",
+        "state_file": str(tmp_path / "rhythm-state.json"),
+        "signals": {
+            "next_alarm_file": str(tmp_path / ".next-alarm"),
+            "charging_file": str(tmp_path / ".phone-charging"),
+        },
+    }
+    rhythm.update(rhythm_extra)
+    return Config.parse({
+        "bridge": {"host": "bridge.local", "application_key": "k"},
+        "location": {"lat": 40.0, "lon": -75.0, "tz": "America/New_York"},
+        "circadian": {"day": {"brightness": 90, "kelvin": 5000},
+                      "evening": {"brightness": 40, "kelvin": 2700},
+                      "night": {"brightness": 15, "kelvin": 2200}},
+        "circadian_daemon": {"zone": "Main"},
+        "rhythm": rhythm,
+    })
+
+
+class _NullClient:
+    """Fails the test if the daemon writes to the bridge in observe stage."""
+
+    class bridge:
+        host = "bridge.local"
+        application_key = "k"
+
+    def update_resource(self, rtype, rid, body):
+        raise AssertionError(f"observe stage wrote to the bridge: {rtype} {rid}")
+
+
+def _daemon(tmp_path, **rhythm_extra):
+    cfg = _config(tmp_path, **rhythm_extra)
+    d = CircadianDaemon.for_test(_NullClient(), cfg, grouped_light_rid="gl-1")
+    d._rhythm_motion_rooms = {"svc-bed": "Bedroom", "svc-kitchen": "Kitchen"}
+    return d
+
+
+def test_stage_beyond_observe_refuses_to_start(tmp_path):
+    from hueman.errors import ConfigError
+    with pytest.raises(ConfigError, match="rhythm.stage"):
+        _daemon(tmp_path, stage="mornings")
+
+
+def test_motion_event_feeds_tracker_by_room(tmp_path):
+    d = _daemon(tmp_path)
+    ev = BridgeEvent(rtype="convenience_area_motion", rid="svc-bed",
+                     data={"motion": {"motion": True}})
+    d._handle_event(ev, 1_000_000.0)
+    s = d._presence.summary(1_000_000.0 + 1)
+    assert s.recent_motion_count == 1
+    assert s.recent_rooms == ("Bedroom",)
+
+
+def test_motion_clear_and_unknown_rid_are_ignored(tmp_path):
+    d = _daemon(tmp_path)
+    d._handle_event(BridgeEvent(rtype="convenience_area_motion", rid="svc-bed",
+                                data={"motion": {"motion": False}}), 1_000_000.0)
+    d._handle_event(BridgeEvent(rtype="convenience_area_motion", rid="svc-nope",
+                                data={"motion": {"motion": True}}), 1_000_000.0)
+    assert d._presence.summary(1_000_001.0).recent_motion_count == 0
+
+
+def test_tick_reads_signal_files_and_persists_state(tmp_path):
+    d = _daemon(tmp_path)
+    (tmp_path / ".next-alarm").write_text("1800000000")
+    (tmp_path / ".phone-charging").write_text("")
+    d._rhythm_tick(1_000_000.0)
+    state = json.loads((tmp_path / "rhythm-state.json").read_text())
+    assert state["version"] == 1
+    assert state["snapshot"]["phase"] in (
+        "dawn", "morning", "daylight", "evening", "wind_down", "night", "sleep")
+
+
+def test_alarm_file_garbage_is_none(tmp_path):
+    d = _daemon(tmp_path)
+    (tmp_path / ".next-alarm").write_text("not-a-number")
+    assert d._read_alarm_epoch() is None
+    (tmp_path / ".next-alarm").write_text("0")
+    assert d._read_alarm_epoch() is None
+
+
+def test_rhythm_disabled_is_inert(tmp_path):
+    cfg = Config.parse({
+        "bridge": {"host": "bridge.local", "application_key": "k"},
+        "location": {"lat": 40.0, "lon": -75.0, "tz": "America/New_York"},
+        "circadian": {"day": {"brightness": 90, "kelvin": 5000},
+                      "evening": {"brightness": 40, "kelvin": 2700},
+                      "night": {"brightness": 15, "kelvin": 2200}},
+        "circadian_daemon": {"zone": "Main"},
+    })
+    d = CircadianDaemon.for_test(_NullClient(), cfg, grouped_light_rid="gl-1")
+    assert d._rhythm is None
+    d._rhythm_tick(1_000_000.0)  # no-op, no crash, no file

From 225503a6ecc31a36669b4a7abbe0cefd59e14f9c Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 02:18:08 -0700
Subject: [PATCH 10/15] feat(rhythm): 'hueman rhythm' status command (anchors,
 phase error, evidence)

---
 hueman/cli.py            | 48 ++++++++++++++++++++++++++++++
 tests/test_cli_rhythm.py | 64 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 112 insertions(+)
 create mode 100644 tests/test_cli_rhythm.py

diff --git a/hueman/cli.py b/hueman/cli.py
index e540cd9..9e1283f 100644
--- a/hueman/cli.py
+++ b/hueman/cli.py
@@ -16,6 +16,7 @@
 
 import argparse
 import datetime as _dt
+import json
 import sys
 from collections.abc import Callable
 
@@ -86,6 +87,8 @@ def _build_parser(self) -> argparse.ArgumentParser:
         circ_sub.add_parser("run", help="run the daemon (foreground; container entrypoint)")
         circ_sub.add_parser("resume", help="hand control back to a running daemon")
 
+        sub.add_parser("rhythm", help="print the rhythm engine's phase, anchors, and evidence")
+
         security_parser = sub.add_parser("security", help="arm/disarm security mode")
         sec_sub = security_parser.add_subparsers(dest="security_cmd", required=True)
         sec_sub.add_parser("on", help="arm security mode (panic)")
@@ -251,6 +254,51 @@ def _cmd_security(self, args: argparse.Namespace) -> int:
         print(f"  pending on-file present: {pending}")
         return 0
 
+    def _cmd_rhythm(self, args: argparse.Namespace) -> int:
+        """Print the rhythm engine's current state, anchors, and evidence."""
+        config = load_config(args.config)
+        rhythm = config.rhythm
+        if rhythm is None:
+            print("config has no 'rhythm' block", file=sys.stderr)
+            return 1
+        try:
+            with open(rhythm.state_file) as fh:
+                doc = json.load(fh)
+        except (OSError, ValueError):
+            print(
+                f"rhythm state not found at {rhythm.state_file} — the daemon "
+                "has not written it yet (is it running with rhythm: enabled?)",
+                file=sys.stderr,
+            )
+            return 1
+        snap = doc.get("snapshot", {})
+
+        def hhmm(minute: int | None) -> str:
+            """Render minutes-after-midnight as ``HH:MM``, or ``--:--`` if unset."""
+            return "--:--" if minute is None else f"{minute // 60:02d}:{minute % 60:02d}"
+
+        learned = snap.get("learned", {})
+        onset = learned.get("sleep_onset_weekday")
+        error_min = None if onset is None else onset - rhythm.bed_target_min
+        print(f"phase:        {snap.get('phase', '?')}  (as of {snap.get('as_of', '?')})")
+        print(f"bed anchor:   {hhmm(snap.get('bed_anchor_min'))}  (target)")
+        print(f"wake anchor:  {hhmm(snap.get('wake_anchor_min'))}")
+        print("learned:")
+        print(
+            f"  wake        weekday {hhmm(learned.get('wake_weekday'))}  "
+            f"weekend {hhmm(learned.get('wake_weekend'))}"
+        )
+        print(
+            f"  sleep onset weekday {hhmm(learned.get('sleep_onset_weekday'))}  "
+            f"weekend {hhmm(learned.get('sleep_onset_weekend'))}"
+        )
+        if error_min is None:
+            print("phase error:  n/a (no observed sleep onsets yet)")
+        else:
+            print(f"phase error:  {error_min:+d} min (observed weekday onset vs target)")
+        print(f"last change:  {json.dumps(snap.get('last_change_evidence', {}))}")
+        return 0
+
     # -- helpers ------------------------------------------------------------ #
     def _build_planner(self, args: argparse.Namespace) -> tuple[Config, Planner]:
         """Load config, connect, and build a planner over live bridge state."""
diff --git a/tests/test_cli_rhythm.py b/tests/test_cli_rhythm.py
new file mode 100644
index 0000000..c470182
--- /dev/null
+++ b/tests/test_cli_rhythm.py
@@ -0,0 +1,64 @@
+"""Tests for the `hueman rhythm` status command."""
+import json
+
+from hueman.cli import main
+
+
+def _write_config(tmp_path, state_file):
+    cfg = tmp_path / "hue.yaml"
+    cfg.write_text(
+        "bridge:\n  host: bridge.local\n  application_key: k\n"
+        "location:\n  lat: 40.0\n  lon: -75.0\n  tz: America/New_York\n"
+        "circadian:\n"
+        "  day:\n    brightness: 90\n    kelvin: 5000\n"
+        "  evening:\n    brightness: 40\n    kelvin: 2700\n"
+        "  night:\n    brightness: 15\n    kelvin: 2200\n"
+        "rhythm:\n"
+        "  bedroom: Bedroom\n"
+        f"  state_file: {state_file}\n"
+    )
+    return cfg
+
+
+def test_rhythm_status_prints_snapshot(tmp_path, capsys):
+    state = tmp_path / "rhythm-state.json"
+    state.write_text(json.dumps({
+        "version": 1,
+        "snapshot": {
+            "phase": "wind_down",
+            "as_of": "2026-07-05T21:45:00-04:00",
+            "bed_anchor_min": 1380,
+            "wake_anchor_min": 405,
+            "learned": {"wake_weekday": 412, "wake_weekend": None,
+                        "sleep_onset_weekday": 1395, "sleep_onset_weekend": None},
+            "last_change_evidence": {"minute": 1305, "reason_detail": "wind-down-lead"},
+        },
+        "anchors": {"weekday": {"wake": [{"date": "2026-07-03", "minute": 412}]}},
+    }))
+    cfg = _write_config(tmp_path, state)
+    assert main(["-c", str(cfg), "rhythm"]) == 0
+    out = capsys.readouterr().out
+    assert "wind_down" in out
+    assert "23:00" in out            # bed anchor rendered as HH:MM
+    assert "06:52" in out            # learned weekday wake 412 -> HH:MM
+    assert "phase error" in out.lower()
+
+
+def test_rhythm_status_without_state_file(tmp_path, capsys):
+    cfg = _write_config(tmp_path, tmp_path / "missing.json")
+    assert main(["-c", str(cfg), "rhythm"]) == 1
+    assert "daemon has not written" in capsys.readouterr().err.lower()
+
+
+def test_rhythm_status_without_rhythm_block(tmp_path, capsys):
+    cfg = tmp_path / "hue.yaml"
+    cfg.write_text(
+        "bridge:\n  host: bridge.local\n  application_key: k\n"
+        "location:\n  lat: 40.0\n  lon: -75.0\n  tz: America/New_York\n"
+        "circadian:\n"
+        "  day:\n    brightness: 90\n    kelvin: 5000\n"
+        "  evening:\n    brightness: 40\n    kelvin: 2700\n"
+        "  night:\n    brightness: 15\n    kelvin: 2200\n"
+    )
+    assert main(["-c", str(cfg), "rhythm"]) == 1
+    assert "no 'rhythm' block" in capsys.readouterr().err.lower()

From f451db54e8c72fd47d8e107c36670a529da2ae20 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 02:27:25 -0700
Subject: [PATCH 11/15] fix(cli): rhythm status degrades cleanly on malformed
 state

---
 hueman/cli.py            | 22 ++++++++++++++++------
 tests/test_cli_rhythm.py | 19 +++++++++++++++++++
 2 files changed, 35 insertions(+), 6 deletions(-)

diff --git a/hueman/cli.py b/hueman/cli.py
index 9e1283f..21a6fcd 100644
--- a/hueman/cli.py
+++ b/hueman/cli.py
@@ -271,15 +271,25 @@ def _cmd_rhythm(self, args: argparse.Namespace) -> int:
                 file=sys.stderr,
             )
             return 1
-        snap = doc.get("snapshot", {})
+        snap = doc.get("snapshot") if isinstance(doc, dict) else None
+        if not isinstance(snap, dict):
+            print(
+                f"rhythm state at {rhythm.state_file} is malformed — delete it "
+                "and let the daemon rewrite it",
+                file=sys.stderr,
+            )
+            return 1
 
-        def hhmm(minute: int | None) -> str:
-            """Render minutes-after-midnight as ``HH:MM``, or ``--:--`` if unset."""
-            return "--:--" if minute is None else f"{minute // 60:02d}:{minute % 60:02d}"
+        def hhmm(minute: object) -> str:
+            """Render minutes-after-midnight as ``HH:MM``, or ``--:--`` if unset/invalid."""
+            if not isinstance(minute, int):
+                return "--:--"
+            return f"{minute // 60:02d}:{minute % 60:02d}"
 
-        learned = snap.get("learned", {})
+        raw_learned = snap.get("learned")
+        learned = raw_learned if isinstance(raw_learned, dict) else {}
         onset = learned.get("sleep_onset_weekday")
-        error_min = None if onset is None else onset - rhythm.bed_target_min
+        error_min = onset - rhythm.bed_target_min if isinstance(onset, int) else None
         print(f"phase:        {snap.get('phase', '?')}  (as of {snap.get('as_of', '?')})")
         print(f"bed anchor:   {hhmm(snap.get('bed_anchor_min'))}  (target)")
         print(f"wake anchor:  {hhmm(snap.get('wake_anchor_min'))}")
diff --git a/tests/test_cli_rhythm.py b/tests/test_cli_rhythm.py
index c470182..1f5b54e 100644
--- a/tests/test_cli_rhythm.py
+++ b/tests/test_cli_rhythm.py
@@ -62,3 +62,22 @@ def test_rhythm_status_without_rhythm_block(tmp_path, capsys):
     )
     assert main(["-c", str(cfg), "rhythm"]) == 1
     assert "no 'rhythm' block" in capsys.readouterr().err.lower()
+
+
+def test_rhythm_status_with_non_dict_snapshot(tmp_path, capsys):
+    state = tmp_path / "rhythm-state.json"
+    state.write_text('{"version": 1, "snapshot": "corrupt"}')
+    cfg = _write_config(tmp_path, state)
+    assert main(["-c", str(cfg), "rhythm"]) == 1
+    assert "malformed" in capsys.readouterr().err.lower()
+
+
+def test_rhythm_status_with_wrong_typed_minutes_degrades(tmp_path, capsys):
+    state = tmp_path / "rhythm-state.json"
+    state.write_text(
+        '{"version": 1, "snapshot": {"phase": "night", "bed_anchor_min": "1380",'
+        ' "learned": {"sleep_onset_weekday": "oops"}}}')
+    cfg = _write_config(tmp_path, state)
+    assert main(["-c", str(cfg), "rhythm"]) == 0
+    out = capsys.readouterr().out
+    assert "--:--" in out and "n/a" in out.lower()

From 4403aa8f3c5f305a99f7e7c9673f991162f1c22b Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 02:35:36 -0700
Subject: [PATCH 12/15] docs(rhythm): observe-stage docs, example config,
 CLAUDE.md pointer

---
 CLAUDE.md          | 21 +++++++++--
 README.md          | 90 +++++++++++++++++++++++++++++++++++++++++++---
 examples/home.yaml | 25 +++++++++++++
 hueman/cli.py      |  9 ++++-
 4 files changed, 138 insertions(+), 7 deletions(-)

diff --git a/CLAUDE.md b/CLAUDE.md
index e7c9441..715087a 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -175,6 +175,21 @@ Triggers (OR-combined, no arming): `hueman security on|off` (writes the
 adapter (e.g. Home Assistant) watches the cue file / webhook and plays the
 matching sound. That adapter is out of this repo's scope.
 
+## Rhythm engine — observe-stage day-phase inference
+
+`rhythm:` (optional, alongside `circadian_daemon:`) runs a closed-loop
+day-phase inference engine (`rhythm_control.py` + `presence.py`) inside the
+daemon: it infers `dawn`/`morning`/`daylight`/`evening`/`wind_down`/`night`/
+`sleep` from MotionAware motion (with pet discounting) plus optional phone
+signals, and learns wake/bed-time anchors over time. **Stage 1 (`stage:
+"observe"`) is read-only — it never writes to the bridge**; every inference is
+logged with a `rhythm:` prefix in the daemon log and persisted to
+`rhythm.state_file` (the path comes from config; delete it to reset
+learning). `stage: "mornings"` and `stage: "full"` parse but the daemon
+deliberately refuses to start with either — only `observe` ships in stage 1.
+See `docs/superpowers/specs/2026-07-04-rhythm-engine-design.md` and the
+README's "Rhythm engine (observe stage)" section.
+
 ## Architecture
 
 The codebase is split into a **pure decision layer** (fully unit-tested, no I/O) and a **thin I/O layer** (needs a real bridge):
@@ -193,6 +208,8 @@ The codebase is split into a **pure decision layer** (fully unit-tested, no I/O)
 | `engine.py` | Per-area state machine (`PolicyEngine`) for the legacy `watch` runtime. Consumes `MotionPolicy` + explicit `ts` floats; emits `Action` values. Four phases: `STANDBY`, `ACTIVE`, `DIMMING`, `OVERRIDDEN`. |
 | `payload.py` | Converts engine `TargetState` → CLIP v2 request bodies (sRGB `#rrggbb` → CIE xy). |
 | `nightmotion.py` | Pure night-motion helpers: builds CLIP `scene` bodies, transforms the MotionAware `behavior_instance`, and `scene_actions_match` (tolerant scene-look diff, reused by the circadian reconciler). |
+| `presence.py` | Pure pet-discounting activity judge for `rhythm` (`PresenceTracker`): light-change / progression / solo-motion rules, quiet-time and confirm-window summaries. |
+| `rhythm_control.py` | Pure day-phase state machine for `rhythm` (`RhythmEngine`, `AnchorStore`, observe stage): phase transitions, sleep vote, wake confirmation, learned wake/bed anchors. |
 | `reconcile.py` | Terraform-style planner. `Planner` runs `AreaReconciler` (room/zone membership), `SensitivityReconciler` (sensor sensitivity), `SmartSceneReconciler` (re-time an existing scene), `CircadianSceneReconciler` (generate the smooth circadian cycle), and `NightMotionReconciler` (night soft-red guidance). `Change.change_type` is one of `CREATE`, `UPDATE`, `NOOP`, `BLOCKED`. |
 
 ### I/O layer
@@ -202,9 +219,9 @@ The codebase is split into a **pure decision layer** (fully unit-tested, no I/O)
 | `client.py` | CLIP API v2 HTTP client (`HueClient`). |
 | `pin.py` | Trust-on-first-use TLS pinning (SHA-256 of bridge cert → `.hue-pin.json`). |
 | `state.py` | Loads live bridge state; resolves resource names → ids (`BridgeState`), including MotionAware areas/services. |
-| `circadian_daemon.py` | The resident daemon (`CircadianDaemon`): 60s curve ticks + SSE event loop + TV-bias triggers (probe thread / SSE / control files) + security show, all serialised under one lock. |
+| `circadian_daemon.py` | The resident daemon (`CircadianDaemon`): 60s curve ticks + SSE event loop + TV-bias triggers (probe thread / SSE / control files) + security show + rhythm-engine ticks, all serialised under one lock. |
 | `watch.py` | **Legacy** SSE event loop (`MotionController`) for `motion_policies`. Its echo-buffer override detection predates the bridge's periodic re-emission of settled values (the daemon's settle-and-compare replaced it), and it requires legacy PIR sensors. |
-| `cli.py` | `argparse`-based CLI (`Cli`). Subcommands: `auth`, `validate`, `preview`, `inventory`, `plan`, `apply`, `circadian run\|resume`, `security on\|off\|status`, `watch`. |
+| `cli.py` | `argparse`-based CLI (`Cli`). Subcommands: `auth`, `validate`, `inventory`, `preview`, `plan`, `apply`, `watch`, `circadian run\|resume`, `rhythm`, `security on\|off\|status`. |
 
 ### Data flow
 
diff --git a/README.md b/README.md
index 7f0ded7..3ab2933 100644
--- a/README.md
+++ b/README.md
@@ -81,6 +81,7 @@ hueman -c my-home.yaml apply            # converge the bridge to the declared st
 hueman -c my-home.yaml circadian run    # run the circadian + TV-bias daemon (foreground)
 hueman -c my-home.yaml circadian resume # clear a manual-override suspension out-of-band
 hueman -c my-home.yaml security on|off  # arm / disarm the panic show
+hueman -c my-home.yaml rhythm           # print the rhythm engine's phase, anchors, evidence
 hueman -c my-home.yaml watch            # legacy live motion controller (see Limitations)
 ```
 
@@ -203,6 +204,83 @@ Key sections:
   photosensitivity floor on luminance flashing, arm/disarm triggers, and a
   decoupled sound-cue contract for an external audio adapter.
 - `motion_policies` — per-sensor policies for the legacy `watch` runtime.
+- `rhythm` — closed-loop day-phase inference (observe stage); see below.
+
+## Rhythm engine (observe stage)
+
+`rhythm:` runs a closed-loop day-phase inference engine inside the circadian
+daemon: it watches MotionAware motion (in your bedroom and elsewhere) plus
+optional phone signals, infers which phase of your day you're in, logs every
+inference with its evidence, and learns your real wake/bed-time anchors over
+time.
+
+**Stage 1 ("observe") never writes to the bridge.** It's a read-only shadow
+layer — nothing it infers changes what your lights do. The config accepts
+`stage: "mornings"` and `stage: "full"` so a config can be staged ahead of the
+code, but the daemon deliberately refuses to start with either; only
+`stage: "observe"` is implemented.
+
+**Phase model.** Phases move forward only, one step per tick:
+`dawn → morning → daylight → evening → wind_down → night → sleep`, with two
+event-driven exceptions:
+- The **sleep vote** can cut `wind_down`/`night` straight to `sleep` once the
+  house has been quiet for `presence.quiet`, the TV and the driven zone are
+  both off, and either the bedroom was the last active room or the phone is
+  charging.
+- **Confirmed wake evidence** ends `dawn`/`sleep` and starts `morning`: enough
+  motion (`presence.wake_confirm_events` events, or 2+ distinct rooms) within
+  `presence.wake_confirm_window`, plus the bedroom being involved or a light
+  change — so a hallway cat patrol can't read as "awake." A missed wake
+  (alarm passed, no motion) force-advances to `morning` after two hours as a
+  failsafe.
+
+`wind_down` starts `wind_down_lead` before `bed_target`; `dawn` starts
+`dawn_lead` before the wake anchor, which resolves in order: a phone alarm due
+within 18h, else your learned median wake time for weekday/weekend (weekend
+capped at `weekend_drift_cap` past weekday), else `wake_default`.
+
+**Signal files.** Two optional files, written by an external automation (e.g.
+Home Assistant) and re-read every tick — the engine never consumes or deletes
+them:
+- `signals.next_alarm_file` — epoch seconds of the phone's next alarm
+  (`0`/empty = none set).
+- `signals.charging_file` — its mere *existence* means the phone is on the
+  charger (a bedtime proxy).
+
+Either can be unset or unreadable; the engine just falls back to learned
+anchors or config defaults.
+
+**Pet discounting.** MotionAware areas report presence, not identity. In a
+one-human-plus-pets home: a light change always counts as human (pets don't
+use switches); motion counts as human when a *different* room was active
+within `presence.pet_progression` minutes (room-to-room movement); solo
+single-room motion is otherwise discounted as a pet. Known blind spot: a human
+sitting nearly still in one room for a long stretch degrades to "pet"
+judgments — but the sleep vote that consumes this signal also requires
+lights-out and TV-off, so a reader with a lamp on is never mistaken for an
+empty house.
+
+**Checking on it:**
+
+```
+hueman -c my-home.yaml rhythm
+```
+
+prints the current phase and when it was last read, today's bed/wake
+anchors, the learned weekday/weekend wake and sleep-onset medians, how far
+the learned weekday sleep-onset has drifted from `bed_target` (once at least
+one onset has been observed), and the JSON evidence behind the most recent
+phase change.
+
+**The state file** (`state_file`, default `rhythm-state.json`) holds the
+learned anchors plus the latest snapshot, written atomically after every
+phase change. Delete it to reset learning — the engine falls back to config
+defaults until it re-learns.
+
+**Debugging.** Every inference is logged with a `rhythm:` prefix — grep the
+daemon log for it to see phase transitions with full evidence, manual
+overrides fed in as human activity, and (at debug level) per-event motion
+judgments.
 
 ## What needs which hardware
 
@@ -210,14 +288,16 @@ Key sections:
   security mode work on any **CLIP v2** bridge (square Hue bridge or Bridge Pro).
 - `night_motion` and motion-area inventory need a **Hue Bridge Pro** running
   **MotionAware** (motion sensed by the bulb grid — no PIR sensors required).
+  `rhythm` also wants MotionAware for its motion evidence; without it, presence
+  inference sees only manual-override activity (still functional, just blinder).
 - The legacy `watch` runtime needs old-style PIR **Hue motion sensors**.
 
 ## Limitations and notes
 
 - The pure decision layer (config, sun, circadian, daemon controller, bias,
-  security, engine, reconcile, nightmotion) is covered by 300+ unit tests.
-  `client`, `pin`, and the daemon's I/O shell talk to a real bridge and should
-  be exercised on your network.
+  security, engine, reconcile, nightmotion, presence, rhythm) is covered by
+  300+ unit tests. `client`, `pin`, and the daemon's I/O shell talk to a real
+  bridge and should be exercised on your network.
 - **`watch` is legacy and known-inadequate against current bridge firmware**:
   its echo-buffer override detection predates the bridge's periodic re-emission
   of settled `grouped_light` values, which it can misread as manual overrides,
@@ -252,11 +332,13 @@ unit-tested; the network surface is thin and replaceable.
 | `engine.py` | Pure motion/timing state machine for `watch` (`PolicyEngine`) |
 | `payload.py` | Decision output → CLIP request bodies, incl. sRGB → CIE xy |
 | `nightmotion.py` | Pure night-motion/scene helpers: scene bodies, automation transform, tolerant scene-look diff |
+| `presence.py` | Pure activity judging for `rhythm`: pet-discounting rules, quiet/confirm-window summaries (`PresenceTracker`) |
+| `rhythm_control.py` | Pure day-phase state machine for `rhythm` (observe stage): phases, sleep/wake votes, learned anchors (`RhythmEngine`, `AnchorStore`) |
 | `reconcile.py` | Terraform-style plan/apply: `Planner` + area, sensitivity, smart-scene, circadian-scene, and night-motion reconcilers |
 | `state.py` | Index the live bridge (incl. MotionAware areas); resolve names → ids |
 | `client.py` | CLIP API v2 client |
 | `pin.py` | Trust-on-first-use TLS certificate pinning |
-| `circadian_daemon.py` | The resident daemon: curve ticks, SSE events, TV-bias triggers, security show |
+| `circadian_daemon.py` | The resident daemon: curve ticks, SSE events, TV-bias triggers, security show, rhythm-engine ticks |
 | `watch.py` | Legacy live runtime: bridge event stream → `PolicyEngine` → commands |
 | `cli.py` | The `hueman` command-line interface |
 
diff --git a/examples/home.yaml b/examples/home.yaml
index fbfd365..b30b270 100644
--- a/examples/home.yaml
+++ b/examples/home.yaml
@@ -141,6 +141,31 @@ night_motion:
     hex: "#ff1400"                   # deep red, easy on night-adjusted eyes
     brightness: 3
 
+# ---------------------------------------------------------------------------
+# Closed-loop day-rhythm inference (stage 1: observe-only — no light control).
+# The daemon infers dawn/morning/.../sleep from MotionAware areas plus optional
+# phone signals, logs every inference with evidence, and learns wake/bed anchors.
+# Inspect with `hueman rhythm`. See README "Rhythm engine (observe stage)".
+# ---------------------------------------------------------------------------
+rhythm:
+  stage: "observe"
+  bedroom: "Bedroom"                 # room name whose motion anchors sleep/wake
+  bed_target: "23:00"
+  wake_default: "07:00"
+  weekend_drift_cap: "90m"
+  state_file: "/data/rhythm-state.json"
+  signals:
+    # Written by an external automation (e.g. Home Assistant):
+    # next_alarm_file contains the epoch seconds of the phone's next alarm;
+    # charging_file's existence means the phone is on the charger.
+    next_alarm_file: "/mnt/signals/.next-alarm"
+    charging_file: "/mnt/signals/.phone-charging"
+  presence:
+    quiet: "30m"
+    wake_confirm_events: 3
+    wake_confirm_window: "10m"
+    pet_progression: "5m"
+
 # ---------------------------------------------------------------------------
 # Legacy `watch` runtime (per-PIR-sensor motion policies). Predates the daemon
 # and MotionAware — see README "Limitations" before relying on it.
diff --git a/hueman/cli.py b/hueman/cli.py
index 21a6fcd..89e4a4a 100644
--- a/hueman/cli.py
+++ b/hueman/cli.py
@@ -3,10 +3,17 @@
 Subcommands:
     auth      Pair with the bridge (press the link button) and print the key.
     validate  Parse and validate the config without touching the bridge.
+    inventory List the bridge's rooms, zones, lights and sensors.
     preview   Print the circadian colour curve for a date and location.
     plan      Show the changes apply would make (read-only).
     apply     Converge the bridge's declarative state to the config.
-    watch     Run the live motion/timing controller.
+    watch     Run the legacy live motion/timing controller.
+    circadian run|resume
+              Run the resident circadian daemon, or hand control back to a
+              running one after a manual-override suspension.
+    rhythm    Print the rhythm engine's phase, anchors, and evidence.
+    security on|off|status
+              Arm/disarm the daemon-native panic mode, or show its config.
 
 The interface mirrors Terraform's verbs deliberately: ``validate`` -> ``plan``
 -> ``apply`` is the same muscle memory, and ``plan`` never mutates anything.

From dbd9ee7d9b84d86c0088a87877f5bf325595be6b Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 02:44:16 -0700
Subject: [PATCH 13/15] =?UTF-8?q?docs(rhythm):=20accuracy=20fixes=20?=
 =?UTF-8?q?=E2=80=94=20logging=20granularity,=20fourth=20pet=20rule,=20spe?=
 =?UTF-8?q?c=20link?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 CLAUDE.md          | 17 +++++++++--------
 README.md          | 19 +++++++++++--------
 examples/home.yaml |  2 +-
 hueman/cli.py      |  4 ++--
 4 files changed, 23 insertions(+), 19 deletions(-)

diff --git a/CLAUDE.md b/CLAUDE.md
index 715087a..4fc67e2 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -182,13 +182,14 @@ day-phase inference engine (`rhythm_control.py` + `presence.py`) inside the
 daemon: it infers `dawn`/`morning`/`daylight`/`evening`/`wind_down`/`night`/
 `sleep` from MotionAware motion (with pet discounting) plus optional phone
 signals, and learns wake/bed-time anchors over time. **Stage 1 (`stage:
-"observe"`) is read-only — it never writes to the bridge**; every inference is
-logged with a `rhythm:` prefix in the daemon log and persisted to
-`rhythm.state_file` (the path comes from config; delete it to reset
-learning). `stage: "mornings"` and `stage: "full"` parse but the daemon
-deliberately refuses to start with either — only `observe` ships in stage 1.
-See `docs/superpowers/specs/2026-07-04-rhythm-engine-design.md` and the
-README's "Rhythm engine (observe stage)" section.
+"observe"`) is read-only — it never writes to the bridge.** Evidence lines
+carry a `rhythm:` prefix in the daemon log: every phase *change* logs at INFO
+with its full evidence dict and rewrites `rhythm.state_file` (the path comes
+from config; delete it to reset learning); individual motion judgments appear
+at DEBUG; unchanged "hold" ticks are silent. `stage: "mornings"` and `stage:
+"full"` parse but the daemon deliberately refuses to start with either — only
+`observe` ships in stage 1. See the README's "Rhythm engine (observe stage)"
+section (the design spec lives in the deployment/ops repository, not here).
 
 ## Architecture
 
@@ -221,7 +222,7 @@ The codebase is split into a **pure decision layer** (fully unit-tested, no I/O)
 | `state.py` | Loads live bridge state; resolves resource names → ids (`BridgeState`), including MotionAware areas/services. |
 | `circadian_daemon.py` | The resident daemon (`CircadianDaemon`): 60s curve ticks + SSE event loop + TV-bias triggers (probe thread / SSE / control files) + security show + rhythm-engine ticks, all serialised under one lock. |
 | `watch.py` | **Legacy** SSE event loop (`MotionController`) for `motion_policies`. Its echo-buffer override detection predates the bridge's periodic re-emission of settled values (the daemon's settle-and-compare replaced it), and it requires legacy PIR sensors. |
-| `cli.py` | `argparse`-based CLI (`Cli`). Subcommands: `auth`, `validate`, `inventory`, `preview`, `plan`, `apply`, `watch`, `circadian run\|resume`, `rhythm`, `security on\|off\|status`. |
+| `cli.py` | `argparse`-based CLI (`Cli`). Subcommands: `validate`, `auth`, `inventory`, `plan`, `apply`, `preview`, `watch`, `circadian run\|resume`, `rhythm`, `security on\|off\|status`. |
 
 ### Data flow
 
diff --git a/README.md b/README.md
index 3ab2933..662ae6f 100644
--- a/README.md
+++ b/README.md
@@ -211,8 +211,8 @@ Key sections:
 `rhythm:` runs a closed-loop day-phase inference engine inside the circadian
 daemon: it watches MotionAware motion (in your bedroom and elsewhere) plus
 optional phone signals, infers which phase of your day you're in, logs every
-inference with its evidence, and learns your real wake/bed-time anchors over
-time.
+phase change with its evidence, and learns your real wake/bed-time anchors
+over time.
 
 **Stage 1 ("observe") never writes to the bridge.** It's a read-only shadow
 layer — nothing it infers changes what your lights do. The config accepts
@@ -253,8 +253,10 @@ anchors or config defaults.
 **Pet discounting.** MotionAware areas report presence, not identity. In a
 one-human-plus-pets home: a light change always counts as human (pets don't
 use switches); motion counts as human when a *different* room was active
-within `presence.pet_progression` minutes (room-to-room movement); solo
-single-room motion is otherwise discounted as a pet. Known blind spot: a human
+within `presence.pet_progression` minutes (room-to-room movement) **or** when
+any light change happened within that same window (a human is clearly up and
+about); solo single-room motion is otherwise discounted as a pet. Known blind
+spot: a human
 sitting nearly still in one room for a long stretch degrades to "pet"
 judgments — but the sleep vote that consumes this signal also requires
 lights-out and TV-off, so a reader with a lamp on is never mistaken for an
@@ -277,10 +279,11 @@ learned anchors plus the latest snapshot, written atomically after every
 phase change. Delete it to reset learning — the engine falls back to config
 defaults until it re-learns.
 
-**Debugging.** Every inference is logged with a `rhythm:` prefix — grep the
-daemon log for it to see phase transitions with full evidence, manual
-overrides fed in as human activity, and (at debug level) per-event motion
-judgments.
+**Debugging.** Evidence lines carry a `rhythm:` prefix — grep the daemon log
+for it. Every phase *change* is logged at INFO with its full evidence dict
+(and rewrites the state file); manual overrides fed in as human activity also
+log at INFO; individual per-event motion judgments appear at DEBUG; quiet
+"hold" ticks that change nothing are silent.
 
 ## What needs which hardware
 
diff --git a/examples/home.yaml b/examples/home.yaml
index b30b270..37aee5a 100644
--- a/examples/home.yaml
+++ b/examples/home.yaml
@@ -144,7 +144,7 @@ night_motion:
 # ---------------------------------------------------------------------------
 # Closed-loop day-rhythm inference (stage 1: observe-only — no light control).
 # The daemon infers dawn/morning/.../sleep from MotionAware areas plus optional
-# phone signals, logs every inference with evidence, and learns wake/bed anchors.
+# phone signals, logs every phase change with evidence, and learns wake/bed anchors.
 # Inspect with `hueman rhythm`. See README "Rhythm engine (observe stage)".
 # ---------------------------------------------------------------------------
 rhythm:
diff --git a/hueman/cli.py b/hueman/cli.py
index 89e4a4a..e34e324 100644
--- a/hueman/cli.py
+++ b/hueman/cli.py
@@ -1,12 +1,12 @@
 """Command-line interface for hueman.
 
 Subcommands:
-    auth      Pair with the bridge (press the link button) and print the key.
     validate  Parse and validate the config without touching the bridge.
+    auth      Pair with the bridge (press the link button) and print the key.
     inventory List the bridge's rooms, zones, lights and sensors.
-    preview   Print the circadian colour curve for a date and location.
     plan      Show the changes apply would make (read-only).
     apply     Converge the bridge's declarative state to the config.
+    preview   Print the circadian colour curve for a date and location.
     watch     Run the legacy live motion/timing controller.
     circadian run|resume
               Run the resident circadian daemon, or hand control back to a

From 00b7b52ca4fcf2da8897fc86609e6d4d64da979a Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 02:57:50 -0700
Subject: [PATCH 14/15] fix(rhythm): restart-safe seeding, human-seen sleep
 gate, night wake escape, anchor day-class

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 README.md                   |  8 +++-
 hueman/circadian_daemon.py  |  3 +-
 hueman/rhythm_control.py    | 44 ++++++++++++++----
 tests/test_rhythm_engine.py | 89 +++++++++++++++++++++++++++++++++++++
 4 files changed, 133 insertions(+), 11 deletions(-)

diff --git a/README.md b/README.md
index 662ae6f..56ec831 100644
--- a/README.md
+++ b/README.md
@@ -221,18 +221,22 @@ code, but the daemon deliberately refuses to start with either; only
 `stage: "observe"` is implemented.
 
 **Phase model.** Phases move forward only, one step per tick:
-`dawn → morning → daylight → evening → wind_down → night → sleep`, with two
+`dawn → morning → daylight → evening → wind_down → night → sleep`, with three
 event-driven exceptions:
 - The **sleep vote** can cut `wind_down`/`night` straight to `sleep` once the
   house has been quiet for `presence.quiet`, the TV and the driven zone are
   both off, and either the bedroom was the last active room or the phone is
-  charging.
+  charging. The vote never passes before *any* human activity has been
+  observed, so a daemon restart during actual sleep can't fabricate an onset.
 - **Confirmed wake evidence** ends `dawn`/`sleep` and starts `morning`: enough
   motion (`presence.wake_confirm_events` events, or 2+ distinct rooms) within
   `presence.wake_confirm_window`, plus the bedroom being involved or a light
   change — so a hallway cat patrol can't read as "awake." A missed wake
   (alarm passed, no motion) force-advances to `morning` after two hours as a
   failsafe.
+- After midnight, the same wake evidence also ends `night` and starts
+  `morning` — the escape hatch for a restart during sleep, which seeds
+  `night` and (per the human-seen gate above) can never reach `sleep`.
 
 `wind_down` starts `wind_down_lead` before `bed_target`; `dawn` starts
 `dawn_lead` before the wake anchor, which resolves in order: a phone alarm due
diff --git a/hueman/circadian_daemon.py b/hueman/circadian_daemon.py
index 4aec7b9..3b7592b 100644
--- a/hueman/circadian_daemon.py
+++ b/hueman/circadian_daemon.py
@@ -669,7 +669,8 @@ def _read_alarm_epoch(self) -> float | None:
         if rhythm is None or not rhythm.signals.next_alarm_file:
             return None
         try:
-            raw = open(rhythm.signals.next_alarm_file).read().strip()
+            with open(rhythm.signals.next_alarm_file) as fh:
+                raw = fh.read().strip()
             value = float(raw)
             return value if value > 0 else None
         except (OSError, ValueError):
diff --git a/hueman/rhythm_control.py b/hueman/rhythm_control.py
index d40a4a4..f3512e7 100644
--- a/hueman/rhythm_control.py
+++ b/hueman/rhythm_control.py
@@ -13,7 +13,7 @@
 from __future__ import annotations
 
 import datetime as _dt
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from statistics import median
 from typing import Any
 from zoneinfo import ZoneInfo
@@ -136,9 +136,13 @@ class RhythmEngine:
 
     Phases move forward only (one step per tick) through
     ``dawn → morning → daylight → evening → wind_down → night → sleep`` with
-    two event-driven exceptions: the sleep vote can end an evening early
-    (``wind_down``/``night`` → ``sleep``) and confirmed wake evidence ends
-    ``dawn``/``sleep`` (→ ``morning``). All thresholds come from the spec;
+    three event-driven exceptions: the sleep vote can end an evening early
+    (``wind_down``/``night`` → ``sleep``); confirmed wake evidence ends
+    ``dawn``/``sleep`` (→ ``morning``); and after midnight the same wake
+    evidence also ends ``night`` (→ ``morning``) — the escape hatch for a
+    restart during actual sleep, which seeds ``night`` and (with the
+    human-seen gate on the sleep vote) can never reach ``sleep``. All
+    thresholds come from the spec;
     learned anchors refine them but never move outside spec bounds.
 
     Args:
@@ -198,6 +202,11 @@ def _wake_day_class(self, local: _dt.datetime) -> str:
         """Day class of a wake observation (the date it happened)."""
         return "weekend" if local.weekday() >= 5 else "weekday"
 
+    def _anchor_day_class(self, local: _dt.datetime) -> str:
+        """Day class of the *upcoming* wake: today before noon, else tomorrow."""
+        date = local.date() if local.hour < 12 else local.date() + _dt.timedelta(days=1)
+        return "weekend" if date.weekday() >= 5 else "weekday"
+
     # -- anchors ----------------------------------------------------------- #
     def _wake_anchor_min(self, now: float, signals: SignalState) -> tuple[float, str]:
         """Tomorrow-or-today's wake anchor minute and its source label.
@@ -210,7 +219,7 @@ def _wake_anchor_min(self, now: float, signals: SignalState) -> tuple[float, str
             alarm_local = self._local(signals.next_alarm_epoch)
             return self._minute(alarm_local), "alarm"
         local = self._local(now)
-        day_class = self._day_class(local)
+        day_class = self._anchor_day_class(local)
         learned = self._store.median("wake", day_class)
         if learned is not None:
             if day_class == "weekend":
@@ -223,13 +232,20 @@ def _wake_anchor_min(self, now: float, signals: SignalState) -> tuple[float, str
 
     # -- votes ------------------------------------------------------------- #
     def _sleep_vote(self, presence: PresenceSummary, signals: SignalState) -> tuple[bool, dict[str, Any]]:
-        """The sleep-onset confidence vote and its evidence."""
+        """The sleep-onset confidence vote and its evidence.
+
+        ``human_seen`` gates cold starts: before any human-judged activity has
+        ever been observed (a daemon restart during actual sleep), the vote
+        must never pass — otherwise the restart minute would be recorded as a
+        fabricated sleep onset.
+        """
         checks = {
             "quiet": presence.quiet_s >= self._spec.presence.quiet_min * 60.0,
             "tv_off": not signals.tv_on,
             "zone_off": not signals.zone_on,
             "bedroom_or_charging": (
                 presence.last_active_room == self._spec.bedroom or signals.phone_charging),
+            "human_seen": presence.last_active_room is not None,
         }
         return all(checks.values()), {"sleep_vote": checks, "quiet_s": round(presence.quiet_s)}
 
@@ -285,14 +301,16 @@ def _decide(
         m_shift = _noon_shifted(minute)
 
         if self._phase is None:  # first tick: seed from wall clock
+            if minute < wake_anchor:
+                return self.NIGHT, "seed"          # pre-dawn: assume night
+            if minute < 720:
+                return self.DAYLIGHT, "seed"       # morning half, past wake
             if m_shift >= night_start:
                 return self.NIGHT, "seed"
             if m_shift >= wind_down_start:
                 return self.WIND_DOWN, "seed"
             if m_shift >= _noon_shifted(signals.sunset_min):
                 return self.EVENING, "seed"
-            if minute < wake_anchor:
-                return self.NIGHT, "seed"          # pre-dawn restart: assume night
             return self.DAYLIGHT, "seed"
 
         if self._phase == self.SLEEP:
@@ -345,6 +363,16 @@ def _decide(
             if asleep:
                 self._record_sleep_onset(local, minute)
                 return self.SLEEP, "sleep-vote"
+            if self._phase == self.NIGHT and minute < 720:
+                # Morning escape: a restart during actual sleep seeds NIGHT
+                # and (human-seen gate) can never reach SLEEP, so NIGHT must
+                # hand off to MORNING itself when the human gets up.
+                woke, wake_ev = self._wake_evidence(presence)
+                evidence.update(wake_ev)
+                if woke:
+                    self._record_wake(local, minute)
+                    self._morning_started_min = minute
+                    return self.MORNING, "wake-detected"
             if self._phase == self.WIND_DOWN and m_shift >= night_start:
                 return self.NIGHT, "bed-anchor"
             return self._phase, "hold"
diff --git a/tests/test_rhythm_engine.py b/tests/test_rhythm_engine.py
index b25cb17..bdf4935 100644
--- a/tests/test_rhythm_engine.py
+++ b/tests/test_rhythm_engine.py
@@ -30,6 +30,11 @@ def _epoch(day, hh, mm):
                                 recent_rooms=("Bedroom",),
                                 recent_motion_count=3,
                                 recent_light_change=False)
+#: What PresenceTracker.summary() yields right after a daemon restart:
+#: no human activity ever judged.
+COLD = PresenceSummary(quiet_s=1e9, last_active_room=None,
+                       recent_rooms=(), recent_motion_count=0,
+                       recent_light_change=False)
 
 
 def _sig(alarm=None, charging=False, tv=False, zone_on=True, sunset=20 * 60 + 30):
@@ -141,6 +146,90 @@ def test_weekend_learned_wake_is_capped_by_drift_rule():
     assert d.evidence["wake_anchor_src"] == "learned-weekend"
 
 
+def test_morning_restart_seeds_daylight_not_night():
+    """A daemon restart mid-morning must not strand the day in NIGHT."""
+    eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+    d = eng.tick(_epoch(1, 9, 30), ACTIVE, _sig())
+    assert d.phase == RhythmEngine.DAYLIGHT and d.reason == "seed"
+
+
+def test_seed_hour_sweep_is_sane():
+    """Seeding at every hour of a Wednesday lands in a sane phase.
+
+    Defaults: wake anchor 07:00, sunset fixture 20:30, wind-down 21:30
+    (bed 23:00 - 90m lead), night 23:00.
+    """
+    expected = {}
+    for h in range(24):
+        if h < 7:
+            expected[h] = RhythmEngine.NIGHT
+        elif h < 21:
+            expected[h] = RhythmEngine.DAYLIGHT
+        elif h == 21:
+            expected[h] = RhythmEngine.EVENING
+        elif h == 22:
+            expected[h] = RhythmEngine.WIND_DOWN
+        else:
+            expected[h] = RhythmEngine.NIGHT
+    for h in range(24):
+        eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)
+        d = eng.tick(_epoch(1, h, 0), QUIET, _sig(zone_on=False))
+        assert d.phase == expected[h], f"hour {h}: {d.phase} != {expected[h]}"
+
+
+def test_cold_start_with_charging_records_no_onset():
+    """A restart during actual sleep must not fabricate a sleep onset.
+
+    Cold presence (no human event ever judged) + phone on charger + zone off
+    would pass the old vote; the human-seen gate blocks it.
+    """
+    store = AnchorStore()
+    eng = RhythmEngine(_spec(), store, tz=TZ)
+    d = eng.tick(_epoch(1, 6, 0), COLD, _sig(charging=True, zone_on=False))
+    assert d.phase == RhythmEngine.NIGHT          # seeded pre-wake-anchor
+    d = eng.tick(_epoch(1, 6, 1), COLD, _sig(charging=True, zone_on=False))
+    assert d.phase == RhythmEngine.NIGHT
+    assert store.median("sleep_onset", "weekday") is None
+    assert store.median("sleep_onset", "weekend") is None
+
+
+def test_overnight_restart_wakes_from_night():
+    """Seeded NIGHT (restart during sleep) hands off to MORNING on wake."""
+    store = AnchorStore()
+    eng = RhythmEngine(_spec(), store, tz=TZ)
+    d = eng.tick(_epoch(1, 2, 0), QUIET, _sig(zone_on=False))
+    assert d.phase == RhythmEngine.NIGHT
+    d = eng.tick(_epoch(1, 6, 50), BEDROOM_BURST, _sig(zone_on=False))
+    assert d.phase == RhythmEngine.MORNING and d.reason == "wake-detected"
+    assert store.median("wake", "weekday") == 6 * 60 + 50
+
+
+def test_anchor_day_class_boundaries():
+    """The wake anchor is classed by the upcoming morning, not the onset rule.
+
+    Friday 06:00 -> today's (weekday) wake; Sunday 06:00 -> weekend; Friday
+    22:00 -> Saturday morning, so weekend. 2026-07-03 is a Friday.
+    """
+    def _store():
+        s = AnchorStore()
+        s.record("wake", "weekday", 420, "2026-06-29")   # 07:00
+        s.record("wake", "weekend", 500, "2026-06-28")   # 08:20
+        return s
+
+    d = RhythmEngine(_spec(), _store(), tz=TZ).tick(_epoch(3, 6, 0), ACTIVE, _sig())
+    assert d.evidence["wake_anchor_min"] == 420
+    assert d.evidence["wake_anchor_src"] == "learned-weekday"
+
+    d = RhythmEngine(_spec(), _store(), tz=TZ).tick(_epoch(5, 6, 0), ACTIVE, _sig())
+    # cap = weekday 420 + 90 drift = 510; 500 is under the cap -> unclamped
+    assert d.evidence["wake_anchor_min"] == 500
+    assert d.evidence["wake_anchor_src"] == "learned-weekend"
+
+    d = RhythmEngine(_spec(), _store(), tz=TZ).tick(_epoch(3, 22, 0), ACTIVE, _sig())
+    assert d.evidence["wake_anchor_min"] == 500
+    assert d.evidence["wake_anchor_src"] == "learned-weekend"
+
+
 def test_snapshot_is_json_serialisable_and_carries_anchors():
     import json
     eng = RhythmEngine(_spec(), AnchorStore(), tz=TZ)

From dc49d66a03b93c35063d2d3c9a1e7f85eac82206 Mon Sep 17 00:00:00 2001
From: Chance Newkirk <40219357+cnewkirk@users.noreply.github.com>
Date: Sun, 5 Jul 2026 03:03:07 -0700
Subject: [PATCH 15/15] fix(rhythm): night wakings do not count as wake anchors

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 README.md                   | 12 ++++++----
 hueman/rhythm_control.py    | 45 +++++++++++++++++++++++++++++--------
 tests/test_rhythm_engine.py | 23 +++++++++++++++++++
 3 files changed, 67 insertions(+), 13 deletions(-)

diff --git a/README.md b/README.md
index 56ec831..7e9fcf1 100644
--- a/README.md
+++ b/README.md
@@ -231,12 +231,16 @@ event-driven exceptions:
 - **Confirmed wake evidence** ends `dawn`/`sleep` and starts `morning`: enough
   motion (`presence.wake_confirm_events` events, or 2+ distinct rooms) within
   `presence.wake_confirm_window`, plus the bedroom being involved or a light
-  change — so a hallway cat patrol can't read as "awake." A missed wake
+  change — so a hallway cat patrol can't read as "awake." Wake evidence only
+  counts when the clock is plausibly morning (from ~2h before the dawn window
+  opens): a 01:30 bathroom trip is logged as a night waking and ignored, so it
+  can't poison the learned wake anchor. A missed wake
   (alarm passed, no motion) force-advances to `morning` after two hours as a
   failsafe.
-- After midnight, the same wake evidence also ends `night` and starts
-  `morning` — the escape hatch for a restart during sleep, which seeds
-  `night` and (per the human-seen gate above) can never reach `sleep`.
+- After midnight, the same wake evidence (same plausibly-morning gate) also
+  ends `night` and starts `morning` — the escape hatch for a restart during
+  sleep, which seeds `night` and (per the human-seen gate above) can never
+  reach `sleep`.
 
 `wind_down` starts `wind_down_lead` before `bed_target`; `dawn` starts
 `dawn_lead` before the wake anchor, which resolves in order: a phone alarm due
diff --git a/hueman/rhythm_control.py b/hueman/rhythm_control.py
index f3512e7..801e19b 100644
--- a/hueman/rhythm_control.py
+++ b/hueman/rhythm_control.py
@@ -24,6 +24,11 @@
 #: Newest observations kept per (kind, day-class); two weeks of weekdays.
 _MAX_SAMPLES = 14
 
+#: Wake evidence only counts within this many minutes before the dawn window
+#: opens; earlier sustained motion is a night waking (bathroom trip), not a
+#: wake, and must not poison the learned wake anchor.
+_EARLY_WAKE_MARGIN_MIN = 120
+
 
 class AnchorStore:
     """Recent observed wake / sleep-onset minutes, per weekday/weekend class.
@@ -141,8 +146,11 @@ class RhythmEngine:
     ``dawn``/``sleep`` (→ ``morning``); and after midnight the same wake
     evidence also ends ``night`` (→ ``morning``) — the escape hatch for a
     restart during actual sleep, which seeds ``night`` and (with the
-    human-seen gate on the sleep vote) can never reach ``sleep``. All
-    thresholds come from the spec;
+    human-seen gate on the sleep vote) can never reach ``sleep``. In both
+    the ``sleep`` and ``night`` cases the evidence only counts when the
+    minute is plausibly morning (within ``_EARLY_WAKE_MARGIN_MIN`` of the
+    dawn window opening); earlier sustained motion is a *night waking* —
+    logged in the evidence but ignored. All thresholds come from the spec;
     learned anchors refine them but never move outside spec bounds.
 
     Args:
@@ -268,6 +276,17 @@ def _wake_evidence(self, presence: PresenceSummary) -> tuple[bool, dict[str, Any
             "bedroom_involved": self._spec.bedroom in presence.recent_rooms,
         }
 
+    def _wake_plausible(self, minute: float, wake_anchor: float) -> bool:
+        """Whether sustained motion at this minute may count as the day's wake.
+
+        True only in the morning half of the day, no earlier than
+        ``_EARLY_WAKE_MARGIN_MIN`` before the dawn window opens; anything
+        earlier is a night waking (a bathroom trip), not a wake, and must
+        not poison the learned wake anchor.
+        """
+        floor = wake_anchor - self._spec.dawn_lead_min - _EARLY_WAKE_MARGIN_MIN
+        return minute < 720 and minute >= floor
+
     # -- tick ---------------------------------------------------------------- #
     def tick(self, now: float, presence: PresenceSummary, signals: SignalState) -> PhaseDecision:
         """Advance at most one phase; return the decision with evidence."""
@@ -318,9 +337,13 @@ def _decide(
             evidence.update(wake_ev)
             in_dawn_window = wake_anchor - spec.dawn_lead_min <= minute < wake_anchor
             if woke:
-                self._record_wake(local, minute)
-                self._morning_started_min = minute
-                return self.MORNING, "wake-detected"
+                # Sustained motion only counts as the wake when the clock is
+                # plausibly morning; a 01:30 bathroom trip is a night waking.
+                if self._wake_plausible(minute, wake_anchor):
+                    self._record_wake(local, minute)
+                    self._morning_started_min = minute
+                    return self.MORNING, "wake-detected"
+                evidence["night_waking"] = True
             if in_dawn_window:
                 return self.DAWN, "dawn-window"
             return self.SLEEP, "hold"
@@ -366,13 +389,17 @@ def _decide(
             if self._phase == self.NIGHT and minute < 720:
                 # Morning escape: a restart during actual sleep seeds NIGHT
                 # and (human-seen gate) can never reach SLEEP, so NIGHT must
-                # hand off to MORNING itself when the human gets up.
+                # hand off to MORNING itself when the human gets up. Same
+                # plausibly-morning gate as SLEEP: pre-margin sustained
+                # motion is a night waking, not a wake.
                 woke, wake_ev = self._wake_evidence(presence)
                 evidence.update(wake_ev)
                 if woke:
-                    self._record_wake(local, minute)
-                    self._morning_started_min = minute
-                    return self.MORNING, "wake-detected"
+                    if self._wake_plausible(minute, wake_anchor):
+                        self._record_wake(local, minute)
+                        self._morning_started_min = minute
+                        return self.MORNING, "wake-detected"
+                    evidence["night_waking"] = True
             if self._phase == self.WIND_DOWN and m_shift >= night_start:
                 return self.NIGHT, "bed-anchor"
             return self._phase, "hold"
diff --git a/tests/test_rhythm_engine.py b/tests/test_rhythm_engine.py
index bdf4935..1f855cb 100644
--- a/tests/test_rhythm_engine.py
+++ b/tests/test_rhythm_engine.py
@@ -204,6 +204,29 @@ def test_overnight_restart_wakes_from_night():
     assert store.median("wake", "weekday") == 6 * 60 + 50
 
 
+def test_night_bathroom_trip_does_not_count_as_wake():
+    """Sustained 01:30 motion during SLEEP is a night waking, not a wake."""
+    store = AnchorStore()
+    eng = RhythmEngine(_spec(), store, tz=TZ)
+    eng.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    eng.tick(_epoch(1, 23, 50), QUIET, _sig(zone_on=False))      # -> SLEEP
+    d = eng.tick(_epoch(2, 1, 30), BEDROOM_BURST, _sig(zone_on=False))
+    assert d.phase == RhythmEngine.SLEEP
+    assert d.evidence.get("night_waking") is True
+    assert store.median("wake", "weekday") is None
+
+
+def test_early_real_wake_within_margin_still_counts():
+    """05:30 sustained motion with a 07:00 anchor is inside the margin."""
+    store = AnchorStore()
+    eng = RhythmEngine(_spec(), store, tz=TZ)
+    eng.tick(_epoch(1, 23, 1), ACTIVE, _sig())
+    eng.tick(_epoch(1, 23, 50), QUIET, _sig(zone_on=False))      # -> SLEEP
+    d = eng.tick(_epoch(2, 5, 30), BEDROOM_BURST, _sig(zone_on=False))
+    assert d.phase == RhythmEngine.MORNING
+    assert store.median("wake", "weekday") == 330
+
+
 def test_anchor_day_class_boundaries():
     """The wake anchor is classed by the upcoming morning, not the onset rule.