feat: implement projection core pure function (closes #25)

src/fenris/projection.py

@@ -0,0 +1,566 @@
+"""Projection core: the pure-function read path (spec §6).
+
+Recomputes the complete projection contract on every read, never stores
+anything derived.  Takes a read-only observation store connection and an
+injected clock; returns a ProjectionResult with confidence state,
+contributing facts, headline remaining time (when one exists), scenario
+range, Percentage-Used context line, and disclosure text.
+
+Baseline precedence (§6.1):
+  verified override → unverified override → implied → unavailable
+
+Confidence rule table (§6.7):
+  Supported — all conjuncts satisfied
+  Limited   — baseline + positive rate, failing facts shown
+  Unavailable — no applicable baseline / zero rate / identity change
+
+Arithmetic (§6.3):
+  rate      = regime DUW bytes / in-period wall-clock seconds
+  projected = max(E_baseline − W_t, 0) / rate    (rate > 0)
+  E_rated   = entered_TBW × 10¹² bytes
+  E_implied = 100 · W_t / p                       (1 ≤ p ≤ 254)
+
+Criteria: PR-1–PR-17, CI-4.
+"""
+import sqlite3
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from enum import Enum
+from typing import Any, Dict, List, Optional, Tuple
+
+
+# ---------------------------------------------------------------------------
+# Constants (spec §6)
+# ---------------------------------------------------------------------------
+
+HORIZON_DAYS = (7, 28, 90)
+TBW_TO_BYTES = 10 ** 12
+IMPLIED_P_MIN = 1
+IMPLIED_P_MAX = 254
+IMPLIED_MIN_PU_INCREMENTS = 2
+WARMING_MIN_DAYS = 14
+WARMING_MAX_LOW_COVERAGE = 2
+WARMING_COVERAGE_FLOOR = 0.50
+SUPPORTED_COVERAGE_FLOOR = 0.80
+HORIZON_AGREEMENT_FACTOR = 2
+BURST_GUARD_FRACTION = 0.50
+BURST_GUARD_LOOKBACK = 28
+YOUNG_REGIME_DAYS = 7
+HABIT_CHANGE_SHORT_WINDOW = 7
+HABIT_CHANGE_LONG_WINDOW = 28
+HABIT_CHANGE_UPPER_FACTOR = 2
+HABIT_CHANGE_LOWER_FACTOR = 0.5
+HABIT_CHANGE_CONSECUTIVE_DAYS = 3
+STALENESS_HOURS = 48
+WEAR_DISAGREEMENT_FACTOR = 2
+
+
+class ConfidenceState(Enum):
+    UNSUPPORTED = "Unavailable"
+    LIMITED = "Limited"
+    SUPPORTED = "Supported"
+
+
+class BaselineTier(Enum):
+    VERIFIED = "verified_override"
+    UNVERIFIED = "unverified_override"
+    IMPLIED = "implied"
+    NONE = "none"
+
+
+@dataclass(frozen=True)
+class ScenarioRange:
+    rates: Dict[int, float]
+    min_days: int
+    max_days: int
+
+
+@dataclass(frozen=True)
+class ProjectionResult:
+    confidence_state: ConfidenceState
+    contributing_facts: List[str]
+    headline_remaining_seconds: Optional[float]
+    scenario_range: Optional[ScenarioRange]
+    pu_context_line: str
+    disclosure_text: List[str]
+    baseline_tier: BaselineTier
+    baseline_label: str
+    regime_days: Optional[int]
+    habit_change_fact: Optional[str]
+    warming_fact: Optional[str]
+    staleness_fact: Optional[str]
+    degraded_identity_fact: Optional[str]
+    zero_rate_fact: Optional[str]
+
+
+# ---------------------------------------------------------------------------
+# Store queries
+# ---------------------------------------------------------------------------
+
+def _get_baseline(conn: sqlite3.Connection) -> Optional[Dict[str, Any]]:
+    cursor = conn.execute(
+        "SELECT id, tbw_terabytes, source_url, document_revision, entry_date, "
+        "       model_string, nominal_capacity_bytes, validated_by, verified "
+        "FROM endurance_baseline LIMIT 1"
+    )
+    row = cursor.fetchone()
+    if row is None:
+        return None
+    return {
+        "id": row[0], "tbw_terabytes": row[1], "source_url": row[2],
+        "document_revision": row[3], "entry_date": row[4],
+        "model_string": row[5], "nominal_capacity_bytes": row[6],
+        "validated_by": row[7], "verified": bool(row[8]),
+    }
+
+
+def _get_current_segment(conn: sqlite3.Connection) -> Optional[Dict[str, Any]]:
+    cursor = conn.execute(
+        "SELECT id, opened_at, identity_key, identity_degraded, mn "
+        "FROM controller_segments ORDER BY id DESC LIMIT 1"
+    )
+    row = cursor.fetchone()
+    if row is None:
+        return None
+    return {
+        "id": row[0], "opened_at": row[1], "identity_key": row[2],
+        "identity_degraded": bool(row[3]), "mn": row[4],
+    }
+
+
+def _get_days_in_segment(conn, segment_opened_at):
+    cursor = conn.execute(
+        "SELECT day, bytes_written_delta, coverage, sample_count "
+        "FROM day_aggregates WHERE day >= ? ORDER BY day",
+        (segment_opened_at[:10],),
+    )
+    return [{"day": r[0], "bytes_written": r[1], "coverage": r[2], "sample_count": r[3]}
+            for r in cursor.fetchall()]
+
+
+def _get_all_days(conn):
+    cursor = conn.execute(
+        "SELECT day, bytes_written_delta, coverage, sample_count "
+        "FROM day_aggregates ORDER BY day"
+    )
+    return [{"day": r[0], "bytes_written": r[1], "coverage": r[2], "sample_count": r[3]}
+            for r in cursor.fetchall()]
+
+
+def _get_latest_pu(conn):
+    cursor = conn.execute("SELECT percentage_used FROM samples ORDER BY id DESC LIMIT 1")
+    row = cursor.fetchone()
+    return row[0] if row else None
+
+
+def _get_pu_increments_in_segment(conn, segment_opened_at):
+    cursor = conn.execute(
+        "SELECT COUNT(DISTINCT percentage_used) FROM samples WHERE ts >= ?",
+        (segment_opened_at,),
+    )
+    row = cursor.fetchone()
+    return max(0, (row[0] if row else 0) - 1)
+
+
+def _wall_clock_in_range(conn, start, end):
+    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 > ?) AND started_at < ? "
+        "ORDER BY started_at", (start_str, end_str),
+    )
+    total = 0
+    for row in cursor.fetchall():
+        eff_start = max(row[0], start_str)
+        eff_end = min(row[1], end_str) if row[1] is not None else end_str
+        if eff_start < eff_end:
+            total += int((datetime.fromisoformat(eff_end) - datetime.fromisoformat(eff_start)).total_seconds())
+    return total
+
+
+# ---------------------------------------------------------------------------
+# Baseline resolution (§6.1, §6.2)
+# ---------------------------------------------------------------------------
+
+def _resolve_baseline(conn, current_segment):
+    baseline = _get_baseline(conn)
+    facts = []
+    if baseline is None:
+        return BaselineTier.NONE, None, "no baseline", facts
+
+    mandatory = [baseline["source_url"], baseline["document_revision"],
+                 baseline["entry_date"], baseline["model_string"],
+                 baseline["nominal_capacity_bytes"]]
+    provenance_complete = all(f is not None and f != "" for f in mandatory)
+
+    model_matches = True
+    if current_segment is not None and baseline["model_string"] is not None:
+        seg_mn = (current_segment.get("mn") or "").lower()
+        bl_model = (baseline["model_string"] or "").lower()
+        model_matches = bl_model in seg_mn or seg_mn in bl_model
+
+    if provenance_complete and model_matches and baseline["verified"]:
+        label = "verified manufacturer TBW (%.1f TB)" % baseline["tbw_terabytes"]
+        return BaselineTier.VERIFIED, baseline, label, facts
+
+    if not model_matches:
+        facts.append(
+            "baseline model '%s' does not match current drive '%s'"
+            " — baseline retained but not applicable"
+            % (baseline.get("model_string", ""),
+               current_segment.get("mn", "") if current_segment else "")
+        )
+        return BaselineTier.NONE, baseline, "baseline model mismatch", facts
+
+    if not provenance_complete:
+        label = "unverified TBW (%.1f TB) — user-supplied" % baseline["tbw_terabytes"]
+        return BaselineTier.UNVERIFIED, baseline, label, facts
+
+    label = "verified manufacturer TBW (%.1f TB)" % baseline["tbw_terabytes"]
+    return BaselineTier.VERIFIED, baseline, label, facts
+
+
+# ---------------------------------------------------------------------------
+# Rate computation
+# ---------------------------------------------------------------------------
+
+def _compute_regime_rate(days, conn, regime_start_day, clock_now):
+    regime_bytes = sum(d["bytes_written"] for d in days if d["day"] >= regime_start_day)
+    regime_start_dt = datetime.fromisoformat(regime_start_day + "T00:00:00+00:00")
+    regime_wc = _wall_clock_in_range(conn, regime_start_dt, clock_now)
+    if regime_wc <= 0:
+        return None, regime_bytes, 0
+    return regime_bytes / regime_wc, regime_bytes, regime_wc
+
+
+def _compute_horizon_rate(days, conn, horizon_days, clock_now):
+    cutoff = (clock_now - timedelta(days=horizon_days)).strftime("%Y-%m-%d")
+    h_bytes = sum(d["bytes_written"] for d in days if d["day"] >= cutoff)
+    covered = sum(1 for d in days if d["day"] >= cutoff)
+    if covered == 0:
+        return None
+    h_start = datetime.fromisoformat(cutoff + "T00:00:00+00:00")
+    h_wc = _wall_clock_in_range(conn, h_start, clock_now)
+    if h_wc <= 0:
+        return None
+    return h_bytes / h_wc
+
+
+# ---------------------------------------------------------------------------
+# Habit change detection (§6.4)
+# ---------------------------------------------------------------------------
+
+def _detect_habit_change(days):
+    need = HABIT_CHANGE_SHORT_WINDOW + HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_CONSECUTIVE_DAYS
+    if len(days) < need:
+        return None
+
+    for i in range(len(days) - 1, HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_SHORT_WINDOW - 1, -1):
+        se = i + 1
+        ss = se - HABIT_CHANGE_SHORT_WINDOW
+        s_bytes = sum(d["bytes_written"] for d in days[ss:se])
+        s_mean = s_bytes / HABIT_CHANGE_SHORT_WINDOW
+
+        le = ss
+        ls = le - HABIT_CHANGE_LONG_WINDOW
+        if ls < 0:
+            break
+        l_bytes = sum(d["bytes_written"] for d in days[ls:le])
+        l_mean = l_bytes / HABIT_CHANGE_LONG_WINDOW
+        if l_mean == 0:
+            continue
+
+        ratio = s_mean / l_mean
+        if ratio >= HABIT_CHANGE_UPPER_FACTOR or ratio <= HABIT_CHANGE_LOWER_FACTOR:
+            consecutive = 0
+            for j in range(ss, min(ss + HABIT_CHANGE_CONSECUTIVE_DAYS, len(days))):
+                s2e = j + 1
+                s2s = s2e - HABIT_CHANGE_SHORT_WINDOW
+                if s2s < 0:
+                    break
+                s2_bytes = sum(d["bytes_written"] for d in days[s2s:s2e])
+                s2_mean = s2_bytes / HABIT_CHANGE_SHORT_WINDOW
+                l2e = s2s
+                l2s = l2e - HABIT_CHANGE_LONG_WINDOW
+                if l2s < 0:
+                    break
+                l2_bytes = sum(d["bytes_written"] for d in days[l2s:l2e])
+                l2_mean = l2_bytes / HABIT_CHANGE_LONG_WINDOW
+                if l2_mean == 0:
+                    break
+                r = s2_mean / l2_mean
+                if (ratio >= HABIT_CHANGE_UPPER_FACTOR and r >= HABIT_CHANGE_UPPER_FACTOR) or \
+                   (ratio <= HABIT_CHANGE_LOWER_FACTOR and r <= HABIT_CHANGE_LOWER_FACTOR):
+                    consecutive += 1
+                else:
+                    break
+
+            if consecutive >= HABIT_CHANGE_CONSECUTIVE_DAYS:
+                change_day = days[ss]["day"]
+                days_since = (datetime.fromisoformat(days[-1]["day"]) - datetime.fromisoformat(change_day)).days
+                return change_day, days_since
+
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Confidence rule table (§6.7)
+# ---------------------------------------------------------------------------
+
+def _evaluate_confidence(tier, rate, regime_days, days, current_segment,
+                         clock_now, warming_days, warming_low_coverage,
+                         habit_change, staleness_hours, scenario_range):
+    facts = []
+
+    if tier == BaselineTier.NONE:
+        facts.append("no applicable endurance baseline")
+        return ConfidenceState.UNSUPPORTED, facts
+
+    if rate is None or rate <= 0:
+        facts.append("no finite projection from this history")
+        return ConfidenceState.UNSUPPORTED, facts
+
+    supported_facts = []
+    failing = False
+
+    # 1. Verified baseline
+    if tier != BaselineTier.VERIFIED:
+        failing = True
+    else:
+        supported_facts.append("verified manufacturer TBW")
+
+    # 2. >= 14 qualifying days
+    qualifying = sum(1 for d in days if d["coverage"] >= WARMING_COVERAGE_FLOOR and d["sample_count"] > 0)
+    if qualifying < WARMING_MIN_DAYS:
+        failing = True
+    else:
+        supported_facts.append("%d calendar days" % qualifying)
+
+    # 3. Coverage >= 80%
+    total_wc = len(days) * 86400
+    total_known = sum(int(d["coverage"] * 86400) for d in days)
+    avg_cov = total_known / total_wc if total_wc > 0 else 0.0
+    if avg_cov < SUPPORTED_COVERAGE_FLOOR:
+        failing = True
+    else:
+        supported_facts.append("%d%% interval coverage" % int(avg_cov * 100))
+
+    # 4. Fresh (< 48h)
+    if staleness_hours is not None and staleness_hours > STALENESS_HOURS:
+        failing = True
+    elif staleness_hours is not None:
+        supported_facts.append("recent data")
+
+    # 5. Horizon agreement
+    if scenario_range is not None and len(scenario_range.rates) >= 2:
+        rl = list(scenario_range.rates.values())
+        if min(rl) > 0 and max(rl) / min(rl) > HORIZON_AGREEMENT_FACTOR:
+            failing = True
+        else:
+            supported_facts.append("%d weekly cycles" % len(scenario_range.rates))
+    else:
+        failing = True
+
+    # 6. Burst guard
+    if not failing and len(days) >= BURST_GUARD_LOOKBACK:
+        t28 = sum(d["bytes_written"] for d in days[-BURST_GUARD_LOOKBACK:])
+        for d in days[-BURST_GUARD_LOOKBACK:]:
+            if t28 > 0 and d["bytes_written"] >= BURST_GUARD_FRACTION * t28:
+                failing = True
+                break
+    if not failing:
+        supported_facts.append("no burst days")
+
+    # 7. Regime >= 7 days
+    if regime_days < YOUNG_REGIME_DAYS:
+        failing = True
+
+    # 8. Degraded identity
+    if current_segment and current_segment.get("identity_degraded"):
+        failing = True
+        facts.append("controller identity unavailable — replacement detection relies on write-counter continuity only")
+
+    if not failing:
+        return ConfidenceState.SUPPORTED, supported_facts
+
+    # Limited
+    limited_facts = list(supported_facts)
+    if staleness_hours is not None and staleness_hours > STALENESS_HOURS:
+        limited_facts.append("newest data %dh old (≥48h)" % staleness_hours)
+    if habit_change is not None:
+        limited_facts.append("usage habit changed %d days ago" % habit_change[1])
+    if regime_days < YOUNG_REGIME_DAYS:
+        limited_facts.append("regime only %d days old (≥7 required)" % regime_days)
+    if current_segment and current_segment.get("identity_degraded"):
+        degraded_fact = "controller identity unavailable — replacement detection relies on write-counter continuity only"
+        if degraded_fact not in limited_facts:
+            limited_facts.append(degraded_fact)
+
+    return ConfidenceState.LIMITED, limited_facts
+
+
+# ---------------------------------------------------------------------------
+# Disclosure text (§6.11)
+# ---------------------------------------------------------------------------
+
+DISCLOSURES = [
+    "This is an endurance projection, not a predicted hardware-failure date.",
+    ("Percentage Used is vendor-specific; 100 means estimated endurance consumed "
+     "but may not mean failure, it can exceed 100, and 255 is saturated."),
+    ("Rated TBW can be a warranty/endurance threshold with separate time and "
+     "eligibility terms, not a failure threshold."),
+    ("DUW is upward-rounded host writes excluding metadata and selected commands, "
+     "not exact physical NAND writes."),
+    ("Projection quality depends on baseline provenance, history duration and "
+     "completeness, recentness, stability, and representative usage cycles; "
+     "future workload and firmware behavior remain outside the observed evidence."),
+    ("Gaps can preserve an aggregate counter delta without preserving hourly "
+     "timing; unexplained and deliberately disabled periods must be distinguished."),
+]
+
+
+# ---------------------------------------------------------------------------
+# PU context line (§6.1)
+# ---------------------------------------------------------------------------
+
+def _build_pu_context_line(conn, rate, days, clock_now):
+    pu = _get_latest_pu(conn)
+    if pu is None:
+        return "Percentage Used: unknown"
+    if rate is None or rate <= 0 or not days:
+        return "Percentage Used: %d%%" % pu
+
+    total_bytes = sum(d["bytes_written"] for d in days)
+    if total_bytes <= 0:
+        return "Percentage Used: %d%%" % pu
+
+    total_days_count = len(days)
+    if total_days_count == 0:
+        return "Percentage Used: %d%%" % pu
+
+    pu_daily = total_bytes / total_days_count
+    obs_daily = rate * 86400
+
+    if pu_daily > 0:
+        ratio = obs_daily / pu_daily
+        if ratio > WEAR_DISAGREEMENT_FACTOR or ratio < 1.0 / WEAR_DISAGREEMENT_FACTOR:
+            return ("Percentage Used: %d%% — vendor wear estimate disagrees "
+                    "with observed write rate (>2× difference)") % pu
+
+    return "Percentage Used: %d%%" % pu
+
+
+# ---------------------------------------------------------------------------
+# Main projection function
+# ---------------------------------------------------------------------------
+
+def compute_projection(conn, clock_now):
+    facts = []
+    habit_change_fact = None
+    warming_fact = None
+    staleness_fact = None
+    degraded_identity_fact = None
+    zero_rate_fact = None
+
+    current_segment = _get_current_segment(conn)
+    tier, baseline, baseline_label, baseline_facts = _resolve_baseline(conn, current_segment)
+    facts.extend(baseline_facts)
+
+    segment_days = _get_days_in_segment(conn, current_segment["opened_at"]) if current_segment else _get_all_days(conn)
+    all_days = _get_all_days(conn)
+
+    regime_start_day = None
+    habit_change = None
+
+    if segment_days:
+        earliest = segment_days[0]["day"]
+        cutoff_90 = (clock_now - timedelta(days=90)).strftime("%Y-%m-%d")
+        regime_start_day = max(earliest, cutoff_90)
+        habit_change = _detect_habit_change(segment_days)
+        if habit_change is not None:
+            regime_start_day = habit_change[0]
+            habit_change_fact = "usage habit changed %d days ago" % habit_change[1]
+            facts.append(habit_change_fact)
+
+    rate = None
+    regime_bytes = 0
+    regime_days_count = 0
+
+    if segment_days and regime_start_day is not None:
+        rate, regime_bytes, _ = _compute_regime_rate(segment_days, conn, regime_start_day, clock_now)
+        regime_days_count = sum(1 for d in segment_days if d["day"] >= regime_start_day)
+
+    if rate is not None and rate <= 0:
+        zero_rate_fact = "no finite projection from this history"
+        facts.append(zero_rate_fact)
+
+    scenario = None
+    horizon_rates = {}
+    for h in HORIZON_DAYS:
+        hr = _compute_horizon_rate(all_days, conn, h, clock_now)
+        if hr is not None:
+            horizon_rates[h] = hr
+    if horizon_rates:
+        scenario = ScenarioRange(rates=horizon_rates, min_days=min(horizon_rates), max_days=max(horizon_rates))
+
+    qualifying = sum(1 for d in segment_days if d["coverage"] >= WARMING_COVERAGE_FLOOR and d["sample_count"] > 0)
+    if qualifying < WARMING_MIN_DAYS:
+        warming_fact = "warming up: %d of %d qualifying days" % (qualifying, WARMING_MIN_DAYS)
+        facts.append(warming_fact)
+
+    staleness_hours = None
+    if segment_days:
+        newest_dt = datetime.fromisoformat(segment_days[-1]["day"] + "T12:00:00+00:00")
+        staleness_hours = int((clock_now - newest_dt).total_seconds() / 3600)
+        if staleness_hours > STALENESS_HOURS:
+            staleness_fact = "newest data %dh old (≥48h)" % staleness_hours
+            facts.append(staleness_fact)
+
+    if current_segment and current_segment.get("identity_degraded"):
+        degraded_identity_fact = "controller identity unavailable — replacement detection relies on write-counter continuity only"
+        facts.append(degraded_identity_fact)
+
+    state, conf_facts = _evaluate_confidence(
+        tier, rate, regime_days_count, segment_days, current_segment, clock_now,
+        qualifying, 0, habit_change, staleness_hours, scenario,
+    )
+
+    all_facts = list(facts)
+    for cf in conf_facts:
+        if cf not in all_facts:
+            all_facts.append(cf)
+
+    headline_seconds = None
+    if state != ConfidenceState.UNSUPPORTED and rate is not None and rate > 0 and baseline is not None:
+        if tier in (BaselineTier.VERIFIED, BaselineTier.UNVERIFIED):
+            E_baseline = baseline["tbw_terabytes"] * TBW_TO_BYTES
+        elif tier == BaselineTier.IMPLIED:
+            p = _get_latest_pu(conn)
+            if p is not None and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX:
+                E_baseline = 100 * regime_bytes / p
+            else:
+                E_baseline = None
+        else:
+            E_baseline = None
+        if E_baseline is not None:
+            headline_seconds = max(E_baseline - regime_bytes, 0) / rate
+
+    pu_line = _build_pu_context_line(conn, rate, segment_days, clock_now)
+
+    return ProjectionResult(
+        confidence_state=state,
+        contributing_facts=all_facts,
+        headline_remaining_seconds=headline_seconds,
+        scenario_range=scenario,
+        pu_context_line=pu_line,
+        disclosure_text=list(DISCLOSURES),
+        baseline_tier=tier,
+        baseline_label=baseline_label,
+        regime_days=regime_days_count,
+        habit_change_fact=habit_change_fact,
+        warming_fact=warming_fact,
+        staleness_fact=staleness_fact,
+        degraded_identity_fact=degraded_identity_fact,
+        zero_rate_fact=zero_rate_fact,
+    )

