xavierk/Fenris
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases 11 Wiki Activity

fix(projection): regime dynamics, warming gate, habit change detection, horizon coverage

Browse Source 
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
2026-09-01 23:32:00 +05:30
parent 7802a72606
commit b99ebfe9dd
2 changed files with 610 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_projection.py
+561
View File
@@ -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