feat(hour/day derivation): hour classification, monitoring periods, day aggregates, pruning

Hour classification (PR-4):
- Powered-off: POH delta < 90% of wall-clock span
- Active: DUW delta >= 256 MiB
- Idle: powered on + sampled + below active threshold
- Unknown: unsampled without POH evidence
- Four splits sum to exactly wall_clock_seconds
- Disabled time is never an hour state

Monitoring periods (FL-8):
- ensure_period_open: opens period at run moment if none exists
- close_period: closes with end cause
- is_inside_period: checks timestamp against period bounds
- Never backdated; wall-clock outside periods excluded from denominator

Day aggregates (ST-4, PR-5):
- Derived monotonically from hour rows
- UTC-bounded; no 23/25-hour days
- Coverage: known seconds / period wall-clock
- Gap hours inside periods contribute unknown seconds
- Hours outside periods excluded entirely
- No absent hour interpolated/estimated/fabricated (FL-3)

Raw sample pruning (ST-5):
- Prunes samples older than 14 days
- Hour observations and day aggregates retained indefinitely

Closes #22

src/fenris/day_aggregate.py

@@ -0,0 +1,165 @@
+"""Day aggregate derivation per spec §5.4, §3.3.
+
+One row per UTC day, derived monotonically from hour rows — the grain at
+which usage-habit evidence is judged.  No absent hour is ever interpolated,
+estimated, or fabricated (§5.3, FL-3).
+
+Coverage: the share of wall-clock seconds inside monitoring periods whose
+usage-habit classification is known rather than unknown (§5.3).
+"""
+import sqlite3
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+
+@dataclass(frozen=True)
+class DayAggregate:
+    """One UTC day's aggregated stats."""
+    day: str  # ISO 8601 UTC date, e.g. "2026-09-01"
+    seconds_active: int
+    seconds_idle: int
+    seconds_powered_off: int
+    seconds_unknown: int
+    bytes_written_delta: int
+    bytes_read_delta: int
+    sample_count: int
+    coverage: float
+
+
+def _period_wall_clock_for_day(conn: sqlite3.Connection, day: str) -> int:
+    """Total wall-clock seconds inside monitoring periods for a UTC day.
+
+    Clamps each period to the day boundary [dayT00:00, dayT24:00).
+    """
+    day_start = datetime.fromisoformat(f"{day}T00:00:00+00:00")
+    day_end = day_start + timedelta(days=1)
+    day_start_str = day_start.isoformat()
+    day_end_str = day_end.isoformat()
+
+    cursor = conn.execute(
+        "SELECT started_at, ended_at FROM monitoring_periods "
+        "WHERE (ended_at IS NULL OR ended_at > ?) AND started_at < ? "
+        "ORDER BY started_at",
+        (day_start_str, day_end_str),
+    )
+
+    total = 0
+    for row in cursor.fetchall():
+        period_start = row[0]
+        period_end = row[1]
+
+        effective_start = max(period_start, day_start_str)
+        if period_end is not None:
+            effective_end = min(period_end, day_end_str)
+        else:
+            effective_end = day_end_str
+
+        if effective_start < effective_end:
+            s = datetime.fromisoformat(effective_start)
+            e = datetime.fromisoformat(effective_end)
+            total += int((e - s).total_seconds())
+
+    return total
+
+
+def _hour_overlaps_period(conn: sqlite3.Connection, hour_iso: str) -> bool:
+    """Check if an hour's wall-clock span overlaps any monitoring period."""
+    hour_start = datetime.fromisoformat(hour_iso)
+    hour_end = hour_start + timedelta(hours=1)
+    hs = hour_start.isoformat()
+    he = hour_end.isoformat()
+
+    cursor = conn.execute(
+        "SELECT 1 FROM monitoring_periods "
+        "WHERE started_at < ? AND (ended_at IS NULL OR ended_at > ?) "
+        "LIMIT 1",
+        (he, hs),
+    )
+    return cursor.fetchone() is not None
+
+
+def derive_day(conn: sqlite3.Connection, day: str) -> DayAggregate | None:
+    """Derive a single day aggregate from its hour rows + monitoring periods.
+
+    Only hours overlapping a monitoring period contribute to the aggregate.
+    Gap hours inside periods contribute unknown seconds.  Hours outside all
+    monitoring periods are excluded entirely (§5.2).
+
+    Returns None if no hours exist for the day.
+    """
+    cursor = conn.execute(
+        "SELECT hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, "
+        "       bytes_written_delta, bytes_read_delta, sample_count "
+        "FROM hour_observations "
+        "WHERE hour LIKE ? "
+        "ORDER BY hour",
+        (day + "T%",),
+    )
+    rows = cursor.fetchall()
+    if not rows:
+        return None
+
+    total_active = 0
+    total_idle = 0
+    total_powered_off = 0
+    total_unknown_from_hours = 0
+    total_bw = 0
+    total_br = 0
+    total_samples = 0
+    total_hour_wall_clock = 0
+
+    for row in rows:
+        # Only count hours overlapping a monitoring period
+        if not _hour_overlaps_period(conn, row[0]):
+            continue
+
+        total_active += row[1]
+        total_idle += row[2]
+        total_powered_off += row[3]
+        total_unknown_from_hours += row[4]
+        total_bw += row[5]
+        total_br += row[6]
+        total_samples += row[7]
+        total_hour_wall_clock += row[1] + row[2] + row[3] + row[4]
+
+    # Wall-clock seconds inside monitoring periods for this day
+    period_wc = _period_wall_clock_for_day(conn, day)
+
+    # Gap seconds = period wall-clock - sum of existing hour wall-clock
+    gap_seconds = max(0, period_wc - total_hour_wall_clock)
+
+    total_unknown = total_unknown_from_hours + gap_seconds
+
+    # Coverage: known seconds / period wall-clock (§5.2, §5.3)
+    known_seconds = total_active + total_idle + total_powered_off
+    coverage = known_seconds / period_wc if period_wc > 0 else 0.0
+
+    return DayAggregate(
+        day=day,
+        seconds_active=total_active,
+        seconds_idle=total_idle,
+        seconds_powered_off=total_powered_off,
+        seconds_unknown=total_unknown,
+        bytes_written_delta=total_bw,
+        bytes_read_delta=total_br,
+        sample_count=total_samples,
+        coverage=coverage,
+    )
+
+
+def derive_all_days(conn: sqlite3.Connection) -> list[DayAggregate]:
+    """Derive day aggregates for all days that have hour rows.
+
+    Returns days sorted by date.
+    """
+    cursor = conn.execute(
+        "SELECT DISTINCT substr(hour, 1, 10) as day FROM hour_observations ORDER BY day"
+    )
+    days = [row[0] for row in cursor.fetchall()]
+
+    results = []
+    for day in days:
+        agg = derive_day(conn, day)
+        if agg is not None:
+            results.append(agg)
+    return results

