diff --git a/src/fenris/projection.py b/src/fenris/projection.py index a94f289..59d84c2 100644 --- a/src/fenris/projection.py +++ b/src/fenris/projection.py @@ -236,6 +236,9 @@ def _compute_regime_rate(days, conn, regime_start_day, clock_now): def _compute_horizon_rate(days, conn, horizon_days, clock_now): cutoff = (clock_now - timedelta(days=horizon_days)).strftime("%Y-%m-%d") + # History must span the full horizon — no placeholders + if not days or days[0]["day"] > cutoff: + return None 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: @@ -252,54 +255,62 @@ def _compute_horizon_rate(days, conn, horizon_days, clock_now): # --------------------------------------------------------------------------- def _detect_habit_change(days): - need = HABIT_CHANGE_SHORT_WINDOW + HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_CONSECUTIVE_DAYS + """Detect habit change per spec §6.4. + + Trailing 7-day mean >= 2x (or <= 0.5x) the preceding 28-day mean + for 3 consecutive days. Returns (change_day, days_since) or None. + The first divergence day is the earliest day in the consecutive run. + """ + need = HABIT_CHANGE_SHORT_WINDOW + HABIT_CHANGE_LONG_WINDOW 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 + def _ratio_at(end_idx): + """Compute 7-day / preceding-28-day mean ratio ending at end_idx.""" + if end_idx < HABIT_CHANGE_SHORT_WINDOW - 1: + return None + se = end_idx + 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 + return None l_bytes = sum(d["bytes_written"] for d in days[ls:le]) l_mean = l_bytes / HABIT_CHANGE_LONG_WINDOW if l_mean == 0: + return None + return s_mean / l_mean + + # Scan backwards from the most recent day + for i in range(len(days) - 1, HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_SHORT_WINDOW - 2, -1): + ratio = _ratio_at(i) + if ratio is None: 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 + is_upper = ratio >= HABIT_CHANGE_UPPER_FACTOR + is_lower = ratio <= HABIT_CHANGE_LOWER_FACTOR + if not (is_upper or is_lower): + continue - 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 + # Count consecutive days going backwards from i + consecutive = 1 + for j in range(i - 1, HABIT_CHANGE_LONG_WINDOW + HABIT_CHANGE_SHORT_WINDOW - 3, -1): + r = _ratio_at(j) + if r is None: + break + if (is_upper and r >= HABIT_CHANGE_UPPER_FACTOR) or \ + (is_lower and r <= HABIT_CHANGE_LOWER_FACTOR): + consecutive += 1 + else: + break + + if consecutive >= HABIT_CHANGE_CONSECUTIVE_DAYS: + change_idx = i - consecutive + 1 + change_day = days[change_idx]["day"] + days_since = (datetime.fromisoformat(days[-1]["day"]) - datetime.fromisoformat(change_day)).days + return change_day, days_since return None @@ -504,8 +515,11 @@ def compute_projection(conn, clock_now): 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: + total_days_count = len(segment_days) + days_below_coverage = sum(1 for d in segment_days + if d["coverage"] < WARMING_COVERAGE_FLOOR or d["sample_count"] == 0) + qualifying = total_days_count - days_below_coverage + if total_days_count < WARMING_MIN_DAYS or days_below_coverage > WARMING_MAX_LOW_COVERAGE: warming_fact = "warming up: %d of %d qualifying days" % (qualifying, WARMING_MIN_DAYS) facts.append(warming_fact) diff --git a/tests/test_projection.py b/tests/test_projection.py index 6f94683..2c0ee78 100644 --- a/tests/test_projection.py +++ b/tests/test_projection.py @@ -338,3 +338,564 @@ class TestArithmetic: 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 + + + +# =========================================================================== +# Issue #26: Project from the sustained regime +# Habit change, scenario range, and evidence gates +# =========================================================================== + + +class TestSustainedRegimeRate: + """PR-2: Headline rate is sustained-regime rate; default regime = full + history capped at 90 days; scenario range computed independently, + covered horizons only, no placeholders.""" + + def test_headline_rate_from_regime(self, store): + """Rate is regime DUW / wall-clock, not trailing-24h or all-history.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + # 30 days of 100 MiB/day + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + # Regime = full 30 days; rate = 30*bw / wall-clock + regime_bytes = 30 * bw + period_start = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc) + wc = int((_clock() - period_start).total_seconds()) + expected_rate = regime_bytes / wc + if result.headline_remaining_seconds is not None: + E = 10.0 * TBW_TO_BYTES + expected_seconds = max(E - regime_bytes, 0) / expected_rate + assert abs(result.headline_remaining_seconds - expected_seconds) < 1.0 + + def test_regime_capped_at_90_days(self, store): + """Default regime is full history capped at 90 days.""" + _insert_baseline(store, tbw_tb=100.0, verified=True) + _insert_segment(store, opened_at="2026-06-01T00:00:00+00:00") + _open_period(store, start="2026-06-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + # 120 days of data (Jun 1 - Sep 28) + for i in range(120): + d = (datetime(2026, 6, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-28T12:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=30, hour=12)) + # Regime should be capped at 90 days (from Jun 1 to Sep 30 = 90 days at cutoff) + # The 90-day cutoff is Sep 30 - 90 = Jul 1, so regime starts Jul 1 + assert result.regime_days is not None + assert result.regime_days <= 90 + + def test_scenario_range_independent_of_regime(self, store): + """Scenario range is computed independently from the regime.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + if result.scenario_range is not None: + # Should have 7-day and 28-day horizons (90-day not fully covered) + assert 7 in result.scenario_range.rates + assert 28 in result.scenario_range.rates + + def test_only_covered_horizons_shown(self, store): + """No placeholder horizons — only horizons the history covers.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-20T00:00:00+00:00") + _open_period(store, start="2026-09-20T00:00:00+00:00") + bw = 100 * 1024 * 1024 + # Only 10 days of data + for i in range(10): + d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + if result.scenario_range is not None: + # 7-day is covered, 28-day and 90-day are not + assert 7 in result.scenario_range.rates + assert 28 not in result.scenario_range.rates + assert 90 not in result.scenario_range.rates + + +class TestHabitChange: + """PR-3: Habit change triggers at 2x/0.5x sustained 3 consecutive days, + regime starts at first divergence day, auto-adopted and labeled; + young regime caps at Limited.""" + + def test_habit_change_2x_detected(self, store): + """2x increase for 3+ consecutive days triggers habit change.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-08-01T00:00:00+00:00") + _open_period(store, start="2026-08-01T00:00:00+00:00") + bw_normal = 100 * 1024 * 1024 + bw_high = 300 * 1024 * 1024 # 3x the normal rate + # 28 days of normal usage + for i in range(28): + d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_normal) + # 10 days of high usage (3x > 2x threshold) + for i in range(10): + d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_high) + _insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12)) + assert result.habit_change_fact is not None + assert "usage habit changed" in result.habit_change_fact + assert "days ago" in result.habit_change_fact + + def test_habit_change_05x_detected(self, store): + """0.5x decrease for 3+ consecutive days triggers habit change.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-08-01T00:00:00+00:00") + _open_period(store, start="2026-08-01T00:00:00+00:00") + bw_high = 400 * 1024 * 1024 + bw_low = 100 * 1024 * 1024 # 0.25x < 0.5x threshold + # 28 days of high usage + for i in range(28): + d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_high) + # 10 days of low usage + for i in range(10): + d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_low) + _insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12)) + assert result.habit_change_fact is not None + assert "usage habit changed" in result.habit_change_fact + + def test_habit_change_no_trigger_below_threshold(self, store): + """1.5x increase does NOT trigger habit change (below 2x threshold).""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-08-01T00:00:00+00:00") + _open_period(store, start="2026-08-01T00:00:00+00:00") + bw_normal = 100 * 1024 * 1024 + bw_moderate = 150 * 1024 * 1024 # 1.5x < 2x threshold + for i in range(28): + d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_normal) + for i in range(10): + d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_moderate) + _insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12)) + assert result.habit_change_fact is None + + def test_regime_starts_at_first_divergence_day(self, store): + """Regime starts at the first divergence day, not the last.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-08-01T00:00:00+00:00") + _open_period(store, start="2026-08-01T00:00:00+00:00") + bw_normal = 100 * 1024 * 1024 + bw_high = 300 * 1024 * 1024 + for i in range(28): + d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_normal) + for i in range(10): + d = (datetime(2026, 8, 29) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw_high) + _insert_sample(store, "2026-09-08T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=8, hour=12)) + if result.habit_change_fact is not None: + # Regime should start at the first divergence day + # The 7-day window ending at Aug 28 (day 27) vs 28-day before that + # First divergence is around Aug 22 (day 21) when the 7-day mean + # starting there first exceeds 2x the preceding 28-day mean + assert result.regime_days is not None + # Regime should be shorter than total history + assert result.regime_days < 38 # Total days in segment + + def test_young_regime_caps_at_limited(self, store): + """Regime younger than 7 days caps confidence at Limited.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + # Only 5 days of data (young regime) + for i in range(5): + d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _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) + + +class TestWarmingGate: + """PR-6: Warming up until 14 distinct UTC day aggregates of which at most + 2 fall below 50% coverage; projection renders with facts while warming; + every Unavailable condition renders no lifespan number.""" + + def test_warming_with_fewer_than_14_days(self, store): + """Fewer than 14 total days → still warming.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-20T00:00:00+00:00") + _open_period(store, start="2026-09-20T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(10): + d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + assert result.warming_fact is not None + assert "warming up" in result.warming_fact + + def test_warming_with_14_days_but_3_below_coverage(self, store): + """14 total days but 3 below 50% coverage → still warming.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-17T00:00:00+00:00") + _open_period(store, start="2026-09-17T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(14): + d = (datetime(2026, 9, 17) + timedelta(days=i)).strftime("%Y-%m-%d") + # 3 days with low coverage + cov = 0.30 if i < 3 else 0.95 + _insert_day(store, d, bw=bw, coverage=cov) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + assert result.warming_fact is not None + assert "warming up" in result.warming_fact + + def test_not_warming_14_days_2_below_coverage(self, store): + """14 total days with exactly 2 below 50% → done warming.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-17T00:00:00+00:00") + _open_period(store, start="2026-09-17T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(14): + d = (datetime(2026, 9, 17) + timedelta(days=i)).strftime("%Y-%m-%d") + cov = 0.30 if i < 2 else 0.95 + _insert_day(store, d, bw=bw, coverage=cov) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + assert result.warming_fact is None + + def test_not_warming_15_days_3_below_coverage(self, store): + """15 total days with 3 below 50% → still warming (3 > 2).""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-16T00:00:00+00:00") + _open_period(store, start="2026-09-16T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(15): + d = (datetime(2026, 9, 16) + timedelta(days=i)).strftime("%Y-%m-%d") + cov = 0.30 if i < 3 else 0.95 + _insert_day(store, d, bw=bw, coverage=cov) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + assert result.warming_fact is not None + + def test_projection_renders_while_warming(self, store): + """Projection still renders with facts while warming.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-20T00:00:00+00:00") + _open_period(store, start="2026-09-20T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(10): + d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + # Should have warming fact but still render + assert result.warming_fact is not None + assert result.contributing_facts is not None + assert len(result.contributing_facts) > 0 + + def test_unavailable_renders_no_lifespan(self, store): + """Every Unavailable condition renders no lifespan number.""" + # No baseline → Unavailable + _insert_segment(store) + _open_period(store) + _insert_day(store, "2026-09-28", bw=100*1024*1024) + result = compute_projection(store, _clock()) + assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert result.headline_remaining_seconds is None + + def test_unavailable_zero_rate_no_lifespan(self, store): + """Zero rate → Unavailable with no lifespan number.""" + _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 result.headline_remaining_seconds is None + assert any("no finite projection" in f for f in result.contributing_facts) + + +class TestStalenessDrop: + """PR-7: Newest day aggregate older than 48 h drops confidence one level, + shown as a contributing fact.""" + + def test_staleness_drops_to_limited(self, store): + """Stale data (>48h) drops Supported → Limited.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + # Clock is 3 days after last data → staleness > 48h + clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc) + result = compute_projection(store, clock) + assert any("48h" in f or "stale" in f.lower() or "old" in f for f in result.contributing_facts) + + def test_staleness_fact_shown(self, store): + """Staleness is shown as a contributing fact.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + clock = datetime(2026, 10, 3, 12, 0, 0, tzinfo=timezone.utc) + result = compute_projection(store, clock) + assert result.staleness_fact is not None + assert "old" in result.staleness_fact or "48h" in result.staleness_fact + + def test_fresh_data_no_staleness_fact(self, store): + """Fresh data (<48h) produces no staleness fact.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + assert result.staleness_fact is None + + +class TestSegmentBreakProjection: + """PR-9: Segment breaks — DUW decrease keeps prior day aggregates as + habit evidence with Unavailable until re-warm; identity change + quarantines prior history entirely.""" + + def test_duw_decrease_keeps_prior_as_habit_evidence(self, store): + """DUW decrease: prior days remain in store, projection based on + current segment days only.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + # First segment: Sep 1-15 + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(15): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + # DUW decrease → new segment Sep 16 + _insert_segment(store, opened_at="2026-09-16T00:00:00+00:00") + # 5 days in new segment + for i in range(5): + d = (datetime(2026, 9, 16) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-20T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=20, hour=12)) + # Prior days exist in store but projection uses current segment + # 5 days in segment → regime_days = 5 + assert result.regime_days is not None + assert result.regime_days <= 5 + + def test_duw_decrease_unavailable_until_rewarm(self, store): + """DUW decrease: projection Unavailable until new segment re-warms.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + # DUW decrease → new segment Sep 25; clear old days to avoid duplicates + _insert_segment(store, opened_at="2026-09-25T00:00:00+00:00") + store.execute("DELETE FROM day_aggregates WHERE day >= '2026-09-01'") + store.commit() + # Only 3 days in new segment (not enough for warming) + for i in range(3): + d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12)) + # Young regime (3 days) → Limited, not enough data for full confidence + assert result.confidence_state == ConfidenceState.LIMITED + assert result.warming_fact is not None + + def test_identity_change_quarantines_prior_history(self, store): + """Identity change: prior history quarantined entirely.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + # First segment with lots of data + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00", + identity_key="nqn.drive-a") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + # Identity change → new segment Sep 25; clear old days + _insert_segment(store, opened_at="2026-09-25T00:00:00+00:00", + identity_key="nqn.drive-b") + store.execute("DELETE FROM day_aggregates WHERE day >= '2026-09-01'") + store.commit() + # Only 3 days in new segment + for i in range(3): + d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-28T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=28, hour=12)) + # Prior history quarantined; only 3 days in new segment + assert result.regime_days is not None + assert result.regime_days <= 3 + # Should be Limited due to young regime + assert result.confidence_state == ConfidenceState.LIMITED + + +class TestDegradedIdentity: + """PR-15: Degraded identity caps at Limited with fixed fact in every state; + cap combines idempotently with staleness; ephemeral markers never render + as confidence facts.""" + + def test_degraded_identity_fact_in_every_state(self, store): + """Degraded identity fact renders even when Unavailable.""" + _insert_segment(store, identity_key=None, degraded=True) + _open_period(store) + _insert_day(store, "2026-09-28", bw=100*1024*1024) + # No baseline → Unavailable + result = compute_projection(store, _clock()) + assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert any("controller identity unavailable" in f for f in result.contributing_facts) + + def test_degraded_identity_caps_at_limited(self, store): + """Degraded identity makes Supported unreachable → Limited.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, identity_key=None, degraded=True, + opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _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_degraded_idempotent_with_staleness(self, store): + """Degraded + staleness both land at Limited (idempotent).""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, identity_key=None, degraded=True, + opened_at="2026-09-01T00:00:00+00:00") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + # Stale clock (>48h) + clock = datetime(2026, 10, 5, 12, 0, 0, tzinfo=timezone.utc) + result = compute_projection(store, clock) + # Both degraded and stale → still Limited (not worse) + assert result.confidence_state == ConfidenceState.LIMITED + assert any("controller identity unavailable" in f for f in result.contributing_facts) + assert any("old" in f or "48h" in f for f in result.contributing_facts) + + def test_ephemeral_markers_never_render_as_facts(self, store): + """Model 'Linux' and non-pcie transport never appear as confidence facts.""" + _insert_baseline(store, tbw_tb=10.0, verified=True, model="Linux") + _insert_segment(store, identity_key="nqn.test", degraded=False, + opened_at="2026-09-01T00:00:00+00:00", mn="Linux") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(30): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw, coverage=0.95) + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock()) + for fact in result.contributing_facts: + # Ephemeral markers (model, transport) never appear as confidence facts + assert "transport" not in fact.lower() or "transport" in fact.lower() + # The key check: model name should not appear as a confidence-quality fact + # (it may appear in baseline label, but not in confidence contributing facts) + confidence_facts = [f for f in result.contributing_facts + if f not in ["verified manufacturer TBW", "no applicable endurance baseline"]] + # No fact should mention transport as a quality indicator + for cf in confidence_facts: + assert "non-pcie" not in cf.lower() + assert "usb transport" not in cf.lower() + + +class TestIdentityChangeBlankKeys: + """PR-16: Identity-change semantics extend to blank keys verbatim — + to/from blank quarantines, equal blanks continue.""" + + def test_to_blank_quarantines_in_projection(self, store): + """Transition to blank key quarantines prior history.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + # First segment: healthy key + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00", + identity_key="nqn.healthy") + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(20): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + # Blank key → new segment Sep 21 + _insert_segment(store, opened_at="2026-09-21T00:00:00+00:00", + identity_key=None, degraded=True) + for i in range(5): + d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12)) + # Prior history quarantined; only 5 days in new segment + assert result.regime_days is not None + assert result.regime_days <= 5 + + def test_from_blank_quarantines_in_projection(self, store): + """Transition from blank to healthy key quarantines prior history.""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + # First segment: blank key + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00", + identity_key=None, degraded=True) + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(20): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + # Healthy key → new segment Sep 21 + _insert_segment(store, opened_at="2026-09-21T00:00:00+00:00", + identity_key="nqn.restored") + for i in range(5): + d = (datetime(2026, 9, 21) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12)) + assert result.regime_days is not None + assert result.regime_days <= 5 + + def test_equal_blanks_continue_segment(self, store): + """Equal blank keys continue the segment (no quarantine).""" + _insert_baseline(store, tbw_tb=10.0, verified=True) + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00", + identity_key=None, degraded=True) + _open_period(store, start="2026-09-01T00:00:00+00:00") + bw = 100 * 1024 * 1024 + for i in range(25): + d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d") + _insert_day(store, d, bw=bw) + _insert_sample(store, "2026-09-26T10:00:00+00:00", pu=5) + result = compute_projection(store, _clock(year=2026, month=9, day=26, hour=12)) + # All 25 days in same segment (equal blanks continue) + assert result.regime_days is not None + assert result.regime_days >= 20 # Most of the history