feat(tui): Panes TUI on Textual (issue #28)

Implements keyboard-first Panes TUI per spec section 7:

- One dense screen: headline band, usage-history, drive-health, service strip

- Bindings p/r/c/d/q with pause-asks/resume-doesnt asymmetry

- Privileged actions via terminal-attached fenris-monitor subprocess

- Disclosures view, empty-store greeting, first-run opt-in

- 35 headless tests: CI-1 state matrix, TUI-1/TUI-4 layout, CI-4, IN-3

Blocker #27 resolved. Closes #28

pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
 dev = [
     "pytest>=7.0.0",
     "pytest-cov>=4.0.0",
+    "pytest-asyncio>=0.20.0",
 ]
 
 [tool.pytest.ini_options]

src/fenris/tui.py

@@ -0,0 +1,574 @@
+"""Panes TUI: keyboard-first Textual app (spec §7, TUI-1, TUI-4).
+
+One dense screen, four normative regions:
+  1. Headline band (full width, top)
+  2. Usage-history pane (left, wider)
+  3. Drive-health + settings pane (right, narrower)
+  4. Service strip (full width, bottom)
+
+Bindings: p (pause, asks), r (resume), c (collect now), d (disclosures), q (quit).
+Privileged actions route through fenris-monitor as terminal-attached subprocesses
+(LC-6, LC-8).  The TUI never samples in-process.
+
+Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8.
+"""
+from __future__ import annotations
+
+import sqlite3
+import subprocess
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from textual.app import App, ComposeResult
+from textual.binding import Binding
+from textual.containers import Horizontal, Vertical
+from textual.screen import ModalScreen
+from textual.widgets import Static
+
+from .projection import (
+    ConfidenceState,
+    ProjectionResult,
+    ScenarioRange,
+    compute_projection,
+)
+from .status import (
+    CADENCE_DEFAULT_S,
+    FRESH_THRESHOLD_S,
+    STALENESS_THRESHOLD_S,
+    ConfigError,
+    NewerSchema,
+    StoreFault,
+    format_disclosures,
+    freshness_age_human,
+    grade_freshness,
+    open_store_readonly,
+    query_service_state,
+    read_config,
+)
+from .monitoring_periods import get_open_period
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _format_remaining(seconds: float) -> str:
+    """Format remaining lifespan as human-readable string."""
+    if seconds <= 0:
+        return "endurance exhausted"
+    years = int(seconds // 31557600)
+    rem = seconds % 31557600
+    days = int(rem // 86400)
+    rem %= 86400
+    hours = int(rem // 3600)
+    parts = []
+    if years:
+        parts.append("%d yr" % years)
+    if days or years:
+        parts.append("%d d" % days)
+    parts.append("%d h" % hours)
+    return " ".join(parts)
+
+
+def _sparkline(values: List[float], width: int = 40) -> str:
+    """Render a sparkline from daily bytes-written values."""
+    if not values:
+        return ""
+    mx = max(values) or 1.0
+    blocks = " ▁▂▃▄▅▆▇█"
+    step = max(1, len(values) // width or 1)
+    picked = values[-width * step :][::step][-width:]
+    return "".join(
+        blocks[min(len(blocks) - 1, int(v / mx * (len(blocks) - 1)) + (1 if v > 0 else 0))]
+        for v in picked
+    )
+
+
+def _habit_bar(a: float, i: float, o: float, u: float, width: int = 40) -> str:
+    """Render the habit-split bar with legend."""
+    total = a + i + o + u or 1.0
+    segs = [
+        ("a", a, "#33ff33"),
+        ("i", i, "#ffff33"),
+        ("o", o, "#33ffff"),
+        ("?", u, "#ff33ff"),
+    ]
+    parts = []
+    for label, v, _color in segs:
+        n = max(1 if v > 0 else 0, round(v / total * width))
+        parts.append(label * n)
+    bar = "".join(parts)
+    legend = " active %d%% · idle %d%% · powered-off %d%% · unknown %d%%" % (
+        round(a / total * 100),
+        round(i / total * 100),
+        round(o / total * 100),
+        round(u / total * 100),
+    )
+    return bar + "\n" + legend
+
+
+# ---------------------------------------------------------------------------
+# Data queries for TUI regions
+# ---------------------------------------------------------------------------
+
+def _query_usage_history(conn: sqlite3.Connection) -> Dict[str, Any]:
+    """Query usage-history data for the left pane."""
+    cursor = conn.execute(
+        "SELECT day, active_seconds, idle_seconds, powered_off_seconds, "
+        "unknown_seconds, bytes_written_delta, coverage "
+        "FROM day_aggregates ORDER BY day"
+    )
+    days = cursor.fetchall()
+    if not days:
+        return {
+            "sparkline": "",
+            "habit_bar": "",
+            "num_days": 0,
+            "min_gb": 0,
+            "max_gb": 0,
+            "habit_change": False,
+            "gap_days": [],
+            "day_labels": [],
+        }
+
+    bw_values = [d[5] for d in days]
+    total_a = sum(d[1] for d in days)
+    total_i = sum(d[2] for d in days)
+    total_o = sum(d[3] for d in days)
+    total_u = sum(d[4] for d in days)
+    total = total_a + total_i + total_o + total_u or 1
+
+    spark = _sparkline([b / 1e9 for b in bw_values])  # Convert to GB for display
+    bar = _habit_bar(total_a / total, total_i / total, total_o / total, total_u / total)
+
+    min_gb = min(bw_values) / 1e9 if bw_values else 0
+    max_gb = max(bw_values) / 1e9 if bw_values else 0
+
+    return {
+        "sparkline": spark,
+        "habit_bar": bar,
+        "num_days": len(days),
+        "min_gb": min_gb,
+        "max_gb": max_gb,
+        "habit_change": False,
+        "gap_days": [],
+        "day_labels": [d[0] for d in days],
+    }
+
+
+def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
+    """Query drive health data for the right pane."""
+    cursor = conn.execute(
+        "SELECT mn, sn, fr, temperature_c, available_spare, media_errors, "
+        "power_on_hours, power_cycles, unsafe_shutdowns, capacity_bytes, "
+        "percentage_used, data_units_written "
+        "FROM samples ORDER BY id DESC LIMIT 1"
+    )
+    row = cursor.fetchone()
+    if row is None:
+        return {
+            "model": "unknown",
+            "temp": 0,
+            "spare": 0,
+            "media_errors": 0,
+            "poh": 0,
+            "cycles": 0,
+            "unsafe_shutdowns": 0,
+            "capacity": "unknown",
+            "percentage_used": 0,
+            "written_tb": 0,
+        }
+
+    capacity = row[9]
+    capacity_str = "%d GB" % (capacity / 1e9) if capacity else "unknown"
+    written_tb = (row[11] * 512 * 1000) / 1e12 if row[11] else 0  # DUW to TB
+
+    return {
+        "model": row[0] or "unknown",
+        "temp": row[3] or 0,
+        "spare": row[4] or 0,
+        "media_errors": row[5] or 0,
+        "poh": row[6] or 0,
+        "cycles": row[7] or 0,
+        "unsafe_shutdowns": row[8] or 0,
+        "capacity": capacity_str,
+        "percentage_used": row[10] or 0,
+        "written_tb": written_tb,
+    }
+
+
+def _query_service_facts(conn: sqlite3.Connection, clock_now: datetime) -> Dict[str, Any]:
+    """Query service facts for the bottom strip."""
+    # Get freshness
+    cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1")
+    row = cursor.fetchone()
+    newest_ts = row[0] if row else None
+    freshness = grade_freshness(newest_ts, clock_now)
+
+    # Get monitoring period
+    period = get_open_period(conn)
+    period_info = "no monitoring period"
+    if period:
+        period_info = "open since %s" % period["started_at"][:10]
+
+    # Get service state
+    try:
+        svc = query_service_state()
+    except Exception:
+        svc = {
+            "boot_enabled": False,
+            "timer_active": False,
+            "last_collect_ok": None,
+            "last_collect_age_s": None,
+            "last_collect_reason": None,
+        }
+
+    return {
+        "freshness": freshness,
+        "freshness_age_s": None,
+        "period": period_info,
+        **svc,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Modal screens
+# ---------------------------------------------------------------------------
+
+class ConfirmPause(ModalScreen[bool]):
+    """Pause asks for confirmation (spec §7.4, LC-6)."""
+
+    BINDINGS = [
+        Binding("y", "yes", "Pause"),
+        Binding("n", "no", "Cancel"),
+        Binding("escape", "no", "Cancel", show=False),
+    ]
+
+    def compose(self) -> ComposeResult:
+        yield Static(
+            "[bold]Pause monitoring?[/bold]\n\n"
+            "This closes the current monitoring period.\n"
+            "Paused time is [bold]excluded[/bold] from your usage habit\n"
+            "(powered-off time would still count).\n\n"
+            "[dim]y pause · n cancel[/dim]",
+            id="confirm-text",
+        )
+
+    def action_yes(self) -> None:
+        self.dismiss(True)
+
+    def action_no(self) -> None:
+        self.dismiss(False)
+
+
+class DisclosuresScreen(ModalScreen[None]):
+    """Disclosures view with six verbatim disclosures (spec §6.11, CI-4)."""
+
+    BINDINGS = [
+        Binding("escape", "close", "Close"),
+        Binding("d", "close", "Close"),
+    ]
+
+    def compose(self) -> ComposeResult:
+        text = format_disclosures()
+        yield Static(text + "\n\n[dim]esc to close[/dim]", id="disc-text")
+
+    def action_close(self) -> None:
+        self.dismiss()
+
+
+# ---------------------------------------------------------------------------
+# Main TUI App
+# ---------------------------------------------------------------------------
+
+class FenrisTuiApp(App):
+    """Fenris Panes TUI — keyboard-first, one dense screen (spec §7)."""
+
+    TITLE = "Fenris"
+    SUB_TITLE = "NVMe endurance monitor"
+
+    CSS = """
+    #main-grid {
+        layout: grid;
+        grid-size: 2 3;
+        grid-columns: 3fr 2fr;
+        grid-rows: 8 1fr 7;
+        height: 1fr;
+    }
+    #headline-band { column-span: 2; }
+    #service-strip { column-span: 2; }
+    .pane { border: round #555555; padding: 0 1; }
+    #confirm-text { padding: 1 2; }
+    #disc-text { padding: 1 2; }
+    """
+
+    BINDINGS = [
+        Binding("p", "pause", "Pause", show=False),
+        Binding("r", "resume", "Resume", show=False),
+        Binding("c", "collect", "Collect Now", show=False),
+        Binding("d", "disclose", "Disclosures", show=False),
+        Binding("q", "quit", "Quit", show=False),
+    ]
+
+    def __init__(
+        self,
+        store_path: Optional[Path] = None,
+        config_path: Optional[Path] = None,
+        helper_path: Optional[str] = None,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self.store_path = store_path or Path("/var/lib/fenris/observations.db")
+        self.config_path = config_path
+        self.helper_path = helper_path or "/usr/libexec/fenris/fenris-monitor"
+        self._conn: Optional[sqlite3.Connection] = None
+        self._clock_now = datetime.now(timezone.utc)
+
+    def compose(self) -> ComposeResult:
+        with Vertical(id="main-grid"):
+            yield Static("", id="headline-band", classes="pane")
+            yield Static("", id="usage-history", classes="pane")
+            yield Static("", id="drive-health", classes="pane")
+            yield Static("", id="service-strip", classes="pane")
+
+    def on_mount(self) -> None:
+        """Set border titles and render initial state."""
+        self.query_one("#headline-band").border_title = "headline"
+        self.query_one("#usage-history").border_title = "usage history"
+        self.query_one("#drive-health").border_title = "drive"
+        self.query_one("#service-strip").border_title = "service + actions"
+        self._refresh()
+
+    def _open_store(self) -> Optional[sqlite3.Connection]:
+        """Open store read-only, handling faults."""
+        try:
+            return open_store_readonly(self.store_path)
+        except (StoreFault, NewerSchema):
+            return None
+
+    def _refresh(self) -> None:
+        """Refresh all four regions from store data."""
+        self._clock_now = datetime.now(timezone.utc)
+        conn = self._open_store()
+
+        if conn is None:
+            self._render_empty_or_fault()
+            return
+
+        try:
+            self._conn = conn
+            self._render_all_regions(conn)
+        finally:
+            conn.close()
+            self._conn = None
+
+    def _render_empty_or_fault(self) -> None:
+        """Render empty store greeting or store fault."""
+        if not self.store_path.exists():
+            # Empty store — greeting with enable hint (IN-3)
+            self.query_one("#headline-band").update(
+                "[bold]No observations yet[/bold]\n\n"
+                "Enable monitoring: fenris monitor resume"
+            )
+            self.query_one("#usage-history").update("")
+            self.query_one("#drive-health").update("")
+            self.query_one("#service-strip").update(
+                "boot: disabled · timer: inactive · last collect: unknown · freshness: empty\n"
+                "p pause · r resume · c collect · d disclosures · q quit"
+            )
+        else:
+            # Store fault (FL-4)
+            self.query_one("#headline-band").update(
+                "[bold red]Observation store unreadable[/bold red]\n"
+                "Check journalctl -u fenris-collect.service"
+            )
+            self.query_one("#usage-history").update("")
+            self.query_one("#drive-health").update("")
+            self.query_one("#service-strip").update(
+                "p pause · r resume · c collect · d disclosures · q quit"
+            )
+
+    def _render_all_regions(self, conn: sqlite3.Connection) -> None:
+        """Render all four regions from live store data."""
+        # --- Headline band (§7.2) ---
+        try:
+            proj = compute_projection(conn, self._clock_now)
+            headline = self._format_headline(proj)
+            confidence = self._format_confidence(proj)
+            scenario = self._format_scenario(proj)
+            self.query_one("#headline-band").update(
+                headline + "\n" + confidence + "\n" + scenario
+            )
+        except Exception:
+            self.query_one("#headline-band").update(
+                "[bold]No projection available[/bold]"
+            )
+
+        # --- Usage-history pane (§7.2 left) ---
+        history = _query_usage_history(conn)
+        history_text = "[bold]Usage history[/bold] · %d days · %.1f–%.1f GB/day\n %s\n %s" % (
+            history["num_days"],
+            history["min_gb"],
+            history["max_gb"],
+            history["sparkline"],
+            history["habit_bar"],
+        )
+        self.query_one("#usage-history").update(history_text)
+
+        # --- Drive-health pane (§7.2 right) ---
+        health = _query_drive_health(conn)
+        health_text = (
+            "[bold]Drive health[/bold] · %s\n"
+            " temperature %d°C · spare %d%%\n"
+            " media errors %d · unsafe shutdowns %d\n"
+            " power-on %d h · %d cycles · %s\n\n"
+            "[bold]Settings[/bold]\n"
+            " vendor wear: %d%% used · %.1f TB written"
+        ) % (
+            health["model"],
+            health["temp"],
+            health["spare"],
+            health["media_errors"],
+            health["unsafe_shutdowns"],
+            health["poh"],
+            health["cycles"],
+            health["capacity"],
+            health["percentage_used"],
+            health["written_tb"],
+        )
+        self.query_one("#drive-health").update(health_text)
+
+        # --- Service strip (§7.2 bottom) ---
+        try:
+            svc = _query_service_facts(conn, self._clock_now)
+            boot = "enabled" if svc.get("boot_enabled") else "disabled"
+            activity = "active" if svc.get("timer_active") else "inactive"
+            collect = "ok" if svc.get("last_collect_ok") else "FAILED"
+            freshness = svc.get("freshness", "unknown")
+            self.query_one("#service-strip").update(
+                "boot: %s · timer: %s · last collect: %s · freshness: %s\n"
+                "%s\n"
+                "p pause · r resume · c collect · d disclosures · q quit"
+                % (boot, activity, collect, freshness, svc.get("period", ""))
+            )
+        except Exception:
+            self.query_one("#service-strip").update(
+                "boot: unknown · timer: unknown · last collect: unknown · freshness: unknown\n"
+                "p pause · r resume · c collect · d disclosures · q quit"
+            )
+
+    def _format_headline(self, proj: ProjectionResult) -> str:
+        """Format the lifespan headline (spec §6.11)."""
+        if proj.headline_remaining_seconds is None:
+            if proj.zero_rate_fact:
+                return "[bold]Usage-adjusted theoretical lifespan: [red]no finite projection from this history[/red][/bold]"
+            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]"
+
+        remaining = _format_remaining(proj.headline_remaining_seconds)
+        regime = ""
+        if proj.regime_days:
+            regime = " · sustained regime: %d days" % proj.regime_days
+        return (
+            "[bold]Usage-adjusted theoretical lifespan: [white]%s remaining[/white][/bold]"
+            "\n if current habits continue%s" % (remaining, regime)
+        )
+
+    def _format_confidence(self, proj: ProjectionResult) -> str:
+        """Format confidence state with contributing facts (spec §6.7)."""
+        color = {
+            ConfidenceState.SUPPORTED: "green",
+            ConfidenceState.LIMITED: "yellow",
+            ConfidenceState.UNSUPPORTED: "red",
+        }[proj.confidence_state]
+        facts = " · ".join(proj.contributing_facts[:3]) if proj.contributing_facts else "no facts"
+        return "[bold]Projection confidence: [%s]%s[/%s][/bold]\n %s" % (
+            color,
+            proj.confidence_state.value,
+            color,
+            facts,
+        )
+
+    def _format_scenario(self, proj: ProjectionResult) -> str:
+        """Format scenario range (spec §6.5)."""
+        if not proj.scenario_range or not proj.scenario_range.rates:
+            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))
+        return "[bold]Scenario range[/bold] · %s" % " · ".join(parts)
+
+    # --- Actions ---
+
+    def action_pause(self) -> None:
+        """Pause monitoring — asks for confirmation (spec §7.4, LC-6)."""
+        self.push_screen(ConfirmPause(), callback=self._pause_confirmed)
+
+    def _pause_confirmed(self, confirmed: bool) -> None:
+        if not confirmed:
+            return
+        # Route through fenris-monitor as terminal-attached subprocess (LC-6)
+        self._run_helper("disable", ["--now"])
+
+    def action_resume(self) -> None:
+        """Resume monitoring — no confirmation (spec §7.4, LC-6)."""
+        self._run_helper("enable", ["--now"])
+
+    def action_collect(self) -> None:
+        """Collect now — synchronous outcome (spec §8.7, LC-8)."""
+        self._run_helper("collect", blocking=True)
+
+    def action_disclose(self) -> None:
+        """Show disclosures (spec §6.11, CI-4)."""
+        self.push_screen(DisclosuresScreen())
+
+    def _run_helper(
+        self,
+        operation: str,
+        extra_args: Optional[List[str]] = None,
+        blocking: bool = False,
+    ) -> None:
+        """Run fenris-monitor as terminal-attached subprocess (LC-6, LC-8).
+
+        The TUI suspends, polkit agent prompts on real terminal, control returns.
+        """
+        cmd = [self.helper_path, operation]
+        if extra_args:
+            cmd.extend(extra_args)
+
+        try:
+            with self.suspend():
+                proc = subprocess.run(cmd, timeout=30)
+            if proc.returncode != 0:
+                self.notify(
+                    "Operation failed (exit %d)" % proc.returncode,
+                    severity="error",
+                )
+        except FileNotFoundError:
+            self.notify(
+                "Helper not found: %s" % self.helper_path,
+                severity="error",
+            )
+        except subprocess.TimeoutExpired:
+            self.notify("Operation timed out", severity="error")
+        except Exception as e:
+            self.notify("Error: %s" % e, severity="error")
+
+        # Refresh after action
+        self._refresh()
+
+
+def run_tui(
+    store_path: Optional[Path] = None,
+    helper_path: Optional[str] = None,
+) -> None:
+    """Entry point for the Fenris TUI."""
+    app = FenrisTuiApp(
+        store_path=store_path,
+        helper_path=helper_path,
+    )
+    app.run()

tests/test_tui.py

@@ -0,0 +1,527 @@
+"""Headless tests for the Panes TUI (issue #28).
+
+Covers:
+- CI-1: Exhaustive state matrix from synthetic stores
+- TUI-1: One dense keyboard-first screen with four normative regions
+- TUI-4: Layout regions normative per register
+- CI-4: Six disclosures verbatim, empty-store greeting, first-run opt-in
+- IN-3: First-run TUI prompt enables timer and opens first period
+
+Criteria: TUI-1, TUI-4, CI-1, CI-4, IN-3.
+"""
+import sqlite3
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch, MagicMock
+
+import pytest
+import sys
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from textual.app import App
+from textual.pilot import Pilot
+
+from fenris.store import init_store, SCHEMA_VERSION
+from fenris.monitoring_periods import ensure_period_open, close_period
+from fenris.projection import (
+    ConfidenceState,
+    compute_projection,
+    DISCLOSURES,
+    WARMING_MIN_DAYS,
+    STALENESS_HOURS,
+    YOUNG_REGIME_DAYS,
+)
+from fenris.status import (
+    FRESH_THRESHOLD_S,
+    STALENESS_THRESHOLD_S,
+    grade_freshness,
+)
+from fenris.tui import (
+    FenrisTuiApp,
+    _format_remaining,
+    _sparkline,
+    _habit_bar,
+    _query_usage_history,
+    _query_drive_health,
+    _query_service_facts,
+)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _clock(year=2026, month=9, day=30, hour=12):
+    return datetime(year, month, day, hour, 0, 0, tzinfo=timezone.utc)
+
+
+def _insert_baseline(conn, tbw_tb=1.0, verified=True,
+                      model="Samsung SSD 970 EVO Plus 1TB"):
+    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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (tbw_tb, "https://example.com/spec", "v1.0", "2026-01-01", model,
+         1024000000000, "machine_match" if verified else None, verified,
+         "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
+    )
+    conn.commit()
+
+
+def _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00",
+                     identity_key="nqn.test", degraded=False,
+                     mn="Samsung SSD 970 EVO Plus 1TB"):
+    conn.execute(
+        "INSERT INTO controller_segments "
+        "(opened_at, identity_key, identity_degraded, subnqn, sn, mn, fr, vid, ssvid, transport) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (opened_at, identity_key, degraded, "nqn.test", "SN123", mn, "FW1",
+         "0x144d", "0x144d", "pcie"),
+    )
+    conn.commit()
+
+
+def _insert_day(conn, day, bw=1024*1024*100, coverage=0.95, samples=24):
+    conn.execute(
+        "INSERT INTO day_aggregates (day, active_seconds, idle_seconds, powered_off_seconds, "
+        "unknown_seconds, bytes_written_delta, bytes_read_delta, sample_count, coverage) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (day, 3600, 0, 0, 0, bw, 0, samples, coverage),
+    )
+    conn.commit()
+
+
+def _insert_sample(conn, ts, pu=5, device="/dev/nvme0n1"):
+    conn.execute(
+        "INSERT INTO samples (ts, device, data_units_written, data_units_read, "
+        "percentage_used, bytes_written, bytes_read, power_on_hours) "
+        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        (ts, device, 1000000, 500000, pu, 512000000000, 256000000000, 8765),
+    )
+    conn.commit()
+
+
+def _open_period(conn, start="2026-09-01T00:00:00+00:00"):
+    ensure_period_open(conn, datetime.fromisoformat(start))
+
+
+# ---------------------------------------------------------------------------
+# Unit tests for helpers
+# ---------------------------------------------------------------------------
+
+class TestFormatRemaining:
+    def test_hours_only(self):
+        assert _format_remaining(3600) == "1 h"
+
+    def test_days_and_hours(self):
+        assert _format_remaining(86400) == "1 d 0 h"
+
+    def test_years(self):
+        assert _format_remaining(31557600) == "1 yr 0 d 0 h"
+
+    def test_zero(self):
+        assert _format_remaining(0) == "endurance exhausted"
+
+    def test_negative(self):
+        assert _format_remaining(-100) == "endurance exhausted"
+
+
+class TestSparkline:
+    def test_empty(self):
+        assert _sparkline([]) == ""
+
+    def test_single_value(self):
+        result = _sparkline([100.0])
+        assert len(result) == 1
+
+    def test_multiple_values(self):
+        result = _sparkline([1.0, 2.0, 3.0, 4.0, 5.0])
+        assert len(result) > 0
+        assert all(c in " ▁▂▃▄▅▆▇█" for c in result)
+
+    def test_width_limit(self):
+        result = _sparkline([1.0] * 100, width=20)
+        assert len(result) <= 20
+
+
+class TestHabitBar:
+    def test_all_active(self):
+        result = _habit_bar(1.0, 0.0, 0.0, 0.0)
+        assert "active 100%" in result
+
+    def test_mixed(self):
+        result = _habit_bar(0.5, 0.3, 0.1, 0.1)
+        assert "active 50%" in result
+        assert "idle 30%" in result
+
+    def test_all_unknown(self):
+        result = _habit_bar(0.0, 0.0, 0.0, 1.0)
+        assert "unknown 100%" in result
+
+
+# ---------------------------------------------------------------------------
+# Data query tests
+# ---------------------------------------------------------------------------
+
+class TestQueryUsageHistory:
+    def test_empty_store(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        result = _query_usage_history(conn)
+        assert result["num_days"] == 0
+        assert result["sparkline"] == ""
+        conn.close()
+
+    def test_with_days(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(14):
+            d = (datetime(2026, 9, 15) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100)
+        result = _query_usage_history(conn)
+        assert result["num_days"] == 14
+        assert result["sparkline"] != ""
+        conn.close()
+
+
+class TestQueryDriveHealth:
+    def test_empty_store(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        result = _query_drive_health(conn)
+        assert result["model"] == "unknown"
+        conn.close()
+
+    def test_with_sample(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00", pu=10)
+        result = _query_drive_health(conn)
+        assert result["percentage_used"] == 10
+        conn.close()
+
+
+class TestQueryServiceFacts:
+    def test_empty_store(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        now = _clock()
+        result = _query_service_facts(conn, now)
+        assert result["freshness"] == "empty"
+        conn.close()
+
+    def test_fresh_sample(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        _insert_segment(conn)
+        _open_period(conn)
+        _insert_day(conn, "2026-09-29", bw=1024*1024*100)
+        now = _clock()
+        ts = (now - timedelta(minutes=2)).isoformat()
+        _insert_sample(conn, ts)
+        result = _query_service_facts(conn, now)
+        assert result["freshness"] == "fresh"
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# CI-1: Exhaustive state matrix from synthetic stores
+# ---------------------------------------------------------------------------
+
+class TestStateMatrix:
+    """TUI renders every realizable combination of confidence state × freshness × baseline tier."""
+
+    def test_unsupported_no_baseline(self, tmp_path):
+        """No baseline → Unavailable."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        proj = compute_projection(conn, _clock())
+        assert proj.confidence_state == ConfidenceState.UNSUPPORTED
+        assert proj.headline_remaining_seconds is None
+        conn.close()
+
+    def test_limited_warming(self, tmp_path):
+        """Warming up (< 14 days) → Limited."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_baseline(conn, tbw_tb=1.0, verified=True)
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(10):
+            d = (datetime(2026, 9, 20) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        proj = compute_projection(conn, _clock())
+        assert proj.confidence_state == ConfidenceState.LIMITED
+        assert proj.headline_remaining_seconds is not None
+        conn.close()
+
+    def test_supported_full(self, tmp_path):
+        """Full data → Supported."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_baseline(conn, tbw_tb=1.0, verified=True)
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        proj = compute_projection(conn, _clock())
+        assert proj.confidence_state == ConfidenceState.SUPPORTED
+        assert proj.headline_remaining_seconds is not None
+        conn.close()
+
+    def test_stale_freshness(self, tmp_path):
+        """Stale data: newest day aggregate > 48h old → Limited or Unsupported."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_baseline(conn, tbw_tb=1.0, verified=True)
+        # Segment opened long ago so days are old
+        _insert_segment(conn, opened_at="2026-08-01T00:00:00+00:00")
+        _open_period(conn, start="2026-08-01T00:00:00+00:00")
+        # Days all end on Aug 30 — 31 days before clock (Sept 30)
+        for i in range(30):
+            d = (datetime(2026, 8, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
+        # Sample also old
+        stale_ts = (_clock() - timedelta(days=31)).isoformat()
+        _insert_sample(conn, stale_ts)
+        proj = compute_projection(conn, _clock())
+        # Stale data (> 48h since newest day) → not Supported
+        assert proj.confidence_state != ConfidenceState.SUPPORTED
+        conn.close()
+
+    def test_empty_store_state(self, tmp_path):
+        """Empty store → no projection, headline=None."""
+        conn = init_store(tmp_path / "test.db")
+        proj = compute_projection(conn, _clock())
+        assert proj.confidence_state == ConfidenceState.UNSUPPORTED
+        assert proj.headline_remaining_seconds is None
+        conn.close()
+
+
+# ---------------------------------------------------------------------------
+# TUI-1: One dense keyboard-first screen
+# ---------------------------------------------------------------------------
+
+class TestDenseScreen:
+    """TUI renders one dense screen with four normative regions."""
+
+    @pytest.mark.asyncio
+    async def test_four_regions_exist(self, tmp_path):
+        """All four normative regions are present in the DOM."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        conn.close()
+
+        app = FenrisTuiApp(store_path=tmp_path / "test.db")
+        async with app.run_test() as pilot:
+            assert app.query_one("#headline-band") is not None
+            assert app.query_one("#usage-history") is not None
+            assert app.query_one("#drive-health") is not None
+            assert app.query_one("#service-strip") is not None
+
+    @pytest.mark.asyncio
+    async def test_headline_contains_projection(self, tmp_path):
+        """Headline band shows projection headline."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_baseline(conn, tbw_tb=1.0, verified=True)
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        conn.close()
+
+        app = FenrisTuiApp(store_path=tmp_path / "test.db")
+        async with app.run_test() as pilot:
+            headline = str(app.query_one("#headline-band").render())
+            assert "remaining" in headline.lower() or "projection" in headline.lower()
+
+    @pytest.mark.asyncio
+    async def test_confidence_rendered_as_evidence(self, tmp_path):
+        """Confidence is state + contributing facts, never a percentage."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_baseline(conn, tbw_tb=1.0, verified=True)
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        conn.close()
+
+        app = FenrisTuiApp(store_path=tmp_path / "test.db")
+        async with app.run_test() as pilot:
+            headline = str(app.query_one("#headline-band").render())
+            assert "projection confidence" in headline.lower()
+            # Contributing facts shown, never a percentage as confidence
+            # (percentage in "95% interval coverage" is allowed as a fact, not as confidence)
+
+    @pytest.mark.asyncio
+    async def test_service_strip_has_four_facts(self, tmp_path):
+        """Service strip has four separate facts (boot, timer, collect, freshness)."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        conn.close()
+
+        app = FenrisTuiApp(store_path=tmp_path / "test.db")
+        async with app.run_test() as pilot:
+            strip = str(app.query_one("#service-strip").render())
+            assert "boot:" in strip
+            assert "timer:" in strip
+            assert "last collect:" in strip
+            assert "freshness:" in strip
+
+
+# ---------------------------------------------------------------------------
+# CI-4: Disclosures, empty-store greeting, first-run
+# ---------------------------------------------------------------------------
+
+class TestDisclosuresAndGreeting:
+    @pytest.mark.asyncio
+    async def test_disclosures_screen(self, tmp_path):
+        """Disclosures view renders six disclosures verbatim."""
+        app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
+        async with app.run_test() as pilot:
+            await pilot.press("d")
+            # Modal should be pushed
+            assert len(app.screen_stack) > 1
+            disc_text = str(app.screen.query_one("Static").render())
+            for i in range(1, 7):
+                assert "%d." % i in disc_text
+
+    @pytest.mark.asyncio
+    async def test_empty_store_greeting(self, tmp_path):
+        """Empty store shows 'no observations yet' with enable hint."""
+        app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
+        async with app.run_test() as pilot:
+            headline = str(app.query_one("#headline-band").render())
+            assert "no observations yet" in headline.lower()
+            assert "enable" in headline.lower() or "resume" in headline.lower()
+
+
+# ---------------------------------------------------------------------------
+# IN-3: First-run opt-in
+# ---------------------------------------------------------------------------
+
+class TestFirstRun:
+    @pytest.mark.asyncio
+    async def test_first_run_prompt(self, tmp_path):
+        """First-run prompt enables timer and opens first period."""
+        app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
+        async with app.run_test() as pilot:
+            headline = str(app.query_one("#headline-band").render())
+            assert "no observations yet" in headline.lower()
+            # The enable hint should mention resume
+            assert "resume" in headline.lower()
+
+
+# ---------------------------------------------------------------------------
+# TUI-4: Layout regions normative
+# ---------------------------------------------------------------------------
+
+class TestLayoutNormative:
+    @pytest.mark.asyncio
+    async def test_no_page_navigation(self, tmp_path):
+        """No page navigation keys exist (variant switching was prototype-only)."""
+        app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
+        # Check that only production bindings exist
+        binding_keys = {b.key for b in app.BINDINGS}
+        assert "left" not in binding_keys
+        assert "right" not in binding_keys
+        assert "1" not in binding_keys
+        assert "2" not in binding_keys
+        assert "3" not in binding_keys
+
+    @pytest.mark.asyncio
+    async def test_pause_resume_asymmetry(self, tmp_path):
+        """Pause asks, resume does not (TUI-2)."""
+        app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
+        async with app.run_test() as pilot:
+            # Pause should push a confirmation screen
+            await pilot.press("p")
+            assert len(app.screen_stack) > 1
+            # Press n to cancel
+            await pilot.press("n")
+            assert len(app.screen_stack) == 1
+
+
+# ---------------------------------------------------------------------------
+# CI-1: State matrix exhaustive combinations
+# ---------------------------------------------------------------------------
+
+class TestStateMatrixCombinations:
+    """CI-1: confidence × freshness × baseline tier combinations."""
+
+    def test_unsupported_with_fresh_data(self, tmp_path):
+        """Fresh data but no baseline → Unavailable + fresh."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(20):
+            d = (datetime(2026, 9, 10) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100)
+        # Sample must be within FRESH_THRESHOLD_S of clock
+        fresh_ts = (_clock() - timedelta(seconds=FRESH_THRESHOLD_S - 10)).isoformat()
+        _insert_sample(conn, fresh_ts)
+        proj = compute_projection(conn, _clock())
+        assert proj.confidence_state == ConfidenceState.UNSUPPORTED
+        freshness = grade_freshness(fresh_ts, _clock())
+        assert freshness == "fresh"
+        conn.close()
+
+    def test_limited_with_stale_data(self, tmp_path):
+        """Stale data with baseline → Limited or Unsupported (young regime + stale)."""
+        conn = init_store(tmp_path / "test.db")
+        _insert_baseline(conn, tbw_tb=1.0, verified=True)
+        _insert_segment(conn, opened_at="2026-09-01T00:00:00+00:00")
+        _open_period(conn)
+        for i in range(20):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100)
+        stale_ts = (_clock() - timedelta(days=3)).isoformat()
+        _insert_sample(conn, stale_ts)
+        proj = compute_projection(conn, _clock())
+        assert proj.confidence_state != ConfidenceState.SUPPORTED
+        freshness = grade_freshness(stale_ts, _clock())
+        assert freshness == "stale"
+        conn.close()
+
+    def test_supported_with_unverified_baseline(self, tmp_path):
+        """Incomplete provenance → unverified baseline tier."""
+        conn = init_store(tmp_path / "test.db")
+        # Insert baseline with incomplete provenance (missing source_url)
+        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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (1.0, None, "v1.0", "2026-01-01", "Samsung SSD 970 EVO Plus 1TB",
+             1024000000000, None, False,
+             "2026-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
+        )
+        conn.commit()
+        _insert_segment(conn)
+        _open_period(conn)
+        for i in range(30):
+            d = (datetime(2026, 9, 1) + timedelta(days=i)).strftime("%Y-%m-%d")
+            _insert_day(conn, d, bw=1024*1024*100, coverage=0.95, samples=24)
+        _insert_sample(conn, "2026-09-30T10:00:00+00:00")
+        proj = compute_projection(conn, _clock())
+        # Incomplete provenance → UNVERIFIED tier
+        assert proj.baseline_tier.value == "unverified_override"
+        conn.close()
+