src/fenris/hour_classify.py

@@ -0,0 +1,94 @@
+"""Hour classification per spec §5.1.
+
+Each UTC hour is classified by named constants, in this order of evidence:
+- Powered-off: power-on-hours delta < 90% of wall-clock span
+- Active: DUW delta >= 256 MiB in the hour
+- Idle: powered on, sampled, below active threshold
+- Unknown: everything else (unsampled without POH evidence)
+
+Four splits sum to exactly wall_clock_seconds. Disabled time is never an
+hour state — it is wall-clock outside monitoring periods (§5.2).
+"""
+from dataclasses import dataclass
+
+# Spec §5.1: Active hour threshold — 256 MiB DUW delta
+ACTIVE_THRESHOLD_BYTES = 256 * 1024 * 1024  # 256 MiB
+
+# Spec §5.1: Powered-off threshold — 90% of wall-clock span
+POWERED_OFF_THRESHOLD_PERCENT = 0.90
+
+
+@dataclass(frozen=True)
+class HourSplit:
+    """Usage-habit split for one UTC hour.  Fields sum to wall_clock_seconds."""
+    seconds_active: int
+    seconds_idle: int
+    seconds_powered_off: int
+    seconds_unknown: int
+
+
+def classify_hour(
+    wall_clock_seconds: int,
+    poh_delta: int,
+    duw_delta: int,
+    dur_delta: int,
+    sampled_seconds: int | None = None,
+) -> HourSplit:
+    """Classify a UTC hour into the four usage-habit states.
+
+    Args:
+        wall_clock_seconds: Total seconds in this hour boundary (3600 for a
+            full hour, less for partial-hours at period edges).
+        poh_delta: Power-on-hours delta since previous sample (in seconds).
+        duw_delta: Data-units-written delta since previous sample (in bytes).
+        dur_delta: Data-units-read delta since previous sample (in bytes).
+        sampled_seconds: Seconds within this hour covered by a sample.
+            None or 0 means no sample fell in this hour.
+
+    Returns:
+        HourSplit whose four fields sum to wall_clock_seconds.
+    """
+    if sampled_seconds is None:
+        sampled_seconds = 0
+
+    # Clamp sampled_seconds to wall_clock_seconds
+    sampled_seconds = min(sampled_seconds, wall_clock_seconds)
+
+    # --- Decision order per spec §5.1 ---
+
+    # 1. Powered-off: POH delta < 90% of wall-clock span
+    powered_off_threshold = wall_clock_seconds * POWERED_OFF_THRESHOLD_PERCENT
+    if poh_delta < powered_off_threshold:
+        return HourSplit(
+            seconds_active=0,
+            seconds_idle=0,
+            seconds_powered_off=wall_clock_seconds,
+            seconds_unknown=0,
+        )
+
+    # 2. Active: DUW delta >= 256 MiB
+    if duw_delta >= ACTIVE_THRESHOLD_BYTES:
+        return HourSplit(
+            seconds_active=wall_clock_seconds,
+            seconds_idle=0,
+            seconds_powered_off=0,
+            seconds_unknown=0,
+        )
+
+    # 3. Idle: powered on, sampled, below active threshold
+    #    Unsampled portion within the hour is unknown
+    if sampled_seconds > 0:
+        return HourSplit(
+            seconds_active=0,
+            seconds_idle=sampled_seconds,
+            seconds_powered_off=0,
+            seconds_unknown=wall_clock_seconds - sampled_seconds,
+        )
+
+    # 4. Unknown: unsampled without POH evidence
+    return HourSplit(
+        seconds_active=0,
+        seconds_idle=0,
+        seconds_powered_off=0,
+        seconds_unknown=wall_clock_seconds,
+    )

