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 >= ?",
+        "SELECT COUNT(DISTINCT percentage_used) FROM samples "
-        (segment_opened_at,),
+        "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 state != ConfidenceState.UNSUPPORTED and rate is not None and rate > 0:
-        if tier in (BaselineTier.VERIFIED, BaselineTier.UNVERIFIED):
+        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:
+        elif (tier == BaselineTier.IMPLIED and W_t is not None and p is not None
-            p = _get_latest_pu(conn)
+                and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX):
-            if p is not None and IMPLIED_P_MIN <= p <= IMPLIED_P_MAX:
+            E_baseline = 100 * W_t / p
-                E_baseline = 100 * regime_bytes / p
+        if E_baseline is not None and W_t is not None:
-            else:
+            headline_seconds = max(E_baseline - W_t, 0) / rate
-                E_baseline = None
-        else:
-            E_baseline = None
-        if E_baseline is not None:
-            headline_seconds = max(E_baseline - regime_bytes, 0) / rate
 
     pu_line = _build_pu_context_line(conn, rate, segment_days, clock_now)
 

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) "
+        "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
-        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)",
         (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
     )
     conn.commit()

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) "
+        "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
-        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)",
         (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
     )
     conn.commit()

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) "
+        "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
-        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
-        (ts, "/dev/nvme0n1", 1000000, 500000, pu, 512000000000, 256000000000, 8765),
+        (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,16 +422,18 @@ 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:
+        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)
-            period_end = _clock()
+        actual_wc = int((_clock() - period_start).total_seconds())
-            actual_wc = int((period_end - period_start).total_seconds())
         rate = regime_bytes / actual_wc
-            expected = max(E_rated - regime_bytes, 0) / rate
+        # 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):
@@ -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,11 +454,11 @@ 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_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
 
 
-
 # ===========================================================================
 # Issue #26: Project from the sustained regime
 # Habit change, scenario range, and evidence gates
@@ -394,16 +480,18 @@ 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:
+        assert result.headline_remaining_seconds is not None
         E = 10.0 * TBW_TO_BYTES
-            expected_seconds = max(E - regime_bytes, 0) / expected_rate
+        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):

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) "
+        "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) "
-        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 1)",
         (ts, device, 1000000, 500000, pu, 512000000000, 256000000000, 8765),
     )
     conn.commit()