From 96fd1702fac73445378c5393a036aa1f8cf68ccb Mon Sep 17 00:00:00 2001 From: xavierk Date: Mon, 5 Oct 2026 19:45:38 +0530 Subject: [PATCH] fix(projection): align projection with ADR 0002 Use lifetime written bytes from the current segment's newest published sample in the headline formula, select the Percentage-Used-implied baseline after two increments in the segment (otherwise show the too-coarse fact), and judge horizon agreement across existing horizons. Make the vacuous projection tests assert unconditionally and add coverage for implied-baseline gating and single-horizon agreement. --- src/fenris/projection.py | 69 +++++++--- tests/test_acceptance_sweep.py | 4 +- tests/test_complete_observation_day_gate.py | 4 +- tests/test_projection.py | 132 ++++++++++++++++---- tests/test_tui.py | 4 +- 5 files changed, 168 insertions(+), 45 deletions(-) diff --git a/src/fenris/projection.py b/src/fenris/projection.py index 9e09da0..96543ff 100644 --- a/src/fenris/projection.py +++ b/src/fenris/projection.py @@ -150,16 +150,40 @@ def _get_all_days(conn): for r in cursor.fetchall()] +def _get_latest_published_sample(conn, segment_id): + """Newest published sample in the segment: (lifetime bytes written, PU). + + W_t is the lifetime counter, not a regime delta. Pending staging lives + in a separate table (ADR 0011), so only published samples are seen here. + bytes_written is DUW x 512000 as stored by the collector; fall back to + that conversion if only the raw counter is present. + """ + if segment_id is not None: + cursor = conn.execute( + "SELECT COALESCE(bytes_written, data_units_written * 512000), percentage_used " + "FROM samples WHERE segment_id = ? ORDER BY id DESC LIMIT 1", + (segment_id,), + ) + else: + cursor = conn.execute( + "SELECT COALESCE(bytes_written, data_units_written * 512000), percentage_used " + "FROM samples ORDER BY id DESC LIMIT 1" + ) + row = cursor.fetchone() + return (row[0], row[1]) if row else (None, None) + + 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): +def _get_pu_increments_in_segment(conn, segment_id): cursor = conn.execute( - "SELECT COUNT(DISTINCT percentage_used) FROM samples WHERE ts >= ?", - (segment_opened_at,), + "SELECT COUNT(DISTINCT percentage_used) FROM samples " + "WHERE segment_id = ? AND percentage_used IS NOT NULL", + (segment_id,), ) row = cursor.fetchone() return max(0, (row[0] if row else 0) - 1) @@ -186,11 +210,24 @@ def _wall_clock_in_range(conn, start, end): # Baseline resolution (§6.1, §6.2) # --------------------------------------------------------------------------- +def _resolve_implied_baseline(conn, current_segment, facts): + """No override exists: Percentage-Used-implied tier, or unavailable (§10).""" + if current_segment is not None: + _, p = _get_latest_published_sample(conn, current_segment["id"]) + increments = _get_pu_increments_in_segment(conn, current_segment["id"]) + if (increments >= IMPLIED_MIN_PU_INCREMENTS and p is not None + and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX): + return (BaselineTier.IMPLIED, None, + "implied from vendor wear (%d%% used) — coarse" % p, facts) + facts.append("vendor wear estimate too coarse to imply endurance") + return BaselineTier.NONE, None, "no baseline", facts + + def _resolve_baseline(conn, current_segment): baseline = _get_baseline(conn) facts: list[str] = [] if baseline is None: - return BaselineTier.NONE, None, "no baseline", facts + return _resolve_implied_baseline(conn, current_segment, facts) mandatory = [baseline["source_url"], baseline["document_revision"], baseline["entry_date"], baseline["model_string"], @@ -386,7 +423,7 @@ def _evaluate_confidence(tier, rate, regime_days, days, current_segment, supported_facts.append("recent data") # 5. Horizon agreement - if scenario_range is not None and len(scenario_range.rates) >= 2: + if scenario_range is not None and len(scenario_range.rates) >= 1: rl = list(scenario_range.rates.values()) if min(rl) > 0 and max(rl) / min(rl) > HORIZON_AGREEMENT_FACTOR: failing = True @@ -633,19 +670,17 @@ def compute_projection(conn, clock_now): 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): + if state != ConfidenceState.UNSUPPORTED and rate is not None and rate > 0: + W_t, p = _get_latest_published_sample( + conn, current_segment["id"] if current_segment else None) + E_baseline = None + if tier in (BaselineTier.VERIFIED, BaselineTier.UNVERIFIED) and baseline is not None: 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 + elif (tier == BaselineTier.IMPLIED and W_t is not None and p is not None + and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX): + E_baseline = 100 * W_t / p + if E_baseline is not None and W_t is not None: + headline_seconds = max(E_baseline - W_t, 0) / rate pu_line = _build_pu_context_line(conn, rate, segment_days, clock_now) diff --git a/tests/test_acceptance_sweep.py b/tests/test_acceptance_sweep.py index e4b8911..070099b 100644 --- a/tests/test_acceptance_sweep.py +++ b/tests/test_acceptance_sweep.py @@ -98,8 +98,8 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24): 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 (?, ?, ?, ?, ?, ?, ?, ?)", + "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)", (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765), ) conn.commit() diff --git a/tests/test_complete_observation_day_gate.py b/tests/test_complete_observation_day_gate.py index c541538..4f075e3 100644 --- a/tests/test_complete_observation_day_gate.py +++ b/tests/test_complete_observation_day_gate.py @@ -78,8 +78,8 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24): 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 (?, ?, ?, ?, ?, ?, ?, ?)", + "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)", (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765), ) conn.commit() diff --git a/tests/test_projection.py b/tests/test_projection.py index cd135e9..e3cd585 100644 --- a/tests/test_projection.py +++ b/tests/test_projection.py @@ -74,12 +74,13 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24): conn.commit() -def _insert_sample(conn, ts, pu=5): +def _insert_sample(conn, ts, pu=5, bytes_written=512000000000, segment_id=1): 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), + "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (ts, "/dev/nvme0n1", bytes_written // 512000, 500000, pu, bytes_written, + 256000000000, 8765, segment_id), ) conn.commit() @@ -223,6 +224,88 @@ class TestImpliedBaseline: result = compute_projection(store, _clock()) assert result.baseline_tier == BaselineTier.VERIFIED + def _setup_implied(self, store, pus): + """No override baseline; one sample per PU value inside the segment.""" + _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) + for i, pu in enumerate(pus): + _insert_sample(store, "2026-09-%02dT10:00:00+00:00" % (20 + i), pu=pu) + _insert_complete_local_days(store, "2026-09-29", 1) + + @pytest.mark.parametrize("pus", [[5], [5, 5, 5], [5, 6]]) + def test_implied_unavailable_before_two_increments(self, store, pus): + self._setup_implied(store, pus) + result = compute_projection(store, _clock()) + assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert result.baseline_tier == BaselineTier.NONE + assert result.headline_remaining_seconds is None + assert "vendor wear estimate too coarse to imply endurance" in result.contributing_facts + + def test_implied_unavailable_ignores_increments_in_other_segments(self, store): + _insert_segment(store, opened_at="2026-09-01T00:00:00+00:00") + _insert_segment(store, opened_at="2026-09-15T00:00:00+00:00", identity_key="nqn.new") + _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) + # Two increments in the old segment, none in the current one + _insert_sample(store, "2026-09-02T10:00:00+00:00", pu=3, segment_id=1) + _insert_sample(store, "2026-09-03T10:00:00+00:00", pu=4, segment_id=1) + _insert_sample(store, "2026-09-20T10:00:00+00:00", pu=5, segment_id=2) + _insert_complete_local_days(store, "2026-09-29", 1) + result = compute_projection(store, _clock()) + assert result.baseline_tier == BaselineTier.NONE + assert "vendor wear estimate too coarse to imply endurance" in result.contributing_facts + + def test_implied_unavailable_when_pu_out_of_range(self, store): + self._setup_implied(store, [0, 1, 2]) + store.execute("UPDATE samples SET percentage_used = 255 WHERE id = 3") + store.commit() + result = compute_projection(store, _clock()) + assert result.baseline_tier == BaselineTier.NONE + assert "vendor wear estimate too coarse to imply endurance" in result.contributing_facts + + def test_implied_eligible_after_two_increments(self, store): + self._setup_implied(store, [5, 6, 7]) + W_t = 512000000000 + result = compute_projection(store, _clock()) + assert result.baseline_tier == BaselineTier.IMPLIED + # Implied baseline is never Supported-eligible + assert result.confidence_state == ConfidenceState.LIMITED + assert "vendor wear estimate too coarse to imply endurance" not in result.contributing_facts + E_implied = 100 * W_t / 7 + regime_bytes = 20 * 1024 * 1024 * 100 + regime_start = datetime(2026, 9, 10, 0, 0, 0, tzinfo=timezone.utc) + rate = regime_bytes / int((_clock() - regime_start).total_seconds()) + assert abs(result.headline_remaining_seconds - max(E_implied - W_t, 0) / rate) < 1.0 + + def test_override_wins_over_implied(self, store): + _insert_baseline(store, tbw_tb=1.0, verified=True) + self._setup_implied(store, [5, 6, 7]) + result = compute_projection(store, _clock()) + assert result.baseline_tier == BaselineTier.VERIFIED + + +class TestHorizonAgreement: + def test_single_horizon_passes_agreement(self, store): + """ADR 0002 §8: agreement is judged across existing horizons; one passes.""" + _insert_baseline(store, tbw_tb=1.0, verified=True) + _insert_segment(store) + _open_period(store) + # 14 days covers only the 7-day horizon + for i in range(14): + d = (datetime(2026, 9, 17) + 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) + _insert_complete_local_days(store, "2026-09-29", 1) + result = compute_projection(store, _clock()) + assert list(result.scenario_range.rates) == [7] + assert not any("regime only" in f for f in result.contributing_facts) + assert result.confidence_state == ConfidenceState.SUPPORTED + class TestZeroRate: def test_zero_rate_fixed_phrase(self, store): @@ -339,17 +422,19 @@ class TestArithmetic: 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) + W_t = 512000000000 # lifetime written bytes of the newest sample + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5, bytes_written=W_t) + _insert_complete_local_days(store, "2026-09-29", 1) 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 + assert 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) + actual_wc = int((_clock() - period_start).total_seconds()) + rate = regime_bytes / actual_wc + # W_t is lifetime written bytes, not the regime delta + expected = max(E_rated - W_t, 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) @@ -359,6 +444,7 @@ class TestArithmetic: 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) + _insert_complete_local_days(store, "2026-09-29", 1) r_slow = compute_projection(store, _clock()) store.execute("DELETE FROM day_aggregates") @@ -368,9 +454,9 @@ class TestArithmetic: _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 - + assert r_slow.headline_remaining_seconds is not None + assert r_fast.headline_remaining_seconds is not None + assert r_fast.headline_remaining_seconds < r_slow.headline_remaining_seconds # =========================================================================== @@ -394,17 +480,19 @@ class TestSustainedRegimeRate: 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) + W_t = 512000000000 + _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5, bytes_written=W_t) + _insert_complete_local_days(store, "2026-09-29", 1) 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 + assert result.headline_remaining_seconds is not None + E = 10.0 * TBW_TO_BYTES + expected_seconds = max(E - W_t, 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.""" diff --git a/tests/test_tui.py b/tests/test_tui.py index 4770b0c..3027469 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -107,8 +107,8 @@ def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24): def _insert_sample(conn, ts, pu=5, device="/dev/nvme0n1"): conn.execute( "INSERT INTO samples (ts, device, data_units_written, data_units_read, " - "percentage_used, bytes_written, bytes_read, power_on_hours) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)", (ts, device, 1000000, 500000, pu, 512000000000, 256000000000, 8765), ) conn.commit()