src/fenris/monitoring_periods.py

@@ -0,0 +1,124 @@
+"""Monitoring period bookkeeping per spec §5.2, §8.6, §9.8.
+
+A monitoring period is a span during which Fenris monitoring is enabled.
+Powered-off time stays inside a period; deliberately disabled time does not.
+
+Key contracts:
+- Run finding no open period opens one at the run moment, never backdated (§9.8)
+- Wall-clock outside periods excluded from numerator and denominator (§5.2)
+- End causes: user_disabled, migrated, unknown_gap
+"""
+import sqlite3
+from datetime import datetime
+
+
+def ensure_period_open(conn: sqlite3.Connection, run_time: datetime) -> None:
+    """Ensure a monitoring period is open.  If none exists, open one at run_time.
+
+    Spec §9.8: A collection run finding no open monitoring period opens one
+    at the run moment, never backdated.
+    """
+    if get_open_period(conn) is not None:
+        return  # Already open — no-op
+
+    ts = run_time.isoformat()
+    conn.execute(
+        "INSERT INTO monitoring_periods (started_at) VALUES (?)",
+        (ts,),
+    )
+    conn.commit()
+
+
+def close_period(
+    conn: sqlite3.Connection,
+    closed_at: datetime,
+    end_cause: str,
+) -> None:
+    """Close the current open monitoring period.
+
+    Spec §8.6: Pause with an open period closes it user_disabled.
+    If no period is open, this is a no-op (pause otherwise).
+    """
+    open_period = get_open_period(conn)
+    if open_period is None:
+        return  # No-op
+
+    ts = closed_at.isoformat()
+    conn.execute(
+        "UPDATE monitoring_periods SET ended_at = ?, end_cause = ? WHERE id = ?",
+        (ts, end_cause, open_period["id"]),
+    )
+    conn.commit()
+
+
+def get_open_period(conn: sqlite3.Connection) -> dict | None:
+    """Return the currently open monitoring period, or None."""
+    cursor = conn.execute(
+        "SELECT id, started_at, ended_at, end_cause "
+        "FROM monitoring_periods WHERE ended_at IS NULL LIMIT 1"
+    )
+    row = cursor.fetchone()
+    if row is None:
+        return None
+    return {
+        "id": row[0],
+        "started_at": row[1],
+        "ended_at": row[2],
+        "end_cause": row[3],
+    }
+
+
+def is_inside_period(conn: sqlite3.Connection, ts: datetime) -> bool:
+    """Check if a timestamp falls inside any monitoring period.
+
+    Spec §5.2: Wall-clock outside periods is excluded from numerator/denominator.
+    """
+    ts_str = ts.isoformat()
+    cursor = conn.execute(
+        "SELECT 1 FROM monitoring_periods "
+        "WHERE started_at <= ? AND (ended_at IS NULL OR ended_at > ?) "
+        "LIMIT 1",
+        (ts_str, ts_str),
+    )
+    return cursor.fetchone() is not None
+
+
+def wall_clock_in_periods(
+    conn: sqlite3.Connection,
+    start: datetime,
+    end: datetime,
+) -> int:
+    """Compute total wall-clock seconds between start and end that fall inside
+    any monitoring period.
+
+    Used for denominator computation (§5.2).
+    """
+    start_str = start.isoformat()
+    end_str = end.isoformat()
+
+    cursor = conn.execute(
+        "SELECT started_at, ended_at FROM monitoring_periods "
+        "WHERE ended_at IS NULL OR ended_at > ? "
+        "ORDER BY started_at",
+        (start_str,),
+    )
+
+    total = 0
+    for row in cursor.fetchall():
+        period_start = row[0]
+        period_end = row[1]  # None if open
+
+        # Clip period to [start, end]
+        effective_start = max(period_start, start_str)
+        if period_end is not None:
+            effective_end = min(period_end, end_str)
+        else:
+            effective_end = end_str
+
+        if effective_start < effective_end:
+            # Parse for arithmetic
+            s = datetime.fromisoformat(effective_start)
+            e = datetime.fromisoformat(effective_end)
+            total += int((e - s).total_seconds())
+
+    return total

src/fenris/pruning.py

@@ -0,0 +1,31 @@
+"""Raw sample pruning per spec §3.4, ST-5.
+
+Raw samples are pruned opportunistically to 14 days.
+Hour observations and day aggregates are retained indefinitely.
+"""
+import sqlite3
+from datetime import datetime, timedelta, timezone
+
+# Spec §3.4: Raw-sample retention
+RAW_SAMPLE_RETENTION_DAYS = 14
+
+
+def prune_old_samples(
+    conn: sqlite3.Connection,
+    now: datetime,
+    retention_days: int = RAW_SAMPLE_RETENTION_DAYS,
+) -> int:
+    """Remove raw samples older than retention_days.
+
+    Args:
+        conn: Connection to the observation store.
+        now: Current UTC time.
+        retention_days: Number of days to retain (default 14).
+
+    Returns:
+        Number of samples removed.
+    """
+    cutoff = (now - timedelta(days=retention_days)).isoformat()
+    cursor = conn.execute("DELETE FROM samples WHERE ts < ?", (cutoff,))
+    conn.commit()
+    return cursor.rowcount

tests/test_day_aggregates.py

@@ -0,0 +1,234 @@
+"""Day aggregate tests.
+
+From spec §5.4, §3.3, ST-4, ST-5:
+- One row per UTC day, derived monotonically from hour rows
+- No 23/25-hour days (UTC-bounded, DST never applies)
+- Coverage: share of wall-clock seconds inside monitoring periods
+  whose classification is known
+- No absent hour ever interpolated/estimated/fabricated (FL-3)
+"""
+import sqlite3
+import sys
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from fenris.store import init_store
+from fenris.monitoring_periods import ensure_period_open, close_period
+from fenris.day_aggregate import derive_day, derive_all_days, DayAggregate
+from fenris.hour_classify import HourSplit, ACTIVE_THRESHOLD_BYTES
+
+
+@pytest.fixture
+def store_conn(tmp_path: Path):
+    db_path = tmp_path / "test.db"
+    conn = init_store(db_path)
+    yield conn
+    conn.close()
+
+
+def _insert_hour(conn, hour_iso, split, bytes_written_delta=0, bytes_read_delta=0,
+                 sample_count=1, coverage=1.0):
+    conn.execute(
+        "INSERT INTO hour_observations "
+        "(hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, "
+        " bytes_written_delta, bytes_read_delta, sample_count, coverage) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            hour_iso,
+            split.seconds_active,
+            split.seconds_idle,
+            split.seconds_powered_off,
+            split.seconds_unknown,
+            bytes_written_delta,
+            bytes_read_delta,
+            sample_count,
+            coverage,
+        ),
+    )
+    conn.commit()
+
+
+class TestDeriveDay:
+    """Spec §5.4: One row per UTC day, derived monotonically from hour rows."""
+
+    def test_single_hour_day(self, store_conn):
+        t_start = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        hour = "2026-09-01T12:00:00+00:00"
+        split = HourSplit(seconds_active=3600, seconds_idle=0, seconds_powered_off=0, seconds_unknown=0)
+        _insert_hour(store_conn, hour, split, bytes_written_delta=1024*1024*100)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.day == "2026-09-01"
+        assert day.seconds_active == 3600
+        assert day.seconds_idle == 0
+        assert day.seconds_powered_off == 0
+        assert day.bytes_written_delta == 1024*1024*100
+        assert day.sample_count == 1
+
+    def test_multiple_hours(self, store_conn):
+        t_start = datetime(2026, 9, 1, 10, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        hours = [
+            ("2026-09-01T10:00:00+00:00", HourSplit(1800, 1800, 0, 0), 50*1024*1024),
+            ("2026-09-01T11:00:00+00:00", HourSplit(3600, 0, 0, 0), 200*1024*1024),
+            ("2026-09-01T12:00:00+00:00", HourSplit(0, 0, 3600, 0), 0),
+        ]
+        for h, s, bw in hours:
+            _insert_hour(store_conn, h, s, bytes_written_delta=bw)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.seconds_active == 1800 + 3600 + 0
+        assert day.seconds_idle == 1800 + 0 + 0
+        assert day.seconds_powered_off == 0 + 0 + 3600
+        assert day.bytes_written_delta == 50*1024*1024 + 200*1024*1024 + 0
+        assert day.sample_count == 3
+
+    def test_no_hours_returns_none(self, store_conn):
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is None
+
+
+class TestUtcBounded:
+    """Spec §3.3, ST-4: Hours and days are UTC-bounded. No 23/25-hour days."""
+
+    def test_day_keys_are_utc_date_strings(self, store_conn):
+        t_start = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        hour = "2026-09-01T12:00:00+00:00"
+        split = HourSplit(0, 0, 3600, 0)
+        _insert_hour(store_conn, hour, split)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.day == "2026-09-01"
+
+    def test_cross_midnight_hours_produce_two_days(self, store_conn):
+        # Period spans both days
+        t_start = datetime(2026, 9, 1, 23, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 2, 1, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        hours = [
+            ("2026-09-01T23:00:00+00:00", HourSplit(3600, 0, 0, 0), 100),
+            ("2026-09-02T00:00:00+00:00", HourSplit(0, 0, 3600, 0), 0),
+        ]
+        for h, s, bw in hours:
+            _insert_hour(store_conn, h, s, bytes_written_delta=bw)
+
+        days = derive_all_days(store_conn)
+        day_keys = [d.day for d in days]
+        assert "2026-09-01" in day_keys
+        assert "2026-09-02" in day_keys
+        assert len(days) == 2
+
+
+class TestCoverage:
+    """Spec §5.2, §5.3, PR-5: Coverage is known share of wall-clock seconds
+    inside monitoring periods."""
+
+    def test_full_coverage_all_known(self, store_conn):
+        """All hours in period classified -> coverage = 1.0."""
+        t_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 3, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        hours = [
+            ("2026-09-01T00:00:00+00:00", HourSplit(3600, 0, 0, 0), 100),
+            ("2026-09-01T01:00:00+00:00", HourSplit(0, 3600, 0, 0), 50),
+            ("2026-09-01T02:00:00+00:00", HourSplit(0, 0, 3600, 0), 0),
+        ]
+        for h, s, bw in hours:
+            _insert_hour(store_conn, h, s, bytes_written_delta=bw)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.coverage == pytest.approx(1.0)
+
+    def test_gap_reduces_coverage(self, store_conn):
+        """Missing hour -> unknown seconds reduce coverage."""
+        t_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 3, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        _insert_hour(store_conn, "2026-09-01T00:00:00+00:00",
+                     HourSplit(3600, 0, 0, 0), bytes_written_delta=100)
+        _insert_hour(store_conn, "2026-09-01T02:00:00+00:00",
+                     HourSplit(0, 3600, 0, 0), bytes_written_delta=50)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.coverage == pytest.approx(7200 / 10800)
+        assert day.seconds_unknown == 3600
+
+    def test_outside_period_excluded(self, store_conn):
+        """Hours outside any monitoring period excluded from denominator."""
+        t_start = datetime(2026, 9, 1, 1, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 2, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        # Hour 01 is inside the period
+        _insert_hour(store_conn, "2026-09-01T01:00:00+00:00",
+                     HourSplit(3600, 0, 0, 0), bytes_written_delta=100)
+        # Hour 02 is outside (period ends at 02:00)
+        _insert_hour(store_conn, "2026-09-01T02:00:00+00:00",
+                     HourSplit(0, 3600, 0, 0), bytes_written_delta=50)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.coverage == pytest.approx(1.0)
+
+    def test_unknown_in_period_reduces_coverage(self, store_conn):
+        """Unknown seconds inside period count in denominator but not numerator."""
+        t_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 1, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        _insert_hour(store_conn, "2026-09-01T00:00:00+00:00",
+                     HourSplit(1800, 0, 0, 1800), bytes_written_delta=100)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.coverage == pytest.approx(0.5)
+        assert day.seconds_unknown == 1800
+
+
+class TestNoFabrication:
+    """Spec §5.3, FL-3: No absent hour is ever interpolated/estimated/fabricated."""
+
+    def test_missing_hours_stay_unknown(self, store_conn):
+        """Gap hours are never filled in — they remain as unknown seconds."""
+        t_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 3, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+
+        # Only hour 02 — hours 00 and 01 are gaps
+        _insert_hour(store_conn, "2026-09-01T02:00:00+00:00",
+                     HourSplit(3600, 0, 0, 0), bytes_written_delta=100)
+
+        day = derive_day(store_conn, "2026-09-01")
+        assert day is not None
+        assert day.seconds_unknown == 2 * 3600
+        assert day.seconds_active == 3600
+        assert day.sample_count == 1

tests/test_hour_classification.py

@@ -0,0 +1,218 @@
+"""Hour classification tests.
+
+From spec §5.1 — each UTC hour is classified by named constants:
+- Powered-off: power-on-hours delta < 90% of wall-clock span
+- Active: DUW delta >= 256 MiB
+- Idle: powered on + sampled + below active threshold
+- Unknown: everything else
+
+Four splits sum to exactly 3600s. Disabled time is never an hour state.
+"""
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from fenris.hour_classify import classify_hour, HourSplit, ACTIVE_THRESHOLD_BYTES
+
+HOUR_SECONDS = 3600
+
+
+class TestPoweredOff:
+    """Spec §5.1: Powered-off when power-on-hours delta < 90% of wall-clock span."""
+
+    def test_below_90_percent_poh_is_powered_off(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=0,
+            duw_delta=0,
+            dur_delta=0,
+        )
+        assert split.seconds_powered_off == HOUR_SECONDS
+        assert split.seconds_active == 0
+        assert split.seconds_idle == 0
+        assert split.seconds_unknown == 0
+
+    def test_exactly_90_percent_poh_is_not_powered_off(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=3240,
+            duw_delta=0,
+            dur_delta=0,
+        )
+        assert split.seconds_powered_off == 0
+        assert split.seconds_unknown == HOUR_SECONDS
+
+    def test_just_below_90_percent_poh_is_powered_off(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=3239,
+            duw_delta=0,
+            dur_delta=0,
+        )
+        assert split.seconds_powered_off == HOUR_SECONDS
+
+    def test_100_percent_poh_is_not_powered_off(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=HOUR_SECONDS,
+            duw_delta=0,
+            dur_delta=0,
+        )
+        assert split.seconds_powered_off == 0
+
+
+class TestActive:
+    """Spec §5.1: Active when DUW delta >= 256 MiB."""
+
+    def test_above_256_mib_is_active(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=HOUR_SECONDS,
+            duw_delta=ACTIVE_THRESHOLD_BYTES,
+            dur_delta=1000,
+            sampled_seconds=HOUR_SECONDS,
+        )
+        assert split.seconds_active == HOUR_SECONDS
+
+    def test_just_below_256_mib_is_idle(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=HOUR_SECONDS,
+            duw_delta=ACTIVE_THRESHOLD_BYTES - 1,
+            dur_delta=1000,
+            sampled_seconds=HOUR_SECONDS,
+        )
+        assert split.seconds_idle == HOUR_SECONDS
+        assert split.seconds_active == 0
+
+    def test_powered_off_takes_priority_over_active_writes(self):
+        """Spec §5.1 order: powered-off checked first."""
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=0,
+            duw_delta=ACTIVE_THRESHOLD_BYTES,
+            dur_delta=1000,
+        )
+        assert split.seconds_powered_off == HOUR_SECONDS
+        assert split.seconds_active == 0
+
+
+class TestIdle:
+    """Spec §5.1: Idle when powered on, sampled, below active threshold."""
+
+    def test_idle_with_writes_below_threshold(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=HOUR_SECONDS,
+            duw_delta=1024 * 1024,  # 1 MiB
+            dur_delta=500,
+            sampled_seconds=HOUR_SECONDS,
+        )
+        assert split.seconds_idle == HOUR_SECONDS
+        assert split.seconds_active == 0
+
+    def test_idle_no_writes(self):
+        """Powered on, sampled, zero writes -> idle."""
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=HOUR_SECONDS,
+            duw_delta=0,
+            dur_delta=0,
+            sampled_seconds=HOUR_SECONDS,
+        )
+        assert split.seconds_idle == HOUR_SECONDS
+
+
+class TestUnknown:
+    """Spec §5.1: Unknown when unsampled without POH evidence."""
+
+    def test_no_sample_no_writes_is_unknown(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=HOUR_SECONDS,
+            duw_delta=0,
+            dur_delta=0,
+        )
+        assert split.seconds_unknown == HOUR_SECONDS
+
+    def test_partial_unknown(self):
+        """Partial hour -> split unknown for unsampled portion."""
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=HOUR_SECONDS,
+            duw_delta=0,
+            dur_delta=0,
+            sampled_seconds=1800,
+        )
+        assert split.seconds_unknown == 1800
+        assert split.seconds_idle == 1800
+
+    def test_powered_off_with_no_sample_still_powered_off(self):
+        """POH < 90% is powered off regardless of sample status."""
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=0,
+            duw_delta=0,
+            dur_delta=0,
+            sampled_seconds=0,
+        )
+        assert split.seconds_powered_off == HOUR_SECONDS
+        assert split.seconds_unknown == 0
+
+
+class TestSumTo3600:
+    """Spec §5.1: Four splits sum to exactly wall_clock_seconds."""
+
+    def test_all_scenarios_sum_to_wall_clock(self):
+        scenarios = [
+            {"wall_clock_seconds": 3600, "poh_delta": 0, "duw_delta": 0, "dur_delta": 0},
+            {"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": 0, "dur_delta": 0},
+            {"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": ACTIVE_THRESHOLD_BYTES, "dur_delta": 1000, "sampled_seconds": 3600},
+            {"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": 1000, "dur_delta": 500, "sampled_seconds": 3600},
+            {"wall_clock_seconds": 3600, "poh_delta": 3239, "duw_delta": 0, "dur_delta": 0},
+            {"wall_clock_seconds": 1800, "poh_delta": 0, "duw_delta": 0, "dur_delta": 0},
+            {"wall_clock_seconds": 3600, "poh_delta": 3600, "duw_delta": 0, "dur_delta": 0, "sampled_seconds": 1800},
+        ]
+        for s in scenarios:
+            split = classify_hour(**s)
+            total = (
+                split.seconds_active
+                + split.seconds_idle
+                + split.seconds_powered_off
+                + split.seconds_unknown
+            )
+            assert total == s["wall_clock_seconds"], (
+                f"Scenario {s}: splits sum to {total}, expected {s['wall_clock_seconds']}"
+            )
+
+    def test_partial_wall_clock(self):
+        split = classify_hour(
+            wall_clock_seconds=1800,
+            poh_delta=0,
+            duw_delta=0,
+            dur_delta=0,
+        )
+        total = (
+            split.seconds_active
+            + split.seconds_idle
+            + split.seconds_powered_off
+            + split.seconds_unknown
+        )
+        assert total == 1800
+        assert split.seconds_powered_off == 1800
+
+
+class TestDisabledTimeNotAnHourState:
+    """Spec §5.2: Disabled time is not an hour state — excluded from numerator/denominator."""
+
+    def test_classify_hour_has_no_disabled_state(self):
+        split = classify_hour(
+            wall_clock_seconds=HOUR_SECONDS,
+            poh_delta=0,
+            duw_delta=0,
+            dur_delta=0,
+        )
+        assert not hasattr(split, "seconds_disabled")

tests/test_monitoring_periods.py

@@ -0,0 +1,147 @@
+"""Monitoring period tests.
+
+From spec §5.2, §8.6, §9.8:
+- Run finding no open period opens one at the run moment, never backdated
+- Wall-clock outside periods excluded from numerator/denominator
+- Powered-off time stays inside a period; disabled time does not
+- End causes: user_disabled, migrated, unknown_gap
+"""
+import sqlite3
+import sys
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from fenris.store import init_store
+from fenris.monitoring_periods import (
+    ensure_period_open,
+    close_period,
+    get_open_period,
+    is_inside_period,
+)
+
+
+@pytest.fixture
+def store_conn(tmp_path: Path):
+    """Initialize an observation store and return a connection."""
+    db_path = tmp_path / "test.db"
+    conn = init_store(db_path)
+    yield conn
+    conn.close()
+
+
+class TestEnsurePeriodOpen:
+    """Spec §9.8: Run finding no open period opens one at the run moment."""
+
+    def test_opens_period_when_none_exists(self, store_conn):
+        """First collection run opens a period at the run moment."""
+        run_time = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, run_time)
+
+        period = get_open_period(store_conn)
+        assert period is not None
+        assert period["started_at"] == "2026-09-01T12:00:00+00:00"
+        assert period["ended_at"] is None
+        assert period["end_cause"] is None
+
+    def test_no_opener_when_already_open(self, store_conn):
+        """If a period is already open, no new period is created."""
+        t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t2 = datetime(2026, 9, 1, 12, 5, 0, tzinfo=timezone.utc)
+
+        ensure_period_open(store_conn, t1)
+        ensure_period_open(store_conn, t2)
+
+        period = get_open_period(store_conn)
+        assert period is not None
+        assert period["started_at"] == "2026-09-01T12:00:00+00:00"
+
+    def test_never_backdated(self, store_conn):
+        """Period starts at the run moment, not the beginning of the hour."""
+        run_time = datetime(2026, 9, 1, 12, 3, 45, tzinfo=timezone.utc)
+        ensure_period_open(store_conn, run_time)
+
+        period = get_open_period(store_conn)
+        assert period["started_at"] == "2026-09-01T12:03:45+00:00"
+
+    def test_new_period_after_close(self, store_conn):
+        """After closing a period, next run opens a new one."""
+        t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t2 = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc)
+        t3 = datetime(2026, 9, 1, 14, 0, 0, tzinfo=timezone.utc)
+
+        ensure_period_open(store_conn, t1)
+        close_period(store_conn, t2, "user_disabled")
+        ensure_period_open(store_conn, t3)
+
+        period = get_open_period(store_conn)
+        assert period is not None
+        assert period["started_at"] == "2026-09-01T14:00:00+00:00"
+
+
+class TestClosePeriod:
+    """Spec §8.6: Pause closes with user_disabled."""
+
+    def test_close_with_user_disabled(self, store_conn):
+        t1 = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t2 = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc)
+
+        ensure_period_open(store_conn, t1)
+        close_period(store_conn, t2, "user_disabled")
+
+        period = get_open_period(store_conn)
+        assert period is None
+
+        cursor = store_conn.execute(
+            "SELECT ended_at, end_cause FROM monitoring_periods WHERE id = 1"
+        )
+        row = cursor.fetchone()
+        assert row[0] == "2026-09-01T13:00:00+00:00"
+        assert row[1] == "user_disabled"
+
+    def test_close_nonexistent_is_noop(self, store_conn):
+        """Closing when no period is open is a no-op (spec §8.6 pause otherwise)."""
+        t = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        close_period(store_conn, t, "user_disabled")
+
+        cursor = store_conn.execute("SELECT COUNT(*) FROM monitoring_periods")
+        assert cursor.fetchone()[0] == 0
+
+
+class TestIsInsidePeriod:
+    """Spec §5.2: Wall-clock outside periods excluded from numerator/denominator."""
+
+    def test_inside_open_period(self, store_conn):
+        t_start = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t_inside = datetime(2026, 9, 1, 12, 30, 0, tzinfo=timezone.utc)
+
+        ensure_period_open(store_conn, t_start)
+        assert is_inside_period(store_conn, t_inside) is True
+
+    def test_outside_closed_period(self, store_conn):
+        t_start = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t_end = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc)
+        t_outside = datetime(2026, 9, 1, 14, 0, 0, tzinfo=timezone.utc)
+
+        ensure_period_open(store_conn, t_start)
+        close_period(store_conn, t_end, "user_disabled")
+        assert is_inside_period(store_conn, t_outside) is False
+
+    def test_outside_no_periods(self, store_conn):
+        t = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        assert is_inside_period(store_conn, t) is False
+
+    def test_inside_second_period(self, store_conn):
+        """Two periods with a gap; time in second period is inside."""
+        t1_start = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
+        t1_end = datetime(2026, 9, 1, 13, 0, 0, tzinfo=timezone.utc)
+        t2_start = datetime(2026, 9, 1, 14, 0, 0, tzinfo=timezone.utc)
+        t_inside = datetime(2026, 9, 1, 14, 30, 0, tzinfo=timezone.utc)
+
+        ensure_period_open(store_conn, t1_start)
+        close_period(store_conn, t1_end, "user_disabled")
+        ensure_period_open(store_conn, t2_start)
+        assert is_inside_period(store_conn, t_inside) is True

