From e26a8511152054afec715719b901af8567b216c4 Mon Sep 17 00:00:00 2001 From: xavierk Date: Mon, 5 Oct 2026 19:57:46 +0530 Subject: [PATCH] feat(projection): show staged projection with evidence ladder Report a projection stage, an evidence ladder and the sustained-regime write rate from compute_projection, per ADR 0012. A provisional lifespan appears after 3 observed hours and carries the hours observed and a short-horizon spread; the complete-local-day condition becomes a fact instead of a gate, and the write rate replaces the lifespan when no baseline applies. The TUI and CLI render the stage, ladder count and full ladder, and drop their own two-sample pre-gates. --- CHANGELOG.md | 10 + src/fenris/projection.py | 292 +++++++++++++------- src/fenris/status.py | 35 +-- src/fenris/tui.py | 31 ++- tests/test_complete_observation_day_gate.py | 42 +-- tests/test_projection.py | 20 +- tests/test_staged_projection.py | 221 +++++++++++++++ tests/test_staged_projection_callers.py | 112 ++++++++ 8 files changed, 615 insertions(+), 148 deletions(-) create mode 100644 tests/test_staged_projection.py create mode 100644 tests/test_staged_projection_callers.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a1b17c..e9503b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,16 @@ backfill releases from before this changelog. ## [Unreleased] +### Added + +- Show a provisional usage-adjusted theoretical lifespan after 3 observed hours, with the hours observed and a short-horizon spread, instead of waiting for a full local day. +- Show the projection stage and an evidence ladder of Supported conditions, each met or unmet with its reason, so confidence visibly strengthens with each collection run. +- Show the sustained-regime write rate when no endurance baseline applies. + +### Changed + +- Treat the complete-local-day requirement as a contributing fact rather than a gate, and stop the TUI and CLI from deciding when a projection can be shown. + ## [0.6.0] - 2026-09-29 ### Added diff --git a/src/fenris/projection.py b/src/fenris/projection.py index 96543ff..3dd02e9 100644 --- a/src/fenris/projection.py +++ b/src/fenris/projection.py @@ -20,6 +20,11 @@ Arithmetic (§6.3): E_rated = entered_TBW × 10¹² bytes E_implied = 100 · W_t / p (1 ≤ p ≤ 254) +Staging (ADR 0012): the result carries a projection stage, the evidence +ladder (each Supported condition met or unmet with its reason) and the +sustained-regime write rate, so callers render the stage and never +pre-gate the projection themselves. + Criteria: PR-1–PR-17, CI-4. """ import sqlite3 @@ -35,6 +40,9 @@ from .monitoring_periods import interval_within_one_monitoring_period # --------------------------------------------------------------------------- HORIZON_DAYS = (7, 28, 90) +SHORT_HORIZON_DAYS = (1, 3) +PROVISIONAL_MIN_HOURS = 3 +DAILY_CYCLE_HOURS = 24 TBW_TO_BYTES = 10 ** 12 IMPLIED_P_MIN = 1 IMPLIED_P_MAX = 254 @@ -62,6 +70,15 @@ class ConfidenceState(Enum): SUPPORTED = "Supported" +class ProjectionStage(Enum): + NO_OBSERVATIONS = "No observations yet" + PROVISIONAL = "Provisional" + WARMING = "Warming" + LIMITED = "Limited" + SUPPORTED = "Supported" + UNAVAILABLE = "Unavailable" + + class BaselineTier(Enum): VERIFIED = "verified_override" UNVERIFIED = "unverified_override" @@ -77,6 +94,14 @@ class ScenarioRange: horizon_reasons: Dict[int, str] = field(default_factory=dict) +@dataclass(frozen=True) +class LadderCondition: + """One Supported condition from ADR 0002 §8, met or unmet with its reason.""" + name: str + met: bool + reason: str + + @dataclass(frozen=True) class ProjectionResult: confidence_state: ConfidenceState @@ -94,6 +119,25 @@ class ProjectionResult: degraded_identity_fact: Optional[str] zero_rate_fact: Optional[str] qualifying_days_progress: Optional[str] = None # Issue #77: honest qualifying-day progress + stage: ProjectionStage = ProjectionStage.NO_OBSERVATIONS + evidence_ladder: List[LadderCondition] = field(default_factory=list) + write_rate: Optional[float] = None # sustained-regime rate, bytes per second + hours_observed: int = 0 + + @property + def stage_label(self) -> str: + """Display wording; observations that are too few are not "none".""" + if self.stage == ProjectionStage.NO_OBSERVATIONS and self.hours_observed > 0: + return "Collecting observations" + return self.stage.value + + @property + def ladder_met(self) -> int: + return sum(1 for c in self.evidence_ladder if c.met) + + @property + def ladder_total(self) -> int: + return len(self.evidence_ladder) # --------------------------------------------------------------------------- @@ -206,6 +250,28 @@ def _wall_clock_in_range(conn, start, end): return total +def _get_hours_observed(conn, segment_opened_at, days): + """Hours of sampled evidence in the current segment. + + Counted from hour observations. A day aggregate without any hour rows + (imported or pruned history) stands for a whole observed day. + """ + hour_floor = segment_opened_at[:13] if segment_opened_at else "" + cursor = conn.execute( + "SELECT substr(hour, 1, 10), COUNT(*) FROM hour_observations " + "WHERE sample_count > 0 AND hour >= ? GROUP BY 1", + (hour_floor,), + ) + by_day = dict(cursor.fetchall()) + total = 0 + for d in days: + if d["day"] in by_day: + total += by_day.pop(d["day"]) + elif d["sample_count"] > 0: + total += DAILY_CYCLE_HOURS + return total + sum(by_day.values()) + + # --------------------------------------------------------------------------- # Baseline resolution (§6.1, §6.2) # --------------------------------------------------------------------------- @@ -378,84 +444,88 @@ def _detect_habit_change(days): # Confidence rule table (§6.7) # --------------------------------------------------------------------------- -def _evaluate_confidence(tier, rate, regime_days, days, current_segment, - clock_now, warming_days, warming_low_coverage, - habit_change, staleness_hours, scenario_range): - facts = [] +DEGRADED_IDENTITY_FACT = "controller identity unavailable — replacement detection relies on write-counter continuity only" - if tier == BaselineTier.NONE: - facts.append("no applicable endurance baseline") - return ConfidenceState.UNSUPPORTED, facts - if rate is None or rate <= 0: - facts.append("no finite projection from this history") - return ConfidenceState.UNSUPPORTED, facts +def _build_evidence_ladder(tier, regime_days, days, current_segment, + staleness_hours, scenario_range): + """The eight Supported conditions of ADR 0002 §8, each met or unmet.""" + ladder = [] - supported_facts = [] - failing = False + def add(name, met, met_reason, unmet_reason): + ladder.append(LadderCondition(name, met, met_reason if met else unmet_reason)) - # 1. Verified baseline - if tier != BaselineTier.VERIFIED: - failing = True - else: - supported_facts.append("verified manufacturer TBW") + add("verified baseline", tier == BaselineTier.VERIFIED, + "verified manufacturer TBW", + {BaselineTier.UNVERIFIED: "baseline is unverified (user-supplied)", + BaselineTier.IMPLIED: "baseline is implied from vendor wear — coarse"}.get( + tier, "no applicable endurance baseline")) - # 2. >= 14 qualifying days - qualifying = sum(1 for d in days if d["coverage"] >= WARMING_COVERAGE_FLOOR and d["sample_count"] > 0) - if qualifying < WARMING_MIN_DAYS: - failing = True - else: - supported_facts.append("%d calendar days" % qualifying) + qualifying = sum(1 for d in days + if d["coverage"] >= WARMING_COVERAGE_FLOOR and d["sample_count"] > 0) + add("14 qualifying days", qualifying >= WARMING_MIN_DAYS, + "%d calendar days" % qualifying, + "%d of %d qualifying days" % (qualifying, WARMING_MIN_DAYS)) - # 3. Coverage >= 80% total_wc = len(days) * 86400 total_known = sum(int(d["coverage"] * 86400) for d in days) avg_cov = total_known / total_wc if total_wc > 0 else 0.0 - if avg_cov < SUPPORTED_COVERAGE_FLOOR: - failing = True + add("coverage at least 80%", avg_cov >= SUPPORTED_COVERAGE_FLOOR, + "%d%% interval coverage" % int(avg_cov * 100), + "interval coverage %d%% (≥%d%% required)" + % (int(avg_cov * 100), int(SUPPORTED_COVERAGE_FLOOR * 100))) + + if staleness_hours is None: + add("fresh data", False, "", "no observation days yet") else: - supported_facts.append("%d%% interval coverage" % int(avg_cov * 100)) + add("fresh data", staleness_hours <= STALENESS_HOURS, "recent data", + "newest data %dh old (≥%dh)" % (staleness_hours, STALENESS_HOURS)) - # 4. Fresh (< 48h) - if staleness_hours is not None and staleness_hours > STALENESS_HOURS: - failing = True - elif staleness_hours is not None: - supported_facts.append("recent data") - - # 5. Horizon agreement - if scenario_range is not None and len(scenario_range.rates) >= 1: + if scenario_range is None or not scenario_range.rates: + add("horizons agree", False, "", "no scenario horizon covered yet") + else: rl = list(scenario_range.rates.values()) - if min(rl) > 0 and max(rl) / min(rl) > HORIZON_AGREEMENT_FACTOR: - failing = True - else: - supported_facts.append("%d weekly cycles" % len(scenario_range.rates)) - else: - failing = True + spread = max(rl) / min(rl) if min(rl) > 0 else None + agree = spread is None or spread <= HORIZON_AGREEMENT_FACTOR + add("horizons agree", agree, "%d weekly cycles" % len(rl), + "horizon rates differ by %.1f× (≤%d× required)" + % (spread or 0.0, HORIZON_AGREEMENT_FACTOR)) - # 6. Burst guard - if not failing and len(days) >= BURST_GUARD_LOOKBACK: - t28 = sum(d["bytes_written"] for d in days[-BURST_GUARD_LOOKBACK:]) - for d in days[-BURST_GUARD_LOOKBACK:]: - if t28 > 0 and d["bytes_written"] >= BURST_GUARD_FRACTION * t28: - failing = True - break - if not failing: - supported_facts.append("no burst days") + recent = days[-BURST_GUARD_LOOKBACK:] + t28 = sum(d["bytes_written"] for d in recent) + burst = t28 > 0 and any(d["bytes_written"] >= BURST_GUARD_FRACTION * t28 for d in recent) + add("no burst day", bool(recent) and not burst, "no burst days", + "one day carries ≥%d%% of trailing writes" % int(BURST_GUARD_FRACTION * 100) + if burst else "no observation days yet") - # 7. Regime >= 7 days - if regime_days < YOUNG_REGIME_DAYS: - failing = True + add("regime at least 7 days", regime_days >= YOUNG_REGIME_DAYS, + "regime %d days" % regime_days, + "regime only %d days old (≥%d required)" % (regime_days, YOUNG_REGIME_DAYS)) - # 8. Degraded identity - if current_segment and current_segment.get("identity_degraded"): - failing = True - facts.append("controller identity unavailable — replacement detection relies on write-counter continuity only") + degraded = bool(current_segment and current_segment.get("identity_degraded")) + add("controller identity", not degraded, "controller identity available", + DEGRADED_IDENTITY_FACT) + return ladder - if not failing: - return ConfidenceState.SUPPORTED, supported_facts - # Limited - limited_facts = list(supported_facts) +def _evaluate_confidence(tier, rate, regime_days, days, current_segment, + habit_change, staleness_hours, scenario_range): + """Return (confidence state, facts, evidence ladder).""" + ladder = _build_evidence_ladder(tier, regime_days, days, current_segment, + staleness_hours, scenario_range) + + if tier == BaselineTier.NONE: + return ConfidenceState.UNSUPPORTED, ["no applicable endurance baseline"], ladder + + if rate is None or rate <= 0: + return ConfidenceState.UNSUPPORTED, ["no finite projection from this history"], ladder + + met_facts = [c.reason for c in ladder if c.met] + if all(c.met for c in ladder): + return ConfidenceState.SUPPORTED, met_facts, ladder + + # Limited: met facts plus the contributing failures + limited_facts = list(met_facts) if staleness_hours is not None and staleness_hours > STALENESS_HOURS: limited_facts.append("newest data %dh old (≥48h)" % staleness_hours) if habit_change is not None: @@ -463,11 +533,27 @@ def _evaluate_confidence(tier, rate, regime_days, days, current_segment, if regime_days < YOUNG_REGIME_DAYS: limited_facts.append("regime only %d days old (≥7 required)" % regime_days) if current_segment and current_segment.get("identity_degraded"): - degraded_fact = "controller identity unavailable — replacement detection relies on write-counter continuity only" - if degraded_fact not in limited_facts: - limited_facts.append(degraded_fact) + if DEGRADED_IDENTITY_FACT not in limited_facts: + limited_facts.append(DEGRADED_IDENTITY_FACT) + return ConfidenceState.LIMITED, limited_facts, ladder - return ConfidenceState.LIMITED, limited_facts + +def ladder_count_text(result): + """Headline form of the evidence ladder: how many conditions are met.""" + return "%d of %d conditions met" % (result.ladder_met, result.ladder_total) + + +def ladder_lines(result): + """Full evidence ladder for the outlook, one line per condition.""" + return ["[%s] %s — %s" % ("x" if c.met else " ", c.name, c.reason) + for c in result.evidence_ladder] + + +def write_rate_gb_day(result): + """The sustained-regime write rate in GB/day, or None when not yet known.""" + if result.write_rate is None: + return None + return result.write_rate * 86400 / 1e9 # --------------------------------------------------------------------------- @@ -548,6 +634,21 @@ def _has_complete_local_day(conn): # Main projection function # --------------------------------------------------------------------------- +def _determine_stage(state, hours_observed, total_days, days_below_coverage): + """Where the projection stands on its way to support (ADR 0012 §1, §2).""" + if hours_observed < PROVISIONAL_MIN_HOURS: + return ProjectionStage.NO_OBSERVATIONS + if state == ConfidenceState.UNSUPPORTED: + return ProjectionStage.UNAVAILABLE + if hours_observed < DAILY_CYCLE_HOURS: + return ProjectionStage.PROVISIONAL + if total_days < WARMING_MIN_DAYS or days_below_coverage > WARMING_MAX_LOW_COVERAGE: + return ProjectionStage.WARMING + if state == ConfidenceState.SUPPORTED: + return ProjectionStage.SUPPORTED + return ProjectionStage.LIMITED + + def compute_projection(conn, clock_now): facts = [] habit_change_fact = None @@ -560,32 +661,14 @@ def compute_projection(conn, clock_now): tier, baseline, baseline_label, baseline_facts = _resolve_baseline(conn, current_segment) facts.extend(baseline_facts) - # --- Complete observation day gate (issue #94) --- - # An endurance outlook requires at least one complete local - # midnight-to-midnight calendar day with usable observation evidence. - has_complete_day = _has_complete_local_day(conn) - if not has_complete_day: + # The complete-local-day condition is a fact, not a gate (ADR 0012 §2). + if not _has_complete_local_day(conn): facts.append("waiting for a full local observation day") - return ProjectionResult( - confidence_state=ConfidenceState.UNSUPPORTED, - contributing_facts=facts, - headline_remaining_seconds=None, - scenario_range=None, - pu_context_line="Percentage Used: unknown" if _get_latest_pu(conn) is None else "Percentage Used: %d%%" % (_get_latest_pu(conn) or 0), - disclosure_text=list(DISCLOSURES), - baseline_tier=tier, - baseline_label=baseline_label, - regime_days=None, - habit_change_fact=None, - warming_fact=None, - staleness_fact=None, - degraded_identity_fact=None, - zero_rate_fact=None, - qualifying_days_progress=None, - ) segment_days = _get_days_in_segment(conn, current_segment["opened_at"]) if current_segment else _get_all_days(conn) all_days = _get_all_days(conn) + hours_observed = _get_hours_observed( + conn, current_segment["opened_at"] if current_segment else None, segment_days) regime_start_day = None habit_change = None @@ -615,12 +698,15 @@ def compute_projection(conn, clock_now): scenario = None horizon_rates = {} horizon_reasons = {} - for h in HORIZON_DAYS: + for h in SHORT_HORIZON_DAYS + HORIZON_DAYS: hr, reason = _compute_horizon_rate(all_days, conn, h, clock_now) if hr is not None: horizon_rates[h] = hr - else: + elif h in HORIZON_DAYS: horizon_reasons[h] = reason + if HORIZON_DAYS[0] in horizon_rates: + for h in SHORT_HORIZON_DAYS: + horizon_rates.pop(h, None) if horizon_rates: scenario = ScenarioRange( rates=horizon_rates, @@ -656,14 +742,30 @@ def compute_projection(conn, clock_now): facts.append(staleness_fact) if current_segment and current_segment.get("identity_degraded"): - degraded_identity_fact = "controller identity unavailable — replacement detection relies on write-counter continuity only" + degraded_identity_fact = DEGRADED_IDENTITY_FACT facts.append(degraded_identity_fact) - state, conf_facts = _evaluate_confidence( - tier, rate, regime_days_count, segment_days, current_segment, clock_now, - qualifying, 0, habit_change, staleness_hours, scenario, + state, conf_facts, ladder = _evaluate_confidence( + tier, rate, regime_days_count, segment_days, current_segment, + habit_change, staleness_hours, scenario, ) + stage = _determine_stage(state, hours_observed, total_days_count, days_below_coverage) + if stage == ProjectionStage.NO_OBSERVATIONS: + # Too little evidence for any number: no rate, no lifespan. + state = ConfidenceState.UNSUPPORTED + rate = None + scenario = None + if hours_observed > 0: + facts.insert(0, "%d of %d hours observed before the first projection" + % (hours_observed, PROVISIONAL_MIN_HOURS)) + elif stage == ProjectionStage.PROVISIONAL: + # Short-horizon spread: the rate over the hours observed so far. + scenario = ScenarioRange(rates={SHORT_HORIZON_DAYS[0]: rate}, + min_days=SHORT_HORIZON_DAYS[0], + max_days=SHORT_HORIZON_DAYS[0]) + facts[:0] = ["%d hours observed" % hours_observed, "daily cycle not yet seen"] + all_facts = list(facts) for cf in conf_facts: if cf not in all_facts: @@ -700,4 +802,8 @@ def compute_projection(conn, clock_now): degraded_identity_fact=degraded_identity_fact, zero_rate_fact=zero_rate_fact, qualifying_days_progress=qualifying_days_progress, + stage=stage, + evidence_ladder=ladder, + write_rate=rate, + hours_observed=hours_observed, ) diff --git a/src/fenris/status.py b/src/fenris/status.py index 147004e..f429701 100644 --- a/src/fenris/status.py +++ b/src/fenris/status.py @@ -23,7 +23,10 @@ from typing import Any, Dict, Iterator, List, Optional, Tuple, TYPE_CHECKING if TYPE_CHECKING: from .status_composition import StatusComposition -from .projection import compute_projection, ConfidenceState, DISCLOSURES +from .projection import ( + compute_projection, ConfidenceState, DISCLOSURES, ProjectionStage, ladder_count_text, + ladder_lines, write_rate_gb_day, +) from .store import SCHEMA_VERSION from .init_system import ( query_service_state as _init_query_service_state, @@ -255,8 +258,7 @@ def check_retired_flag(flag: str) -> Optional[str]: # --------------------------------------------------------------------------- def _format_projection(proj, freshness: str, drive_facts: List[str], - config_error: Optional[str], - sample_count: int = 0, day_count: int = 0) -> str: + config_error: Optional[str]) -> str: """Format projection details; monitoring status has its own renderer.""" lines = [] @@ -272,15 +274,6 @@ def _format_projection(proj, freshness: str, drive_facts: List[str], lines.append("Enable monitoring: fenris monitor resume") return "\n".join(lines) - # --- Single sample: awaiting another sample (issue #73 AC3) --- - # Only show awaiting state when there are no day aggregates (e.g., legacy import - # or hand-crafted stores can have 1 sample but sufficient day data for projection) - if sample_count <= 1 and day_count == 0: - lines.append("awaiting another sample") - lines.append("") - lines.append("Collecting usage data — the first projection requires at least two samples.") - return "\n".join(lines) - if proj is None: lines.append("no projection available") return "\n".join(lines) @@ -290,6 +283,9 @@ def _format_projection(proj, freshness: str, drive_facts: List[str], lines.append(headline) lines.append("") + # --- Stage and ladder count (ADR 0012) --- + lines.append("Projection stage: %s · %s" % (proj.stage_label, ladder_count_text(proj))) + # --- Confidence state + contributing facts (§6.7, §6.11) --- state_label = proj.confidence_state.value facts_list = list(proj.contributing_facts) if proj.contributing_facts else [] @@ -305,6 +301,11 @@ def _format_projection(proj, freshness: str, drive_facts: List[str], lines.append("%s evidence" % state_label) lines.append("") + # --- Full evidence ladder in the outlook (ADR 0012 §5) --- + lines.append("Evidence ladder") + lines.extend(" " + line for line in ladder_lines(proj)) + lines.append("") + # --- Scenario range (§6.5) with horizon reasons --- if proj.scenario_range: parts = [] @@ -333,8 +334,13 @@ def _format_projection(proj, freshness: str, drive_facts: List[str], def _format_headline(proj) -> str: """Format the lifespan headline or its no-projection wording (§6.11).""" if proj.headline_remaining_seconds is None: + if proj.stage == ProjectionStage.NO_OBSERVATIONS: + return "no projection yet — collecting observations" if proj.zero_rate_fact: return "no finite projection from this history" + rate = write_rate_gb_day(proj) + if rate is not None: + return "write rate: %.2f GB/day · no lifespan without an endurance baseline" % rate if proj.warming_fact: return proj.warming_fact return "no projection available" @@ -507,10 +513,7 @@ def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] proj = compute_projection(conn, clock_now) except (sqlite3.Error, ValueError, TypeError): pass - parts.append(_format_projection( - proj, comp.freshness, drive_facts, config_error, - comp.sample_count, comp.day_count, - )) + parts.append(_format_projection(proj, comp.freshness, drive_facts, config_error)) if query_journal and ( comp.store_fault or comp.last_collect_ok is False or comp.freshness in ("missed", "stale") diff --git a/src/fenris/tui.py b/src/fenris/tui.py index bb0bdf1..45fa7b9 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -18,6 +18,7 @@ from pathlib import Path from typing import Any, Callable, Dict, List, Optional from zoneinfo import ZoneInfo +from rich.markup import escape from rich.text import Text from textual.app import App, ComposeResult from textual.binding import Binding @@ -38,8 +39,12 @@ from .activity_plot import VolumePoint, volume_plot from .projection import ( ConfidenceState, ProjectionResult, + ProjectionStage, ScenarioRange, compute_projection, + ladder_count_text, + ladder_lines, + write_rate_gb_day, ) from .status import ( CADENCE_DEFAULT_S, @@ -2061,12 +2066,6 @@ class FenrisTuiApp(App): self._render_headline( "[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT ) - elif comp.sample_count <= 1 and comp.day_count == 0: - # Single sample: awaiting another sample - self._render_headline( - "[bold]Awaiting another sample[/bold]\n\n" - "Collecting usage data — the first projection requires at least two samples." - ) else: try: proj = compute_projection(conn, self._clock_now) @@ -2076,7 +2075,8 @@ class FenrisTuiApp(App): summary, _, context = headline.partition("\n") confidence_title, _, facts = confidence.partition("\n") self._render_headline(summary + "\n" + confidence_title + "\n\n" - + context + "\n" + facts + "\n" + scenario) + + context + "\n" + facts + "\n" + scenario + + "\n\n" + self._format_ladder(proj)) except Exception: self._render_headline("[bold]No projection available[/bold]") @@ -2267,8 +2267,14 @@ class FenrisTuiApp(App): def _format_headline(self, proj: ProjectionResult) -> str: """Format the lifespan headline (spec §6.11).""" if proj.headline_remaining_seconds is None: + if proj.stage == ProjectionStage.NO_OBSERVATIONS: + return "[bold]Usage-adjusted theoretical lifespan: [yellow]no projection yet — collecting observations[/yellow][/bold]" if proj.zero_rate_fact: return "[bold]Usage-adjusted theoretical lifespan: [red]no finite projection from this history[/red][/bold]" + rate = write_rate_gb_day(proj) + if rate is not None: + return ("[bold]Write rate: %.2f GB/day[/bold]\nNo lifespan without an endurance baseline" + % rate) if proj.warming_fact: return "[bold]Usage-adjusted theoretical lifespan: [yellow]%s[/yellow][/bold]" % proj.warming_fact return "[bold]Usage-adjusted theoretical lifespan: [red]no projection available[/red][/bold]" @@ -2295,13 +2301,20 @@ class FenrisTuiApp(App): if proj.qualifying_days_progress: facts = proj.qualifying_days_progress + " · " + facts - return "[bold]Projection confidence: [%s]%s[/%s][/bold]\n %s" % ( + return "[bold]Projection confidence: [%s]%s[/%s] · %s[/bold]\n %s" % ( color, - proj.confidence_state.value, + proj.stage_label, color, + ladder_count_text(proj), facts[:1].upper() + facts[1:], ) + def _format_ladder(self, proj: ProjectionResult) -> str: + """Format the full evidence ladder for the outlook (ADR 0012 §5).""" + lines = ["[bold]Evidence ladder[/bold] · %s" % ladder_count_text(proj)] + lines.extend(" " + escape(line) for line in ladder_lines(proj)) + return "\n".join(lines) + def _format_scenario(self, proj: ProjectionResult) -> str: """Format scenario range with horizon reasons (spec §6.5).""" if not proj.scenario_range: diff --git a/tests/test_complete_observation_day_gate.py b/tests/test_complete_observation_day_gate.py index 4f075e3..db520e0 100644 --- a/tests/test_complete_observation_day_gate.py +++ b/tests/test_complete_observation_day_gate.py @@ -1,16 +1,18 @@ -"""Complete observation day gate tests (issue #94). +"""Complete observation day condition tests (issues #94, #106). -Verifies that the endurance projection is withheld until at least one -complete local calendar day has been observed within a monitoring period. +Originally a gate (issue #94); ADR 0012 turned it into a contributing +fact. The projection is no longer withheld until one complete local +calendar day has been observed within a monitoring period, but it says +so while that condition is unmet. Seams: - compute_projection() → gate check via local_days table - ProjectionResult.contributing_facts → "waiting for a full local observation day" Acceptance criteria: -- Gate-1: No complete local day → UNSUPPORTED with waiting fact -- Gate-2: One complete local day → Limited confidence (if other conditions met) -- Gate-3: Partial days don't satisfy the gate +- Gate-1: No complete local day → projection renders with waiting fact +- Gate-2: One complete local day → no waiting fact +- Gate-3: Partial days don't satisfy the condition - Gate-4: CLI and TUI share the same gate via compute_projection() """ import sqlite3 @@ -111,14 +113,14 @@ def _open_period(conn, start="2026-09-01T00:00:00+00:00"): # --------------------------------------------------------------------------- -# Gate-1: No complete local day → UNSUPPORTED with waiting fact +# Gate-1: No complete local day → projection renders with waiting fact # --------------------------------------------------------------------------- class TestGateNoCompleteDay: - """Projection is unavailable before any complete local observation day.""" + """Projection still renders before any complete local observation day.""" - def test_no_local_days_unsupported(self, store): - """With no local_days entries, projection is UNSUPPORTED.""" + def test_no_local_days_states_waiting_fact(self, store): + """With no local_days entries, the projection renders and says it is waiting.""" _insert_baseline(store) _insert_segment(store) _open_period(store) @@ -128,11 +130,11 @@ class TestGateNoCompleteDay: _insert_day(store, d, bw=1024*1024*100) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) result = compute_projection(store, _clock()) - assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert result.confidence_state != ConfidenceState.UNSUPPORTED assert any("full local observation day" in f for f in result.contributing_facts) - assert result.headline_remaining_seconds is None + assert result.headline_remaining_seconds is not None - def test_only_partial_local_days_unsupported(self, store): + def test_only_partial_local_days_state_waiting_fact(self, store): """Partial (incomplete) local days don't satisfy the gate.""" _insert_baseline(store) _insert_segment(store) @@ -146,7 +148,7 @@ class TestGateNoCompleteDay: d = (datetime(2026, 9, 25) + timedelta(days=i)).strftime("%Y-%m-%d") _insert_local_day(store, d, complete=False, coverage=0.3) result = compute_projection(store, _clock()) - assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert result.confidence_state != ConfidenceState.UNSUPPORTED assert any("full local observation day" in f for f in result.contributing_facts) def test_gate_before_warming_check(self, store): @@ -166,7 +168,7 @@ class TestGateNoCompleteDay: # The key assertion: gate message should NOT appear when gate IS met assert not any("full local observation day" in f for f in result.contributing_facts) - def test_complete_legacy_day_without_trusted_activity_does_not_open_gate( + def test_complete_legacy_day_without_trusted_activity_keeps_waiting_fact( self, store, ): """A complete flag cannot make unavailable local activity qualify.""" @@ -183,11 +185,11 @@ class TestGateNoCompleteDay: "WHERE local_date = '2026-09-29'" ) result = compute_projection(store, _clock()) - assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert result.confidence_state != ConfidenceState.UNSUPPORTED assert any("full local observation day" in fact for fact in result.contributing_facts) - def test_day_split_by_deliberate_pause_does_not_open_gate(self, store): + def test_day_split_by_deliberate_pause_keeps_waiting_fact(self, store): """A complete-looking summary cannot span separate monitoring periods.""" _insert_baseline(store) _insert_segment(store) @@ -216,7 +218,7 @@ class TestGateNoCompleteDay: result = compute_projection(store, _clock()) - assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert result.confidence_state != ConfidenceState.UNSUPPORTED assert any("full local observation day" in fact for fact in result.contributing_facts) @@ -290,7 +292,7 @@ class TestGateOneCompleteDay: class TestGatePartialFirstDay: """Starting monitoring at noon means the partial first day doesn't count.""" - def test_partial_first_day_not_enough(self, store): + def test_partial_first_day_keeps_waiting_fact(self, store): """A single incomplete local day (started at noon) doesn't open the gate.""" _insert_baseline(store) _insert_segment(store, opened_at="2026-09-29T12:00:00+00:00") @@ -303,7 +305,7 @@ class TestGatePartialFirstDay: _insert_local_day(store, "2026-09-29", complete=False, coverage=0.5) _insert_local_day(store, "2026-09-30", complete=False, coverage=0.5) result = compute_projection(store, _clock()) - assert result.confidence_state == ConfidenceState.UNSUPPORTED + assert result.confidence_state != ConfidenceState.UNSUPPORTED assert any("full local observation day" in f for f in result.contributing_facts) diff --git a/tests/test_projection.py b/tests/test_projection.py index e3cd585..54c5e15 100644 --- a/tests/test_projection.py +++ b/tests/test_projection.py @@ -398,16 +398,16 @@ class TestArithmetic: _insert_day(store, d, bw=1024*1024*100) _insert_sample(store, "2026-09-30T10:00:00+00:00", pu=5) result = compute_projection(store, _clock()) - if result.headline_remaining_seconds is not None: - E_rated = 1.0 * TBW_TO_BYTES - regime_bytes = 30 * 1024 * 1024 * 100 - # Actual wall-clock: Sep 1 00:00 -> Sep 30 12:00 = 29.5 days - 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 + E_rated = 1.0 * TBW_TO_BYTES + W_t = 512000000000 # lifetime bytes written of the newest published sample + regime_bytes = 30 * 1024 * 1024 * 100 + # Actual wall-clock: Sep 1 00:00 -> Sep 30 12:00 = 29.5 days + 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 + expected = max(E_rated - W_t, 0) / rate + assert result.headline_remaining_seconds is not None + assert abs(result.headline_remaining_seconds - expected) < 1.0 def test_implied_baseline_formula(self, store): W_t = 1024 * 1024 * 1000 diff --git a/tests/test_staged_projection.py b/tests/test_staged_projection.py new file mode 100644 index 0000000..d1e89d6 --- /dev/null +++ b/tests/test_staged_projection.py @@ -0,0 +1,221 @@ +"""Staged projection with an evidence ladder (issue #106, ADR 0012). + +Every gate is exercised through compute_projection() against SQLite +fixtures built by seed_evidence(): N hours or days of complete evidence +written as hour observations, then derived into day aggregates the same +way a collection run does. +""" +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from fenris.day_aggregate import derive_all_days, persist_day_aggregate +from fenris.monitoring_periods import ensure_period_open +from fenris.projection import ( + BaselineTier, ConfidenceState, ProjectionStage, compute_projection, +) +from fenris.store import init_store + +START = datetime(2026, 9, 1, 0, 0, 0, tzinfo=timezone.utc) +HOUR_BYTES = 10 * 1024 ** 3 +MODEL = "Samsung SSD 970 EVO Plus 1TB" + + +@pytest.fixture +def store(tmp_path): + conn = init_store(tmp_path / "test.db") + yield conn + conn.close() + + +def seed_evidence(conn, hours, baseline="verified", start=START, degraded=False): + """Seed *hours* hours of complete evidence and return the clock. + + Monitoring opens at *start*; each hour is fully sampled and active. + The returned clock is the end of the last observed hour. + """ + conn.execute( + "INSERT INTO controller_segments " + "(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) " + "VALUES (?, ?, ?, 'nqn.test', 'SN1', ?, 'FW1', '0x144d', '0x144d', 'pcie')", + (start.isoformat(), "" if degraded else "nqn.test", degraded, MODEL), + ) + if baseline: + conn.execute( + "INSERT INTO endurance_baseline " + "(tbw_terabytes, source_url, document_revision, entry_date, model_string, " + " nominal_capacity_bytes, validated_by, verified, created_at, updated_at) " + "VALUES (600, ?, 'v1', '2026-01-01', ?, 1024000000000, ?, ?, " + " '2026-01-01T00:00:00+00:00', '2026-01-01T00:00:00+00:00')", + ("https://example.com/spec" if baseline == "verified" else None, MODEL, + "machine_match" if baseline == "verified" else None, + baseline == "verified"), + ) + ensure_period_open(conn, start) + for h in range(hours): + hour = start + timedelta(hours=h) + conn.execute( + "INSERT INTO hour_observations " + "(hour, active_seconds, idle_seconds, powered_off_seconds, unknown_seconds, " + " bytes_written_delta, bytes_read_delta, sample_count, coverage) " + "VALUES (?, 3600, 0, 0, 0, ?, 0, 1, 1.0)", + (hour.strftime("%Y-%m-%dT%H:00:00Z"), HOUR_BYTES), + ) + written = HOUR_BYTES * (h + 1) + conn.execute( + "INSERT INTO samples (ts, device, data_units_written, data_units_read, " + "percentage_used, bytes_written, bytes_read, power_on_hours, segment_id) " + "VALUES (?, '/dev/nvme0n1', ?, 0, 1, ?, 0, 100, 1)", + ((hour + timedelta(hours=1)).isoformat(), written // 512000, written), + ) + for aggregate in derive_all_days(conn): + persist_day_aggregate(conn, aggregate) + conn.commit() + return start + timedelta(hours=hours) + + +class TestHourStages: + def test_empty_store_has_no_observations(self, store): + result = compute_projection(store, START) + assert result.stage == ProjectionStage.NO_OBSERVATIONS + assert result.headline_remaining_seconds is None + + def test_two_hours_is_too_few_for_a_number(self, store): + clock = seed_evidence(store, 2) + result = compute_projection(store, clock) + assert result.stage == ProjectionStage.NO_OBSERVATIONS + assert result.headline_remaining_seconds is None + assert result.hours_observed == 2 + + def test_three_hours_is_provisional_with_a_number(self, store): + clock = seed_evidence(store, 3) + result = compute_projection(store, clock) + assert result.stage == ProjectionStage.PROVISIONAL + assert result.headline_remaining_seconds is not None + assert result.headline_remaining_seconds > 0 + assert result.hours_observed == 3 + assert "3 hours observed" in result.contributing_facts + assert "daily cycle not yet seen" in result.contributing_facts + + def test_three_hours_shows_short_horizon_spread(self, store): + clock = seed_evidence(store, 3) + result = compute_projection(store, clock) + assert result.scenario_range is not None + assert set(result.scenario_range.rates) <= {1, 3} + assert 1 in result.scenario_range.rates + + def test_provisional_rate_is_the_observed_write_rate(self, store): + clock = seed_evidence(store, 3) + result = compute_projection(store, clock) + assert result.write_rate == pytest.approx(HOUR_BYTES / 3600, rel=0.01) + + def test_complete_local_day_is_a_fact_not_a_gate(self, store): + clock = seed_evidence(store, 3) + result = compute_projection(store, clock) + assert result.headline_remaining_seconds is not None + assert "waiting for a full local observation day" in result.contributing_facts + + def test_provisional_until_24_hours_then_warming(self, store): + clock = seed_evidence(store, 23) + assert compute_projection(store, clock).stage == ProjectionStage.PROVISIONAL + + def test_twenty_four_hours_is_warming(self, store): + clock = seed_evidence(store, 24) + result = compute_projection(store, clock) + assert result.stage == ProjectionStage.WARMING + assert "daily cycle not yet seen" not in result.contributing_facts + + +class TestDayStages: + def test_warming_until_fourteen_qualifying_days(self, store): + clock = seed_evidence(store, 24 * 13) + assert compute_projection(store, clock).stage == ProjectionStage.WARMING + + def test_fourteen_days_with_verified_baseline_is_supported(self, store): + clock = seed_evidence(store, 24 * 14) + result = compute_projection(store, clock) + assert result.stage == ProjectionStage.SUPPORTED + assert result.confidence_state == ConfidenceState.SUPPORTED + + def test_unverified_baseline_is_limited_after_warm_up(self, store): + clock = seed_evidence(store, 24 * 14, baseline="unverified") + result = compute_projection(store, clock) + assert result.stage == ProjectionStage.LIMITED + assert result.baseline_tier == BaselineTier.UNVERIFIED + + def test_degraded_identity_is_limited(self, store): + clock = seed_evidence(store, 24 * 14, degraded=True) + assert compute_projection(store, clock).stage == ProjectionStage.LIMITED + + def test_stale_history_is_limited(self, store): + clock = seed_evidence(store, 24 * 14) + timedelta(days=4) + assert compute_projection(store, clock).stage == ProjectionStage.LIMITED + + +class TestNoBaseline: + def test_write_rate_replaces_lifespan(self, store): + clock = seed_evidence(store, 3, baseline=None) + result = compute_projection(store, clock) + assert result.stage == ProjectionStage.UNAVAILABLE + assert result.headline_remaining_seconds is None + assert result.write_rate == pytest.approx(HOUR_BYTES / 3600, rel=0.01) + + def test_no_baseline_is_not_synthesised(self, store): + clock = seed_evidence(store, 24 * 14, baseline=None) + result = compute_projection(store, clock) + assert result.baseline_tier == BaselineTier.NONE + assert result.headline_remaining_seconds is None + assert result.write_rate is not None + + +class TestShortHorizons: + def test_short_horizons_appear_once_covered(self, store): + clock = seed_evidence(store, 24 * 2) + rates = compute_projection(store, clock).scenario_range.rates + assert 1 in rates + assert 3 not in rates + + def test_three_day_horizon_covered_at_four_days(self, store): + clock = seed_evidence(store, 24 * 4) + rates = compute_projection(store, clock).scenario_range.rates + assert {1, 3} <= set(rates) + + def test_short_horizons_drop_when_seven_day_covered(self, store): + clock = seed_evidence(store, 24 * 9) + rates = compute_projection(store, clock).scenario_range.rates + assert 7 in rates + assert 1 not in rates and 3 not in rates + + +class TestEvidenceLadder: + def test_ladder_lists_each_supported_condition_with_reason(self, store): + clock = seed_evidence(store, 3) + ladder = compute_projection(store, clock).evidence_ladder + assert len(ladder) == 8 + assert all(c.reason for c in ladder) + assert all(isinstance(c.met, bool) for c in ladder) + + def test_ladder_all_met_when_supported(self, store): + clock = seed_evidence(store, 24 * 14) + result = compute_projection(store, clock) + assert result.ladder_met == result.ladder_total == 8 + + def test_ladder_count_rises_as_evidence_accrues(self, store, tmp_path): + counts = [] + for hours in (3, 24 * 2, 24 * 9, 24 * 14): + conn = init_store(tmp_path / ("ladder-%d.db" % hours)) + clock = seed_evidence(conn, hours) + counts.append(compute_projection(conn, clock).ladder_met) + conn.close() + assert counts == sorted(counts) + assert counts[0] < counts[-1] + assert len(set(counts)) >= 3 + + def test_unmet_condition_names_its_reason(self, store): + clock = seed_evidence(store, 3) + unmet = [c for c in compute_projection(store, clock).evidence_ladder if not c.met] + assert any("qualifying days" in c.reason for c in unmet) diff --git a/tests/test_staged_projection_callers.py b/tests/test_staged_projection_callers.py new file mode 100644 index 0000000..711f1c8 --- /dev/null +++ b/tests/test_staged_projection_callers.py @@ -0,0 +1,112 @@ +"""CLI and TUI render the projection stage, never pre-gate it (issue #106). + +Seams: get_status() and the TUI headline band. Both must show what +compute_projection() reports: the stage, the ladder count, and the full +ladder in the outlook; no caller decides what the projection can show. +""" +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) +sys.path.insert(0, str(Path(__file__).parent)) + +from fenris.status import get_status +from fenris.store import init_store +from fenris.tui import FenrisTuiApp +from test_staged_projection import seed_evidence + +SERVICES = { + "boot_enabled": False, "timer_active": False, + "last_collect_ok": None, "last_collect_age_s": None, + "last_collect_reason": None, +} + + +def _status(db, clock): + with patch("fenris.status.query_service_state", return_value=SERVICES): + return get_status(store_path=db, clock_now=clock, + query_services=False, query_journal=False) + + +@pytest.fixture +def db(tmp_path): + return tmp_path / "observations.db" + + +class TestCliStatus: + def test_provisional_stage_ladder_count_and_full_ladder(self, db): + conn = init_store(db) + clock = seed_evidence(conn, 3) + conn.close() + out = _status(db, clock) + assert "Provisional" in out + assert "3 hours observed" in out + assert "daily cycle not yet seen" in out + assert "remaining" in out + assert "conditions met" in out + assert "Evidence ladder" in out + assert "[ ] 14 qualifying days" in out + assert "[x] verified baseline" in out + + def test_single_sample_is_not_pre_gated(self, db): + conn = init_store(db) + seed_evidence(conn, 1) + conn.close() + out = _status(db, datetime(2026, 9, 1, 1, 0, tzinfo=timezone.utc)) + assert "requires at least two samples" not in out + assert "1 of 3 hours observed" in out + + def test_no_baseline_shows_write_rate_not_lifespan(self, db): + conn = init_store(db) + clock = seed_evidence(conn, 3, baseline=None) + conn.close() + out = _status(db, clock) + assert "write rate" in out.lower() + assert "GB/day" in out + + def test_supported_stage(self, db): + conn = init_store(db) + clock = seed_evidence(conn, 24 * 14) + conn.close() + out = _status(db, clock) + assert "Supported" in out + assert "8 of 8 conditions met" in out + + +class TestTuiHeadline: + @pytest.mark.asyncio + async def test_provisional_headline_and_ladder(self, db): + now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0) + conn = init_store(db) + seed_evidence(conn, 3, start=now - timedelta(hours=3)) + conn.close() + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = FenrisTuiApp(store_path=db) + async with app.run_test(size=(120, 50)): + text = str(app.query_one("#headline-band").render()) + assert "Provisional" in text + assert "3 hours observed" in text + assert "conditions met" in text + assert "Evidence ladder" in text + + @pytest.mark.asyncio + async def test_single_sample_is_not_pre_gated(self, db): + now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0) + conn = init_store(db) + seed_evidence(conn, 1, start=now - timedelta(hours=1)) + conn.close() + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = FenrisTuiApp(store_path=db) + async with app.run_test(size=(120, 50)): + text = str(app.query_one("#headline-band").render()) + assert "at least two samples" not in text + + +def test_callers_carry_no_projection_pre_gate(): + root = Path(__file__).parent.parent / "src" / "fenris" + for name in ("tui.py", "status.py"): + assert "at least two samples" not in (root / name).read_text()