From ef61ececc18ad0ba9115b931a37319e73b5684a2 Mon Sep 17 00:00:00 2001 From: xavierk Date: Mon, 5 Oct 2026 20:01:32 +0530 Subject: [PATCH] feat(tui): enter rated TBW from the dashboard Add a b key and a no-baseline banner that open a form for rated TBW and an optional source URL. The value is persisted as an unverified override through fenris-monitor baseline set under the existing polkit action, and the outcome is reported synchronously. Verified overrides stay CLI-only (ADR 0012 section 7). --- CHANGELOG.md | 1 + src/fenris/tui.py | 119 ++++++++++++++++- tests/test_tui_baseline_entry.py | 214 +++++++++++++++++++++++++++++++ 3 files changed, 331 insertions(+), 3 deletions(-) create mode 100644 tests/test_tui_baseline_entry.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e9503b3..a304035 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ backfill releases from before this changelog. - 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. +- Enter a rated TBW from the dashboard with `b`, saved as an unverified override, so the lifespan appears without the CLI. ### Changed diff --git a/src/fenris/tui.py b/src/fenris/tui.py index 45fa7b9..fb79515 100644 --- a/src/fenris/tui.py +++ b/src/fenris/tui.py @@ -11,6 +11,8 @@ Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8. """ from __future__ import annotations +import json +import math import sqlite3 import sys from datetime import date, datetime, timedelta, timezone @@ -37,6 +39,7 @@ from .activity_selection import ( from .activity_plot import VolumePoint, volume_plot from .projection import ( + BaselineTier, ConfidenceState, ProjectionResult, ProjectionStage, @@ -75,7 +78,7 @@ from .themes import THEMES, get_graph_colors _RESUME_HINT = "r Resume — enable monitoring and future boots" _ACTION_LEGEND = ( "p Pause · " + _RESUME_HINT + "\n" - "c Collect now · t Today · [ ] Day · g Date · m Motion · d Disclosures · ? Help" + "b Rated TBW · c Collect now · t Today · [ ] Day · g Date · m Motion · d Disclosures · ? Help" ) _LOCAL_ACTIVITY_STATE_LABELS = { "so_far": "totals so far", @@ -1678,6 +1681,74 @@ class HelpScreen(ModalScreen[None]): self.dismiss() +class BaselineScreen(ModalScreen["tuple[float, str | None] | None"]): + """Rated TBW entry; always stored as an unverified override (ADR 0012 §7).""" + + BINDINGS = [ + Binding("escape", "cancel", "Cancel", show=False), + ] + + CSS = """ + BaselineScreen { + align: center middle; + } + #baseline-content { + width: 64; + height: auto; + border: heavy $accent; + padding: 1 2; + } + #baseline-error { + width: 100%; + height: auto; + color: $error; + } + """ + + def compose(self) -> ComposeResult: + yield Container( + Static( + "[bold]Enter rated TBW[/bold]\n" + "Saved as an [bold]unverified[/bold] user-supplied override. " + "Verified overrides need full provenance via the CLI.", + id="baseline-header", + ), + Static("Rated TBW (terabytes written, required):"), + Input(placeholder="e.g. 600", id="tbw-input"), + Static("Source URL (optional):"), + Input(placeholder="https://…", id="source-input"), + Static("", id="baseline-error"), + Static("[dim]Enter Save · Esc Cancel[/dim]", id="baseline-footer"), + id="baseline-content", + ) + + def on_mount(self) -> None: + self.query_one("#tbw-input").focus() + + def on_input_submitted(self, event: Any) -> None: + error = self.query_one("#baseline-error") + raw = self.query_one("#tbw-input").value.strip() + if not raw: + error.update("[red]Rated TBW is required[/red]") + return + try: + tbw = float(raw) + except ValueError: + error.update("[red]Rated TBW must be a number of terabytes[/red]") + return + if not math.isfinite(tbw) or tbw <= 0: + error.update("[red]Rated TBW must be a positive number[/red]") + return + source = self.query_one("#source-input").value.strip() + if source and not source.startswith(("http://", "https://")): + error.update("[red]Source URL must start with http:// or https://[/red]") + return + self.dismiss((tbw, source or None)) + + def action_cancel(self) -> None: + self.dismiss(None) + + class DatePickerScreen(ModalScreen[str | None]): """Modal screen for entering a date to browse.""" @@ -1782,6 +1853,7 @@ class FenrisTuiApp(App): #usage-history, #live-activity { height: 1fr; min-height: 6; } #bar-render, #live-render { min-height: 4; } #local-day { height: auto; min-height: 2; color: $text; } + #baseline-banner { display: none; height: auto; background: $accent 15%; padding: 0 1; } #paused-banner { display: none; height: auto; background: $warning 15%; padding: 0 1; } #status-area { height: auto; max-height: 35%; scrollbar-size: 1 1; } #service-strip { height: auto; padding: 0 1; border-top: solid $primary 35%; } @@ -1806,6 +1878,7 @@ class FenrisTuiApp(App): BINDINGS = [ Binding("p", "pause", "Pause", show=False), Binding("r", "resume", "Resume", show=False), + Binding("b", "baseline", "Rated TBW", show=False), Binding("c", "collect", "Collect now", show=False), Binding("d", "disclose", "Disclosures", show=False), Binding("t", "today", "Today", show=False), @@ -1873,6 +1946,11 @@ class FenrisTuiApp(App): yield Static(_IDENTITY_FULL, id="identity") yield Static("Open with fenris (no sudo). Actions authenticate via polkit. ? Help", id="auth-notice") yield Static("", id="paused-banner") + yield Static( + "No endurance baseline — press [bold]b[/bold] to enter your drive's rated TBW " + "and unlock the lifespan.", + id="baseline-banner", + ) with VerticalScroll(id="dashboard-scroll"): with Container(id="main-grid"): with VerticalScroll(id="endurance-panel", classes="dashboard-pane"): @@ -2024,8 +2102,12 @@ class FenrisTuiApp(App): self.query_one("#service-strip").update("\n".join(lines)) self._render_paused_banner(comp) + def _show_baseline_banner(self, show: bool) -> None: + self.query_one("#baseline-banner").display = show + def _render_empty_or_fault(self, comp: StatusComposition) -> None: """Render empty history or the shared store-fault classification.""" + self._show_baseline_banner(not (comp.newer_schema or comp.store_fault)) if comp.newer_schema or comp.store_fault: message = comp.explanation[:1].upper() + comp.explanation[1:] self._render_headline("[bold red]%s[/bold red]" % message) @@ -2063,12 +2145,17 @@ class FenrisTuiApp(App): """Render all four regions from live store data.""" # --- Headline band (§7.2) --- if comp.sample_count == 0 and comp.day_count == 0: + self._show_baseline_banner( + conn.execute("SELECT COUNT(*) FROM endurance_baseline").fetchone()[0] == 0 + ) self._render_headline( "[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT ) else: + self._show_baseline_banner(False) try: proj = compute_projection(conn, self._clock_now) + self._show_baseline_banner(proj.baseline_tier == BaselineTier.NONE) headline = self._format_headline(proj) confidence = self._format_confidence(proj) scenario = self._format_scenario(proj) @@ -2429,6 +2516,27 @@ class FenrisTuiApp(App): """Collect now — synchronous outcome (spec §8.7, LC-8).""" self._run_helper("collect") + def action_baseline(self) -> None: + """Enter a rated TBW as an unverified override (ADR 0012 §7).""" + self.push_screen(BaselineScreen(), callback=self._baseline_entered) + + def _baseline_entered(self, entry: "tuple[float, str | None] | None") -> None: + if entry is None: + return + tbw, source_url = entry + payload = json.dumps({ + "tbw_terabytes": tbw, + "source_url": source_url, + "document_revision": None, + "entry_date": date.today().isoformat(), + "model_string": None, + "nominal_capacity_bytes": None, + "validated_by": "user", + "verified": False, + }) + if self._run_helper("baseline", ["set", payload]): + self.notify("Rated TBW saved as an unverified override.") + def action_disclose(self) -> None: """Show disclosures (spec §6.11, CI-4).""" self.push_screen(DisclosuresScreen()) @@ -2441,15 +2549,20 @@ class FenrisTuiApp(App): self, operation: str, extra_args: Optional[List[str]] = None, - ) -> None: - """Suspend the TUI while the shared control module owns the action.""" + ) -> bool: + """Suspend the TUI while the shared control module owns the action. + + Returns whether the helper succeeded; failures are reported here. + """ try: with self.suspend(): run_monitor(operation, *(extra_args or []), helper_path=self.helper_path) except MonitorError as exc: self.notify(str(exc), severity="error") + return False finally: self._refresh() + return True def run_tui( diff --git a/tests/test_tui_baseline_entry.py b/tests/test_tui_baseline_entry.py new file mode 100644 index 0000000..51aecfd --- /dev/null +++ b/tests/test_tui_baseline_entry.py @@ -0,0 +1,214 @@ +"""Enter rated TBW from the TUI (issue #107, ADR 0012 §7). + +Seam: the app's helper invocation (`_run_helper`), faked here the way the +other TUI tests fake it. The form builds an unverified override and hands it +to `fenris-monitor baseline set`; it never offers verified provenance. +""" +import json +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path +from types import SimpleNamespace +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.monitor import cmd_baseline_set +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, +} + + +@pytest.fixture +def db(tmp_path): + return tmp_path / "observations.db" + + +def _app(db): + return FenrisTuiApp(store_path=db) + + +def _seed_without_baseline(db): + now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0) + conn = init_store(db) + seed_evidence(conn, 3, baseline=None, start=now - timedelta(hours=3)) + conn.close() + + +def _text(app, selector): + return str(app.query_one(selector).render()) + + +async def _open_form(pilot): + await pilot.press("b") + await pilot.pause() + + +async def _fill(pilot, app, tbw, url=""): + app.screen.query_one("#tbw-input").value = tbw + app.screen.query_one("#source-input").value = url + await pilot.press("enter") + await pilot.pause() + + +class TestForm: + @pytest.mark.asyncio + async def test_b_opens_form_with_required_tbw_and_optional_source(self, db): + _seed_without_baseline(db) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + async with app.run_test(size=(120, 50)) as pilot: + await _open_form(pilot) + assert len(app.screen_stack) == 2 + assert app.screen.query_one("#tbw-input") + assert app.screen.query_one("#source-input") + text = str(app.screen.query_one("#baseline-header").render()) + assert "unverified" in text.lower() + + @pytest.mark.asyncio + @pytest.mark.parametrize("value", ["", "abc", "0", "-5", "nan", "inf"]) + async def test_invalid_tbw_is_rejected_without_calling_helper(self, db, value): + _seed_without_baseline(db) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + with patch.object(app, "_run_helper") as run_helper: + async with app.run_test(size=(120, 50)) as pilot: + await _open_form(pilot) + await _fill(pilot, app, value) + assert len(app.screen_stack) == 2 + assert str(app.screen.query_one("#baseline-error").render()).strip() + run_helper.assert_not_called() + + @pytest.mark.asyncio + async def test_invalid_source_url_is_rejected(self, db): + _seed_without_baseline(db) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + with patch.object(app, "_run_helper") as run_helper: + async with app.run_test(size=(120, 50)) as pilot: + await _open_form(pilot) + await _fill(pilot, app, "600", "not a url") + assert len(app.screen_stack) == 2 + assert "url" in str(app.screen.query_one("#baseline-error").render()).lower() + run_helper.assert_not_called() + + @pytest.mark.asyncio + async def test_escape_cancels_without_calling_helper(self, db): + _seed_without_baseline(db) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + with patch.object(app, "_run_helper") as run_helper: + async with app.run_test(size=(120, 50)) as pilot: + await _open_form(pilot) + await pilot.press("escape") + await pilot.pause() + assert len(app.screen_stack) == 1 + run_helper.assert_not_called() + + +class TestSubmit: + @pytest.mark.asyncio + async def test_submit_routes_unverified_override_to_baseline_set(self, db): + _seed_without_baseline(db) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + with patch.object(app, "_run_helper", return_value=True) as run_helper: + async with app.run_test(size=(120, 50)) as pilot: + await _open_form(pilot) + await _fill(pilot, app, "600", "https://example.com/spec") + assert len(app.screen_stack) == 1 + run_helper.assert_called_once() + operation, args = run_helper.call_args.args + assert operation == "baseline" + assert args[0] == "set" + payload = json.loads(args[1]) + assert payload["tbw_terabytes"] == 600 + assert payload["source_url"] == "https://example.com/spec" + assert payload["verified"] is False + assert payload["validated_by"] == "user" + # Incomplete provenance keeps the override unverified. + assert payload["document_revision"] is None + + @pytest.mark.asyncio + async def test_blank_source_is_stored_as_absent(self, db): + _seed_without_baseline(db) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + with patch.object(app, "_run_helper", return_value=True) as run_helper: + async with app.run_test(size=(120, 50)) as pilot: + await _open_form(pilot) + await _fill(pilot, app, "1.5") + payload = json.loads(run_helper.call_args.args[1][1]) + assert payload["tbw_terabytes"] == 1.5 + assert payload["source_url"] is None + + @pytest.mark.asyncio + async def test_headline_changes_from_write_rate_to_lifespan(self, db): + _seed_without_baseline(db) + + def fake_helper(operation, extra_args=None): + cmd_baseline_set(SimpleNamespace(store_path=db, baseline_json=extra_args[1])) + return True + + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + with patch.object(app, "_run_helper", side_effect=fake_helper): + async with app.run_test(size=(120, 50)) as pilot: + before = _text(app, "#headline-band") + assert "Write rate" in before + assert "remaining" not in before + await _open_form(pilot) + await _fill(pilot, app, "600") + app.on_refresh_tick() + await pilot.pause() + after = _text(app, "#headline-band") + assert "remaining" in after + assert "Write rate" not in after + + +class TestBanner: + @pytest.mark.asyncio + async def test_banner_shown_when_no_baseline_applies(self, db): + _seed_without_baseline(db) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + async with app.run_test(size=(120, 50)): + banner = app.query_one("#baseline-banner") + assert banner.display + assert "b" in str(banner.render()).lower() + assert "rated tbw" in str(banner.render()).lower() + + @pytest.mark.asyncio + async def test_banner_shown_on_first_run(self, tmp_path): + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(tmp_path / "missing.db") + async with app.run_test(size=(120, 50)): + assert app.query_one("#baseline-banner").display + + @pytest.mark.asyncio + async def test_banner_hidden_once_baseline_exists(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 = _app(db) + async with app.run_test(size=(120, 50)): + assert not app.query_one("#baseline-banner").display + + @pytest.mark.asyncio + async def test_banner_hidden_for_unreadable_store(self, db): + db.write_bytes(b"not a sqlite database" * 100) + with patch("fenris.status.query_service_state", return_value=SERVICES): + app = _app(db) + async with app.run_test(size=(120, 50)): + assert not app.query_one("#baseline-banner").display