tests/test_pruning.py

@@ -0,0 +1,104 @@
+"""Raw sample pruning tests.
+
+Spec §3.4, ST-5: Raw samples pruned to 14 days; hour observations and
+day aggregates retained indefinitely.
+"""
+import sqlite3
+import sys
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from fenris.store import init_store
+from fenris.pruning import prune_old_samples
+
+
+@pytest.fixture
+def store_conn(tmp_path: Path):
+    db_path = tmp_path / "test.db"
+    conn = init_store(db_path)
+    yield conn
+    conn.close()
+
+
+def _insert_sample(conn, ts_iso, device="/dev/nvme0"):
+    conn.execute(
+        "INSERT INTO samples (ts, device, data_units_written, data_units_read, "
+        " bytes_written, bytes_read, percentage_used) VALUES (?, ?, 0, 0, 0, 0, 0)",
+        (ts_iso, device),
+    )
+    conn.commit()
+
+
+class TestPruneOldSamples:
+    """Spec §3.4: Raw samples pruned to 14 days."""
+
+    def test_keeps_recent_samples(self, store_conn):
+        now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
+        # Insert a sample 1 day ago
+        ts = (now - timedelta(days=1)).isoformat()
+        _insert_sample(store_conn, ts)
+
+        pruned = prune_old_samples(store_conn, now, retention_days=14)
+        assert pruned == 0
+
+        cursor = store_conn.execute("SELECT COUNT(*) FROM samples")
+        assert cursor.fetchone()[0] == 1
+
+    def test_removes_old_samples(self, store_conn):
+        now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
+        # Insert samples at 10, 14, and 15 days ago
+        for days_ago in [10, 14, 15]:
+            ts = (now - timedelta(days=days_ago)).isoformat()
+            _insert_sample(store_conn, ts)
+
+        pruned = prune_old_samples(store_conn, now, retention_days=14)
+        assert pruned == 1  # Only the 15-day-old sample removed
+
+        cursor = store_conn.execute("SELECT COUNT(*) FROM samples")
+        assert cursor.fetchone()[0] == 2
+
+    def test_removes_many_old_samples(self, store_conn):
+        now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
+        for days_ago in range(1, 30):
+            ts = (now - timedelta(days=days_ago)).isoformat()
+            _insert_sample(store_conn, ts)
+
+        pruned = prune_old_samples(store_conn, now, retention_days=14)
+        assert pruned == 15  # Days 15-29 removed
+
+        cursor = store_conn.execute("SELECT COUNT(*) FROM samples")
+        assert cursor.fetchone()[0] == 14  # Days 1-14 kept
+
+    def test_empty_store_no_error(self, store_conn):
+        now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
+        pruned = prune_old_samples(store_conn, now, retention_days=14)
+        assert pruned == 0
+
+    def test_hour_observations_not_pruned(self, store_conn):
+        """Hour observations are retained indefinitely."""
+        now = datetime(2026, 9, 15, 12, 0, 0, tzinfo=timezone.utc)
+        # Insert an old sample and a recent sample
+        _insert_sample(store_conn, (now - timedelta(days=20)).isoformat())
+        _insert_sample(store_conn, (now - timedelta(days=1)).isoformat())
+
+        # Insert an old hour observation
+        store_conn.execute(
+            "INSERT INTO hour_observations (hour, active_seconds, sample_count) "
+            "VALUES (?, 3600, 1)",
+            ((now - timedelta(days=20)).replace(hour=0, minute=0, second=0).isoformat(),),
+        )
+        store_conn.commit()
+
+        prune_old_samples(store_conn, now, retention_days=14)
+
+        # Sample removed
+        cursor = store_conn.execute("SELECT COUNT(*) FROM samples")
+        assert cursor.fetchone()[0] == 1
+
+        # Hour observation retained
+        cursor = store_conn.execute("SELECT COUNT(*) FROM hour_observations")
+        assert cursor.fetchone()[0] == 1