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.
This commit is contained in:
xavierk
2026-10-05 19:57:46 +05:30
parent 96fd1702fa
commit e26a851115
8 changed files with 615 additions and 148 deletions
+199 -93
View File
@@ -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,
)
+19 -16
View File
@@ -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")
+22 -9
View File
@@ -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: