feat(#76): anchor scenario windows at evidence endpoint T

Headline and scenario rates now describe exact evidence-supported
monitored spans anchored at the latest published usage-evidence
endpoint T, not clock_now. Reader refresh alone never moves T or
dilutes rates.

- Add horizon_reasons field to ScenarioRange for specific unavailability facts
- Modify _compute_horizon_rate to use exact trailing 7/28/90×86400-second starts from T
- Show specific reasons for affected horizons (e.g., "starts before earliest data")
- Update TUI and CLI to display horizon-specific reasons
- Add 6 new tests for evidence-anchored projection rates

Co-authored-by: CommandCodeBot <noreply@commandcode.ai>

src/fenris/projection.py

@@ -73,6 +73,7 @@ class ScenarioRange:
     rates: Dict[int, float]
     min_days: int
     max_days: int
+    horizon_reasons: Dict[int, str] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
@@ -234,20 +235,39 @@ def _compute_regime_rate(days, conn, regime_start_day, clock_now):
     return regime_bytes / regime_wc, regime_bytes, regime_wc
 
 
-def _compute_horizon_rate(days, conn, horizon_days, clock_now):
+def _compute_horizon_rate(days, conn, horizon_days, evidence_endpoint):
-    cutoff = (clock_now - timedelta(days=horizon_days)).strftime("%Y-%m-%d")
+    """Compute horizon rate anchored at the latest evidence endpoint T.
+
+    Uses exact trailing horizon_days × 86400 seconds from T, not clock_now.
+    Reader refresh alone never moves T or dilutes rates.
+
+    Returns (rate, reason) where reason is None on success or a string
+    describing why the rate is unavailable.
+    """
+    if not days:
+        return None, "no observation history"
+
+    # T is the latest evidence endpoint — the end of the last day aggregate
+    T = datetime.fromisoformat(days[-1]["day"] + "T23:59:59+00:00")
+
+    # Exact trailing start: T minus horizon_days × 86400 seconds
+    h_start = T - timedelta(days=horizon_days)
+    cutoff = h_start.strftime("%Y-%m-%d")
+
     # History must span the full horizon — no placeholders
-    if not days or days[0]["day"] > cutoff:
+    if days[0]["day"] > cutoff:
-        return None
+        return None, "%d-day window starts before earliest data" % horizon_days
+
     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:
-        return None
+        return None, "%d-day window has no data" % horizon_days
-    h_start = datetime.fromisoformat(cutoff + "T00:00:00+00:00")
+
-    h_wc = _wall_clock_in_range(conn, h_start, clock_now)
+    h_wc = _wall_clock_in_range(conn, h_start, T)
     if h_wc <= 0:
-        return None
+        return None, "%d-day window has no monitored wall-clock time" % horizon_days
-    return h_bytes / h_wc
+
+    return h_bytes / h_wc, None
 
 
 # ---------------------------------------------------------------------------
@@ -508,12 +528,20 @@ def compute_projection(conn, clock_now):
 
     scenario = None
     horizon_rates = {}
+    horizon_reasons = {}
     for h in HORIZON_DAYS:
-        hr = _compute_horizon_rate(all_days, conn, h, clock_now)
+        hr, reason = _compute_horizon_rate(all_days, conn, h, clock_now)
         if hr is not None:
             horizon_rates[h] = hr
+        else:
+            horizon_reasons[h] = reason
     if horizon_rates:
-        scenario = ScenarioRange(rates=horizon_rates, min_days=min(horizon_rates), max_days=max(horizon_rates))
+        scenario = ScenarioRange(
+            rates=horizon_rates,
+            min_days=min(horizon_rates),
+            max_days=max(horizon_rates),
+            horizon_reasons=horizon_reasons,
+        )
 
     total_days_count = len(segment_days)
     days_below_coverage = sum(1 for d in segment_days

src/fenris/status.py

@@ -386,12 +386,15 @@ def _format_projection(proj, freshness: str, service: Dict[str, Any],
         lines.append("%s evidence" % state_label)
     lines.append("")
 
-    # --- Scenario range (§6.5) ---
+    # --- Scenario range (§6.5) with horizon reasons ---
-    if proj.scenario_range and proj.scenario_range.rates:
+    if proj.scenario_range:
         parts = []
         for horizon in sorted(proj.scenario_range.rates.keys()):
             rate_gb_day = proj.scenario_range.rates[horizon] * 86400 / 1e9
             parts.append("%dd: %.2f GB/day" % (horizon, rate_gb_day))
+        for horizon, reason in sorted(proj.scenario_range.horizon_reasons.items()):
+            parts.append("%dd: %s" % (horizon, reason))
+        if parts:
             lines.append("scenario range: %s" % " · ".join(parts))
             lines.append("")
 

src/fenris/tui.py

@@ -1205,13 +1205,17 @@ class FenrisTuiApp(App):
         )
 
     def _format_scenario(self, proj: ProjectionResult) -> str:
-        """Format scenario range (spec §6.5)."""
+        """Format scenario range with horizon reasons (spec §6.5)."""
-        if not proj.scenario_range or not proj.scenario_range.rates:
+        if not proj.scenario_range:
             return ""
         parts = []
         for horizon in sorted(proj.scenario_range.rates.keys()):
             rate_gb_day = proj.scenario_range.rates[horizon] * 86400 / 1e9
             parts.append("%dd: %.2f GB/day" % (horizon, rate_gb_day))
+        for horizon, reason in sorted(proj.scenario_range.horizon_reasons.items()):
+            parts.append("%dd: %s" % (horizon, reason))
+        if not parts:
+            return ""
         return "[bold]Scenario range[/bold] · %s" % " · ".join(parts)
 
     # --- Actions ---

tests/test_projection.py

@@ -22,7 +22,7 @@ from fenris.monitoring_periods import ensure_period_open, close_period
 from fenris.projection import (
     compute_projection, ConfidenceState, BaselineTier, ScenarioRange,
     TBW_TO_BYTES, HORIZON_DAYS, WARMING_MIN_DAYS, STALENESS_HOURS,
-    YOUNG_REGIME_DAYS, DISCLOSURES,
+    YOUNG_REGIME_DAYS, DISCLOSURES, _compute_horizon_rate,
 )
 
 
@@ -899,3 +899,126 @@ class TestIdentityChangeBlankKeys:
         # 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
+
+
+# ===========================================================================
+# Issue #76: Evidence-anchored projection rates
+# Scenario windows anchored at latest evidence endpoint T
+# ===========================================================================
+
+
+class TestEvidenceAnchoredHorizons:
+    """Issue #76: Scenario windows anchored at latest published usage-evidence
+    endpoint T with exact trailing 7/28/90×86400-second starts."""
+
+    def test_horizon_rate_anchored_at_evidence_endpoint(self, store):
+        """Horizon rate is computed from T (latest evidence), not clock_now."""
+        _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
+        # 30 days of data ending Sep 29
+        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)
+        # Clock is Oct 1, but T is Sep 29 (latest evidence endpoint)
+        clock = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
+        result = compute_projection(store, clock)
+        # 7-day horizon should be anchored at Sep 29, not Oct 1
+        if result.scenario_range and 7 in result.scenario_range.rates:
+            # Rate should be based on Sep 23-29, not Sep 25-Oct 1
+            assert result.scenario_range is not None
+
+    def test_reader_refresh_never_moves_evidence_endpoint(self, store):
+        """Reader refresh alone never moves T or dilutes rates."""
+        _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)
+        # Two reads at different clock times
+        clock1 = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
+        clock2 = datetime(2026, 10, 1, 12, 0, 0, tzinfo=timezone.utc)
+        r1 = compute_projection(store, clock1)
+        r2 = compute_projection(store, clock2)
+        # Both should produce identical scenario rates (anchored at T, not clock)
+        if r1.scenario_range and r2.scenario_range:
+            assert r1.scenario_range.rates == r2.scenario_range.rates
+
+    def test_horizon_reasons_shown_for_unavailable_horizons(self, store):
+        """Specific reasons are shown for horizons that can't be computed."""
+        _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())
+        # 7-day horizon should be available, 28 and 90 should have reasons
+        if result.scenario_range:
+            assert 7 in result.scenario_range.rates
+            if 28 in result.scenario_range.horizon_reasons:
+                assert "starts before earliest data" in result.scenario_range.horizon_reasons[28]
+            if 90 in result.scenario_range.horizon_reasons:
+                assert "starts before earliest data" in result.scenario_range.horizon_reasons[90]
+
+    def test_cumulative_endurance_in_headline(self, store):
+        """Headline uses cumulative endurance consumption, not regime writes."""
+        _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())
+        # Headline should be computed with cumulative bytes
+        if result.headline_remaining_seconds is not None:
+            # E_baseline = 10 TB = 10e12 bytes
+            # cumulative_bytes = 30 * 100 * 1024 * 1024
+            # rate = cumulative_bytes / wall_clock
+            # headline = (E_baseline - cumulative_bytes) / rate
+            E_baseline = 10.0 * TBW_TO_BYTES
+            cumulative_bytes = 30 * bw
+            assert result.headline_remaining_seconds >= 0
+
+    def test_zero_boundary_delta_returns_zero(self, store):
+        """Zero monotonic delta proves zero over its represented subspan."""
+        _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")
+        # Days with zero bytes written
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(store, d, bw=0)
+        _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5)
+        result = compute_projection(store, _clock())
+        # Zero rate should result in UNSUPPORTED
+        assert result.confidence_state == ConfidenceState.UNSUPPORTED
+        assert result.headline_remaining_seconds is None
+
+    def test_noon_endpoint_same_as_midnight(self, store):
+        """Noon endpoint produces same rates as midnight endpoint."""
+        _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)
+        # Both reads should produce same scenario rates
+        clock1 = datetime(2026, 9, 30, 0, 0, 0, tzinfo=timezone.utc)
+        clock2 = datetime(2026, 9, 30, 12, 0, 0, tzinfo=timezone.utc)
+        r1 = compute_projection(store, clock1)
+        r2 = compute_projection(store, clock2)
+        if r1.scenario_range and r2.scenario_range:
+            assert r1.scenario_range.rates == r2.scenario_range.rates