From 7802a72606cc098e12b577ca93b7788663bd3840 Mon Sep 17 00:00:00 2001 From: xavierk Date: Tue, 1 Sep 2026 23:16:33 +0530 Subject: [PATCH] feat: implement projection core pure function (closes #25) --- src/fenris/projection.py | 566 +++++++++++++++++++++++++++++++++++++++ tests/test_projection.py | 340 +++++++++++++++++++++++ 2 files changed, 906 insertions(+) create mode 100644 src/fenris/projection.py create mode 100644 tests/test_projection.py diff --git a/src/fenris/projection.py b/src/fenris/projection.py new file mode 100644 index 0000000..a94f289 --- /dev/null +++ b/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, + ) diff --git a/tests/test_projection.py b/tests/test_projection.py new file mode 100644 index 0000000..6f94683 --- /dev/null +++ b/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