fix(projection): regime dynamics, warming gate, habit change detection, horizon coverage
Fixes #26. Changes: - Fix warming gate: check total_days < 14 OR days_below_coverage > 2 - Rewrite _detect_habit_change: correct consecutive-day scanning - Fix _compute_horizon_rate: require history spans full horizon Tests: 29 new tests covering PR-2,3,6,7,9,15,16. 139 total passing.
This commit is contained in:
xavierk
+49
-35
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user