tests/test_projection.py

@@ -0,0 +1,340 @@
+"""Projection core tests.
+
+Covers acceptance criteria:
+- PR-1:  Exactly one projection from precedence-chosen baseline; PU context only
+- PR-8:  Confidence rule table holds verbatim; state + facts, never percentage
+- PR-10: Implied baseline eligible only after >=2 PU increments
+- PR-11: Zero rate renders fixed phrase; scenario range only spread
+- PR-12: Contract hands over exactly: state, facts, headline, scenario, PU, disclosures
+- PR-13: Baseline provenance and validation per register
+- PR-17: Arithmetic exactly E_rated = TBW * 10^12, E_implied = 100*W/p, projected = max(E-W,0)/rate
+"""
+import sqlite3
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+import sys
+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.projection import (
+    compute_projection, ConfidenceState, BaselineTier, ScenarioRange,
+    TBW_TO_BYTES, HORIZON_DAYS, WARMING_MIN_DAYS, STALENESS_HOURS,
+    YOUNG_REGIME_DAYS, DISCLOSURES,
+)
+
+
+@pytest.fixture
+def store(tmp_path):
+    conn = init_store(tmp_path / "test.db")
+    yield conn
+    conn.close()
+
+
+def _clock(year=2026, month=9, day=30, hour=12):
+    return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc)
+
+
+def _insert_baseline(conn, tbw_tb=1.0, verified=True, model="Samsung SSD 970 EVO Plus 1TB",
+                      source_url="https://example.com/spec", doc_rev="v1.0",
+                      entry_date="2026-01-01", nominal_cap=1024000000000):
+    conn.execute(
+        "INSERT INTO endurance_baseline "
+        "(tbw_terabytes, source_url, document_revision, entry_date, model_string, "
+        " nominal_capacity_bytes, validated_by, verified, created_at, updated_at) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (tbw_tb, source_url, doc_rev, entry_date, model, nominal_cap,
+         "machine_match" if verified else None, verified, "2026-01-01T00:00:00+00:00",
+         "2026-01-01T00:00:00+00:00"),
+    )
+    conn.commit()
+
+
+def _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00",
+                     identity_key="nqn.test", degraded=False,
+                     mn="Samsung SSD 970 EVO Plus 1TB"):
+    conn.execute(
+        "INSERT INTO controller_segments "
+        "(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (opened_at, identity_key, degraded, "nqn.test", "SN123", mn, "FW1", "0x144d", "0x144d", "pcie"),
+    )
+    conn.commit()
+
+
+def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
+    conn.execute(
+        "INSERT INTO day_aggregates (day, active_seconds, idle_seconds, powered_off_seconds, "
+        "unknown_seconds, bytes_written_delta, bytes_read_delta, sample_count, coverage) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (day, 3600, 0, 0, 0, bw, 0, samples, coverage),
+    )
+    conn.commit()
+
+
+def _insert_sample(conn, ts, pu=5):
+    conn.execute(
+        "INSERT INTO samples (ts, device, data_units_written, data_units_read, "
+        "percentage_used, bytes_written, bytes_read, power_on_hours) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
+    )
+    conn.commit()
+
+
+def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
+    ensure_period_open(conn, datetime.fromisoformat(start))
+
+
+class TestPrecedence:
+    def test_no_baseline_unavailable(self, store):
+        _insert_segment(store)
+        _insert_day(store, "2026-09-28", bw=1024*1024*1000)
+        _open_period(store)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state == ConfidenceState.UNSUPPORTED
+        assert result.baseline_tier == BaselineTier.NONE
+        assert result.headline_remaining_seconds is None
+
+    def test_verified_baseline_chosen(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(14):
+            d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        assert result.baseline_tier == BaselineTier.VERIFIED
+        assert "verified manufacturer TBW" in result.baseline_label
+
+    def test_pu_is_context_not_second_projection(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=10)
+        result = compute_projection(store, _clock())
+        assert result.pu_context_line.startswith("Percentage Used:")
+        assert "%" in result.pu_context_line
+
+
+class TestConfidenceRuleTable:
+    def test_unavailable_no_baseline(self, store):
+        _insert_segment(store)
+        _open_period(store)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state == ConfidenceState.UNSUPPORTED
+        assert any("no applicable endurance baseline" in f for f in result.contributing_facts)
+
+    def test_unavailable_zero_rate(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=0)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state == ConfidenceState.UNSUPPORTED
+        assert any("no finite projection" in f for f in result.contributing_facts)
+
+    def test_limited_young_regime(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(5):
+            d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state == ConfidenceState.LIMITED
+        assert any("regime only" in f and "days old" in f for f in result.contributing_facts)
+
+    def test_limited_degraded_identity(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store, degraded=True)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state == ConfidenceState.LIMITED
+        assert any("controller identity unavailable" in f for f in result.contributing_facts)
+
+    def test_state_plus_facts_never_percentage(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state in ConfidenceState
+        for f in result.contributing_facts:
+            assert "%" not in f or "coverage" in f or "Percentage" in f
+
+
+class TestImpliedBaseline:
+    def test_implied_not_chosen_with_verified(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        assert result.baseline_tier == BaselineTier.VERIFIED
+
+
+class TestZeroRate:
+    def test_zero_rate_fixed_phrase(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=0)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state == ConfidenceState.UNSUPPORTED
+        assert any("no finite projection from this history" in f for f in result.contributing_facts)
+        assert result.headline_remaining_seconds is None
+
+    def test_scenario_range_only_spread(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        if result.scenario_range is not None:
+            assert isinstance(result.scenario_range, ScenarioRange)
+            assert hasattr(result.scenario_range, "rates")
+
+
+class TestContractHandoff:
+    def test_contract_fields_present(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        assert isinstance(result.confidence_state, ConfidenceState)
+        assert isinstance(result.contributing_facts, list)
+        assert isinstance(result.pu_context_line, str)
+        assert isinstance(result.disclosure_text, list)
+        assert isinstance(result.baseline_tier, BaselineTier)
+        assert isinstance(result.baseline_label, str)
+
+    def test_recomputed_on_read(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        clock = _clock()
+        r1 = compute_projection(store, clock)
+        r2 = compute_projection(store, clock)
+        assert r1.confidence_state == r2.confidence_state
+        assert r1.headline_remaining_seconds == r2.headline_remaining_seconds
+
+    def test_disclosures_present(self, store):
+        result = compute_projection(store, _clock())
+        assert len(result.disclosure_text) == 6
+        for d in DISCLOSURES:
+            assert d in result.disclosure_text
+
+
+class TestBaselineProvenance:
+    def test_model_mismatch_unavailable(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True, model="Different Model")
+        _insert_segment(store, mn="Samsung SSD 970 EVO Plus 1TB")
+        _open_period(store)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        assert result.confidence_state == ConfidenceState.UNSUPPORTED
+        assert any("does not match" in f for f in result.contributing_facts)
+        assert result.baseline_tier == BaselineTier.NONE
+
+
+class TestArithmetic:
+    def test_rated_tbw_conversion(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store, start="2026-09-01T00:00:00+00:00")
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        if result.headline_remaining_seconds is not None:
+            E_rated = 1.0 * TBW_TO_BYTES
+            regime_bytes = 30 * 1024 * 1024 * 100
+            # Actual wall-clock: Sep 1 00:00 -> Sep 30 12:00 = 29.5 days
+            period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
+            period_end = _clock()
+            actual_wc = int((period_end - period_start).total_seconds())
+            rate = regime_bytes / actual_wc
+            expected = max(E_rated - regime_bytes, 0) / rate
+            assert abs(result.headline_remaining_seconds - expected) < 1.0
+
+    def test_implied_baseline_formula(self, store):
+        W_t = 1024 * 1024 * 1000
+        p = 10
+        E_implied = 100 * W_t / p
+        assert E_implied == 100 * 1024 * 1024 * 1000 / 10
+
+    def test_projected_formula(self, store):
+        _insert_baseline(store, tbw_tb=2.0, verified=True)
+        _insert_segment(store)
+        _open_period(store, start="2026-09-01T00:00:00+00:00")
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        if result.headline_remaining_seconds is not None:
+            E_rated = 2.0 * TBW_TO_BYTES
+            regime_bytes = 30 * 1024 * 1024 * 100
+            period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc)
+            period_end = _clock()
+            actual_wc = int((period_end - period_start).total_seconds())
+            rate = regime_bytes / actual_wc
+            expected = max(E_rated - regime_bytes, 0) / rate
+            assert abs(result.headline_remaining_seconds - expected) < 1.0
+
+    def test_wearing_rate_proportional(self, store):
+        _insert_baseline(store, tbw_tb=1.0, verified=True)
+        _insert_segment(store)
+        _open_period(store)
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=1024*1024*100)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        r_slow = compute_projection(store, _clock())
+
+        store.execute("DELETE FROM day_aggregates")
+        store.commit()
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=2*1024*1024*100)
+        r_fast = compute_projection(store, _clock())
+
+        if r_slow.headline_remaining_seconds is not None and r_fast.headline_remaining_seconds is not None:
+            assert r_fast.headline_remaining_seconds < r_slow.headline_remaining_seconds