Release Fenris 0.3.7

.gitea/workflows/release.yml

@@ -64,6 +64,20 @@ jobs:
             -o /tmp/nfpm.tar.gz
           sudo tar -xzf /tmp/nfpm.tar.gz -C /usr/local/bin nfpm
           nfpm --version
+          # Ubuntu does not package the XBPS build tools.  Use Void's static
+          # toolchain, pinned and checksum-verified before it reaches PATH.
+          XBPS_STATIC_VERSION=0.60.4_1
+          XBPS_STATIC_ARCHIVE="xbps-static-static-${XBPS_STATIC_VERSION}.x86_64-musl.tar.xz"
+          XBPS_STATIC_SHA256=603b3c55e9cabd5af79b461b929b14e1556a443c97b5714d188681c2172d9e28
+          curl --fail --silent --show-error --location \
+            "https://repo-default.voidlinux.org/static/${XBPS_STATIC_ARCHIVE}" \
+            -o "/tmp/${XBPS_STATIC_ARCHIVE}"
+          echo "${XBPS_STATIC_SHA256}  /tmp/${XBPS_STATIC_ARCHIVE}" | sha256sum --check --strict
+          mkdir -p /tmp/xbps-static
+          tar -xJf "/tmp/${XBPS_STATIC_ARCHIVE}" -C /tmp/xbps-static
+          export PATH="/tmp/xbps-static/usr/bin:${PATH}"
+          echo "/tmp/xbps-static/usr/bin" >> "$GITHUB_PATH"
+          xbps-create --version
 
       - name: Build packages
         run: make package

CHANGELOG.md

@@ -9,6 +9,26 @@ backfill releases from before this changelog.
 
 ## [Unreleased]
 
+## [0.3.7] - 2026-09-16
+
+### Changed
+
+- Make usage-history axes, units, UTC boundaries, active 7/14/30/90-day window, gaps, partial periods, stacked write attribution, and hourly drill-down explicit; keep live graph data refreshed on the existing five-minute cadence.
+
+## [0.3.6] - 2026-09-16
+
+### Changed
+
+- Use sentence case throughout the dashboard and add persistent keyboard and sudo guidance.
+- Share read-only status acquisition between the CLI and TUI, preserving unknown monitoring state and store-fault recovery guidance.
+- Use one authenticated action path for the CLI and TUI; let valid collection runs finish without the former 30-second dashboard cutoff.
+- Remove the unused sparkline and habit-bar rendering path while retaining the interactive history graph.
+- Align all package formats on the MIT license and Python 3.10 minimum; include the license text in native packages.
+
+### Fixed
+
+- Correct observation-store directory permissions in native packages, including repair of older runit installations during upgrade.
+
 ## [0.3.5] - 2026-09-14
 
 ### Added

CONTEXT.md

@@ -40,6 +40,14 @@ _Avoid_: Hourly record, hourly.jsonl entry
 One row per UTC day derived from hour observations; the grain at which usage-habit evidence is judged.
 _Avoid_: Daily summary, daily stats
 
+**Usage-history window**:
+An exact consecutive span of UTC calendar days ending today, shown from day aggregates; a day without trustworthy evidence remains an explicit gap rather than disappearing or being estimated.
+_Avoid_: Available records, dataset range
+
+**Unallocated write evidence**:
+Writes known to belong to a UTC day but which cannot be assigned honestly to a particular hour; they contribute to that day's total but are never distributed across hourly bars.
+_Avoid_: Missing writes, estimated hourly writes
+
 **Controller segment**:
 A span of observation history within which the drive's controller identity is unchanged and counters are monotonic; write deltas are never computed across a segment boundary.
 _Avoid_: Counter reset handling, drive swap detection

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 Fenris contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Makefile

@@ -74,7 +74,7 @@ install: check-python check-smartctl dist/fenris-*.whl
 	
 	@echo "=== Creating data directory (root-written, group-read) ==="
 	@sudo groupadd -f fenris
-	@sudo install -d -o root -g fenris -m 2750 $(DATA_DIR)
+	@sudo install -d -o root -g fenris -m 2770 $(DATA_DIR)
 	
 	@echo "=== Installing version-neutral runtime packages ==="
 	@sudo rm -rf $(VENV_DIR)

README.md

@@ -252,11 +252,33 @@ is retried at the next interval; it never fabricates missing observations.
 
 ## Reading the dashboard
 
-`fenris` opens the TUI dashboard.
+Run `fenris` as your normal user to open the TUI dashboard. The dashboard does
+not need `sudo`. Use `sudo` for package installation and system configuration;
+pause, resume, and collect-now actions normally authenticate through polkit.
+
+If the observation store is inaccessible, add your login user to the `fenris`
+group with `sudo usermod -aG fenris "$USER"`, then log out and back in.
+
+If polkit authentication is unavailable, quit the dashboard and run only the
+required administrative action in your terminal:
+
+```bash
+sudo fenris monitor resume   # Enable monitoring now and across reboots
+sudo fenris monitor pause    # Confirm a deliberate monitoring pause
+sudo fenris sample           # Request one collection run
+```
+
+Reopen the dashboard with `fenris` afterward. Press `?` for these instructions
+and keyboard controls at any time; use the arrow keys to scroll and `Esc` to close.
+
+The CLI and TUI share the same authenticated action path. Authentication and
+waiting for collection have no separate dashboard deadline; the native collector
+enforces its 90-second runtime limit. An interrupted or failed action is not
+automatically retried—check `fenris status` before retrying.
 
 - **Continuity** — the service strip's continuity line (and `fenris status`) reports whether monitoring survives reboots: `monitoring: active in background · persists across reboots`, or `monitoring: does not start on next boot`.
 - **Paused vs. quit** — a full-width `monitoring: paused — deliberate disable` block means collection is stopped (`fenris monitor pause`); resume with `fenris monitor resume`. Pressing `q` only leaves the screen — monitoring keeps running in the background.
-- **Auth banner** — at launch, `privileged actions will prompt for authentication (polkit)` shows once and clears on the first refresh. Privileged actions elevate via polkit; Fenris never asks for sudo.
+- **Auth banner** — the launch notice explains normal-user startup and polkit authentication, then clears on the first refresh. The `?` help screen remains available.
 
 Per-release notes live on the [releases page](https://git.bongbetic.com/xavierk/Fenris/releases): each entry is the version's `CHANGELOG.md` section — what was added, changed, and fixed — plus standing install and verification instructions.
 

docs/adr/0009-mit-license.md

@@ -0,0 +1,8 @@
+# One MIT license across source and packages
+
+The native package metadata previously disagreed: deb/rpm declared Proprietary,
+while XBPS declared MIT and the repository carried no license text. On
+2026-09-16 the maintainer chose MIT for Fenris. The repository now includes the
+standard MIT license, and Python, deb, rpm, and XBPS distributions must preserve
+that same licensing decision; bundled third-party dependencies retain their own
+license notices.

packaging/nfpm.yaml

@@ -7,7 +7,7 @@ description: >
   NVMe wear monitor with persistent TUI — observes real-world drive use and
   translates it into an understandable endurance outlook.
 homepage: https://git.bongbetic.com/xavierk/Fenris
-license: Proprietary
+license: MIT
 
 depends:
   - python3 (>= 3.10)
@@ -38,7 +38,7 @@ contents:
   - dst: /var/lib/fenris
     type: dir
     file_info:
-      mode: 2750
+      mode: 02770
       group: fenris
 
 scripts:

packaging/postinst.sh

@@ -63,7 +63,8 @@ print(f'Fenris migration: {n} step(s) applied') if n else None
         else
             # runit: create group, set directory permissions
             groupadd -f fenris
-            install -d -o root -g fenris -m 2750 "${STORE_DIR}" 2>/dev/null || true
+            install -d -o root -g fenris -m 2770 "${STORE_DIR}"
+            chmod 02770 "${STORE_DIR}"
             # Mark runit service as dormant (down) for fresh install
             if [ -d /etc/sv/fenris-collect ] && [ ! -e /var/service/fenris-collect ]; then
                 touch /etc/sv/fenris-collect/down

packaging/rpm/post.sh

@@ -24,7 +24,8 @@ if [ "$1" -eq 1 ]; then
     else
         # runit: create group, set directory permissions
         groupadd -f fenris
-        install -d -o root -g fenris -m 2750 "${STORE_DIR}" 2>/dev/null || true
+        install -d -o root -g fenris -m 2770 "${STORE_DIR}"
+        chmod 02770 "${STORE_DIR}"
         # Mark runit service as dormant (down) for fresh install
         if [ -d /etc/sv/fenris-collect ] && [ ! -e /var/service/fenris-collect ]; then
             touch /etc/sv/fenris-collect/down
@@ -69,6 +70,8 @@ print(f'Fenris migration: {n} step(s) applied') if n else None
     else
         # runit: repair log access when upgrading from an older package
         groupadd -f fenris
+        install -d -o root -g fenris -m 2770 "${STORE_DIR}"
+        chmod 02770 "${STORE_DIR}"
         install -d -o root -g fenris -m 2770 /var/log/fenris-collect 2>/dev/null || true
     fi
 fi

packaging/stage.sh

@@ -58,6 +58,9 @@ python3 -m pip install --disable-pip-version-check --no-compile \
 # observation store.
 chmod -R a+rX "${VENDOR_DIR}"
 
+# Ship the application license in every native package.
+install -D -m 0644 "${REPO_ROOT}/LICENSE" "${STAGE_DIR}/usr/share/licenses/fenris/LICENSE"
+
 # --- Inject version into wrapper from pyproject.toml ---
 # The wrapper has a hardcoded version string; patch it for packaging.
 WRAPPER_SRC="${REPO_ROOT}/scripts/fenris"

packaging/xbps/install.sh

@@ -34,6 +34,10 @@ case "${ACTION}" in
         fi
         ;;
     post)
+        # Keep observation-store access consistent across fresh installs and upgrades.
+        groupadd -f fenris
+        install -d -o root -g fenris -m 2770 "${STORE_DIR}"
+        chmod 02770 "${STORE_DIR}"
         if [ "${UPDATE}" = "yes" ]; then
             # Upgrade — snapshot, migration, runit-aware reload
             if [ -f "${STORE_DB}" ]; then
@@ -53,7 +57,6 @@ print(f'Fenris migration: {n} step(s) applied') if n else None
         else
             # Fresh install — runit service setup
             groupadd -f fenris
-            install -d -o root -g fenris -m 2770 "${STORE_DIR}" 2>/dev/null || true
             # Mark runit service as dormant (down) for fresh install
             if [ -d /etc/sv/fenris-collect ] && [ ! -e /var/service/fenris-collect ]; then
                 touch /etc/sv/fenris-collect/down

pyproject.toml

@@ -1,8 +1,9 @@
 [project]
 name = "fenris"
-version = "0.3.5"
+version = "0.3.7"
 description = "NVMe wear monitor with persistent TUI"
-requires-python = ">=3.9"
+requires-python = ">=3.10"
+license = {file = "LICENSE"}
 dependencies = [
     "textual>=0.40.0",
 ]

requirements.txt

@@ -9,3 +9,4 @@ mdurl==0.1.2
 platformdirs==4.11.7
 Pygments==2.21.0
 linkify-it-py==2.2.0
+typing-extensions==4.16.0

scripts/fenris

@@ -7,8 +7,6 @@ Subcommands route through fenris-monitor for privileged operations.
 Spec: §1.2, §8.4
 """
 import argparse
-import os
-import subprocess
 import sys
 from pathlib import Path
 
@@ -35,35 +33,16 @@ def add_runtime_packages() -> None:
 add_runtime_packages()
 
 
-def is_root() -> bool:
-    """Check if running as root."""
-    return os.geteuid() == 0
-
-
 def run_monitor(*args: str) -> None:
-    """Run fenris-monitor with the given arguments.
+    """Render the shared privileged-action outcome for the CLI."""
+    from fenris.control import MonitorError, run_monitor as invoke_monitor
 
-    If not root, re-exec under pkexec.
+    try:
-    """
+        invoke_monitor(*args)
-    monitor_cmd = "/usr/libexec/fenris/fenris-monitor"
+    except MonitorError as exc:
-
+        print(str(exc), file=sys.stderr)
-    if is_root():
+        sys.exit(exc.exit_code)
-        result = subprocess.run([monitor_cmd] + list(args))
+    sys.exit(0)
-        sys.exit(result.returncode)
-    else:
-        # Use pkexec to elevate
-        pkexec = subprocess.run(
-            ["which", "pkexec"], capture_output=True
-        )
-        if pkexec.returncode != 0:
-            print(
-                "Error: No polkit agent available. "
-                "Run as root: sudo fenris-monitor ...",
-                file=sys.stderr,
-            )
-            sys.exit(1)
-        result = subprocess.run(["pkexec", monitor_cmd] + list(args))
-        sys.exit(result.returncode)
 
 
 def cmd_tui(args: argparse.Namespace) -> None:
@@ -143,7 +122,7 @@ def main() -> None:
         description="Fenris NVMe endurance monitor",
     )
     parser.add_argument(
-        "--version", action="version", version="%(prog)s 0.3.0"
+        "--version", action="version", version="%(prog)s 0.3.6"
     )
 
     subparsers = parser.add_subparsers(dest="command")

src/fenris/__init__.py

@@ -1,2 +1,2 @@
 """Fenris: NVMe wear monitor with persistent TUI."""
-__version__ = "0.3.1"
+__version__ = "0.3.7"

src/fenris/control.py

@@ -0,0 +1,56 @@
+"""Terminal-attached invocation of Fenris's fixed privileged operations.
+
+Both human entry points use this module. Authentication has no frontend
+deadline; collection runtime is bounded by the native scheduler (ADR 0003).
+"""
+import os
+import shlex
+import subprocess
+
+
+MONITOR_HELPER = "/usr/libexec/fenris/fenris-monitor"
+
+
+class MonitorError(Exception):
+    """An action failed, with a message and exit status for either renderer."""
+
+    def __init__(self, message: str, exit_code: int = 1):
+        super().__init__(message)
+        self.exit_code = exit_code
+
+
+def run_monitor(*args: str, helper_path: str = MONITOR_HELPER) -> None:
+    """Run one helper operation, inheriting the terminal for authentication.
+
+    Never invoke a shell, retry an action, or fall back to sudo automatically.
+    The helper owns the operation allow-list and privileged state changes.
+    """
+    helper_command = [helper_path, *args]
+    needs_auth = os.geteuid() != 0
+    command = ["pkexec", *helper_command] if needs_auth else helper_command
+    root_hint = (
+        " If authentication is unavailable, run in your terminal: "
+        + shlex.join(["sudo", *helper_command])
+    ) if needs_auth else ""
+
+    try:
+        # The collector owns its 90-second runtime limit. A frontend timeout
+        # would also count time spent authenticating or waiting for a run.
+        result = subprocess.run(command)
+    except FileNotFoundError as exc:
+        raise MonitorError(
+            "Command not found: %s.%s" % (exc.filename or command[0], root_hint), 127,
+        ) from exc
+    except OSError as exc:
+        raise MonitorError("Cannot run monitoring action: %s.%s" % (exc, root_hint)) from exc
+    except KeyboardInterrupt as exc:
+        raise MonitorError(
+            "Action interrupted. Check fenris status before retrying.", 130,
+        ) from exc
+
+    if result.returncode:
+        exit_code = result.returncode if result.returncode > 0 else 128 - result.returncode
+        raise MonitorError(
+            "Action failed (exit %d). Check fenris status before retrying.%s"
+            % (exit_code, root_hint), exit_code,
+        )

src/fenris/init_system.py

@@ -166,11 +166,11 @@ def _systemd_query_state() -> Dict[str, Any]:
         "ActiveState", "ExecMainStatus", "ExecMainExitTimestamp",
     )
 
-    boot_enabled_str = timer_props.get("UnitFileState", "")
+    boot_enabled_str = timer_props.get("UnitFileState")
-    boot_enabled = boot_enabled_str == "enabled"
+    boot_enabled = boot_enabled_str == "enabled" if boot_enabled_str else None
 
-    active_state = timer_props.get("ActiveState", "inactive")
+    active_state = timer_props.get("ActiveState")
-    timer_active = active_state == "active"
+    timer_active = active_state == "active" if active_state else None
 
     last_collect_ok = None
     last_collect_age_s = None

src/fenris/status.py

@@ -14,10 +14,14 @@ Freshness constants are defined once here and shared with the TUI (§8.9):
 Criteria: LC-9, CI-2, CI-4, FL-4, FL-5, FL-7.
 """
 import sqlite3
+from contextlib import contextmanager
 import sys
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Tuple
+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 .store import SCHEMA_VERSION
@@ -98,7 +102,7 @@ def open_store_readonly(store_path: Path) -> sqlite3.Connection:
         # PermissionError before any StoreFault can be raised (issue #54).
         raise StoreFault("observation store not readable: %s" % e)
     if not exists:
-        raise StoreFault("observation store not found at %s" % store_path)
+        raise MissingStore("observation store not found at %s" % store_path)
 
     try:
         conn = sqlite3.connect("file:%s?mode=ro" % store_path, uri=True)
@@ -125,6 +129,10 @@ class StoreFault(Exception):
     pass
 
 
+class MissingStore(StoreFault):
+    """No observation history has been created yet."""
+
+
 class NewerSchema(Exception):
     """Store has a newer user_version (§9.5)."""
     def __init__(self, version: int):
@@ -246,27 +254,12 @@ def check_retired_flag(flag: str) -> Optional[str]:
 # Formatting
 # ---------------------------------------------------------------------------
 
-def _format_projection(proj, freshness: str, service: Dict[str, Any],
+def _format_projection(proj, freshness: str, drive_facts: List[str],
-                       drive_facts: List[str], config_error: Optional[str],
+                       config_error: Optional[str],
-                       store_fault: Optional[str], newer_schema: Optional[str],
-                       journal_hint: Optional[str],
                        sample_count: int = 0, day_count: int = 0) -> str:
-    """Format the complete status output."""
+    """Format projection details; monitoring status has its own renderer."""
     lines = []
 
-    # --- Store/system fault overrides (§9.4, §9.5) ---
-    if store_fault:
-        lines.append("observation store unreadable")
-        if journal_hint:
-            lines.append("")
-            lines.append("Recent journal entries:")
-            lines.append(journal_hint)
-        return "\n".join(lines)
-
-    if newer_schema:
-        lines.append("observation store written by a newer Fenris — upgrade Fenris")
-        return "\n".join(lines)
-
     # --- Configuration error (§8.3) ---
     if config_error:
         lines.append("configuration error: %s" % config_error)
@@ -277,7 +270,6 @@ def _format_projection(proj, freshness: str, service: Dict[str, Any],
         lines.append("no observations yet")
         lines.append("")
         lines.append("Enable monitoring: fenris monitor resume")
-        _append_service_facts(lines, service)
         return "\n".join(lines)
 
     # --- Single sample: awaiting another sample (issue #73 AC3) ---
@@ -287,7 +279,10 @@ def _format_projection(proj, freshness: str, service: Dict[str, Any],
         lines.append("awaiting another sample")
         lines.append("")
         lines.append("Collecting usage data — the first projection requires at least two samples.")
-        _append_service_facts(lines, service)
+        return "\n".join(lines)
+
+    if proj is None:
+        lines.append("no projection available")
         return "\n".join(lines)
 
     # --- Projection headline ---
@@ -332,16 +327,6 @@ def _format_projection(proj, freshness: str, service: Dict[str, Any],
             lines.append(fact)
         lines.append("")
 
-    # --- Four separate service facts (§7.3, LC-9) ---
-    _append_service_facts(lines, service)
-
-    # --- Journal hint on failure or staleness (§8.8) ---
-    if journal_hint:
-        if freshness in ("missed", "stale"):
-            lines.append("")
-            lines.append("Recent journal entries:")
-            lines.append(journal_hint)
-
     return "\n".join(lines)
 
 
@@ -386,36 +371,6 @@ def _format_headline(proj) -> str:
     return headline
 
 
-def _append_service_facts(lines: List[str], service: Dict[str, Any]) -> None:
-    """Append service facts and dashboard-clarity monitoring state."""
-    boot = "enabled" if service.get("boot_enabled") else "disabled"
-    activity = "active" if service.get("timer_active") else "inactive"
-
-    if service.get("last_collect_ok") is True:
-        collect = "ok"
-    elif service.get("last_collect_ok") is False:
-        collect = "FAILED"
-        if service.get("last_collect_reason"):
-            collect += " (%s)" % service["last_collect_reason"]
-    else:
-        collect = "unknown"
-
-    collect_age = ""
-    if service.get("last_collect_age_s") is not None:
-        collect_age = " %s" % freshness_age_human(service["last_collect_age_s"])
-
-    freshness_str = service.get("freshness", "unknown")
-    freshness_age = ""
-    if service.get("freshness_age_s") is not None:
-        freshness_age = " (%s)" % freshness_age_human(service["freshness_age_s"])
-
-    lines.append("boot: %s · timer: %s · last collect: %s%s · freshness: %s%s"
-                 % (boot, activity, collect, collect_age, freshness_str, freshness_age))
-    lines.append("CONTINUITY: %s" % monitoring_continuity(service))
-    if service.get("deliberately_paused"):
-        lines.extend(deliberate_pause_lines())
-
-
 # ---------------------------------------------------------------------------
 # Dashboard clarity parity wording (DC-2, DC-3)
 # ---------------------------------------------------------------------------
@@ -430,7 +385,9 @@ _PAUSED_CONSEQUENCE = (
 
 def monitoring_continuity(service: Dict[str, Any]) -> str:
     """Return the boot-persistence wording, independent of timer runtime."""
-    return _CONTINUITY_ACTIVE if service.get("boot_enabled") else _CONTINUITY_DISABLED
+    if service.get("boot_enabled") is None:
+        return "monitoring: boot persistence unknown"
+    return _CONTINUITY_ACTIVE if service["boot_enabled"] else _CONTINUITY_DISABLED
 
 
 def deliberate_pause_lines() -> List[str]:
@@ -476,121 +433,92 @@ def format_disclosures() -> str:
 # Main status entry point
 # ---------------------------------------------------------------------------
 
-def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None,
+@contextmanager
-               query_services: bool = True, query_journal: bool = True) -> str:
+def read_status(
-    """Render the complete read-only status (§8.8, LC-9).
+    store_path: Optional[Path] = None,
+    clock_now: Optional[datetime] = None,
+    query_services: bool = True,
+    collecting: bool = False,
+    reduced_motion: bool = False,
+) -> Iterator[Tuple[Optional[sqlite3.Connection], "StatusComposition"]]:
+    """Yield a read-only store snapshot and its composed monitoring status.
 
-    This is the single entry point for 'fenris status'.  It never auto-samples,
+    Own acquisition, fault classification, and connection lifetime for both
-    never prompts, and never writes to the store.
+    renderers. An absent store is empty; an unreadable or newer store exposes
+    no connection. Unknown monitoring facts are never coerced to disabled.
     """
-    if clock_now is None:
+    from .status_composition import compose_status
-        clock_now = datetime.now(timezone.utc)
 
-    # --- Configuration (§8.3) ---
+    clock_now = clock_now or datetime.now(timezone.utc)
-    config_error = None
+    service = None
-    device = None
-    try:
-        config = read_config()
-        device = config["device"]
-    except ConfigError as e:
-        config_error = str(e)
-
-    # --- Service state ---
-    service = {}
     if query_services:
-        service = query_service_state()
-
-    # --- Store open ---
-    store_fault = None
-    newer_schema = None
-    conn = None
-
-    if store_path is None:
-        store_path = Path("/var/lib/fenris/observations.db")
-
-    try:
-        conn = open_store_readonly(store_path)
-    except StoreFault as e:
-        store_fault = str(e)
-    except NewerSchema as e:
-        newer_schema = str(e)
-
-    # --- Store fault / newer schema short-circuit ---
-    if store_fault or newer_schema:
-        journal_hint = None
-        if query_journal:
-            journal_hint = _journalctl_hint()
-        service["freshness"] = "unknown"
-        service["freshness_age_s"] = None
-        return _format_projection(
-            None, "unknown", service, [], config_error, store_fault, newer_schema, journal_hint
-        )
-
-    # --- Freshness grading (§8.9) ---
-    try:
-        cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1")
-        row = cursor.fetchone()
-        newest_ts = row[0] if row else None
-    except sqlite3.Error:
-        newest_ts = None
-
-    freshness = grade_freshness(newest_ts, clock_now)
-
-    # --- Sample count for single-sample state (issue #73 AC3) ---
-    sample_count = 0
-    day_count = 0
-    try:
-        cursor = conn.execute("SELECT COUNT(*) FROM samples")
-        sample_count = cursor.fetchone()[0]
-        cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates")
-        day_count = cursor.fetchone()[0]
-    except sqlite3.Error:
-        pass
-
-    # Freshness age for the service fact
-    freshness_age_s = None
-    if newest_ts:
         try:
-            ts = datetime.fromisoformat(newest_ts)
+            service = query_service_state()
-            if ts.tzinfo is None:
+        except (OSError, RuntimeError):
-                ts = ts.replace(tzinfo=timezone.utc)
-            freshness_age_s = int((clock_now - ts).total_seconds())
-        except (ValueError, TypeError):
             pass
 
-    service["freshness"] = freshness
+    conn = None
-    service["freshness_age_s"] = freshness_age_s
+    store_fault = newer_schema = None
     try:
-        service["deliberately_paused"] = is_deliberately_paused(conn, service)
+        try:
-    except sqlite3.Error:
+            conn = open_store_readonly(store_path or Path("/var/lib/fenris/observations.db"))
-        service["deliberately_paused"] = False
+            conn.execute("BEGIN")
+        except MissingStore:
+            pass
+        except (StoreFault, sqlite3.Error) as exc:
+            store_fault = str(exc)
+        except NewerSchema as exc:
+            newer_schema = str(exc)
 
-    # --- Drive anomalies (§9.7, FL-7) ---
+        comp = compose_status(
-    drive_facts = []
+            conn, service, clock_now, store_fault=store_fault,
+            newer_schema=newer_schema, collecting=collecting,
+            reduced_motion=reduced_motion,
+        )
+        if conn is not None and (comp.store_fault or comp.newer_schema):
+            conn.close()
+            conn = None
+        yield conn, comp
+    finally:
+        if conn is not None:
+            conn.close()
+
+
+def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None,
+               query_services: bool = True, query_journal: bool = True) -> str:
+    """Render CLI status through the shared read-only acquisition path."""
+    from .status_composition import render_status_cli
+
+    clock_now = clock_now or datetime.now(timezone.utc)
+    config_error = None
     try:
-        drive_facts = _query_drive_facts(conn)
+        read_config()
-    except sqlite3.Error:
+    except ConfigError as exc:
-        pass
+        config_error = str(exc)
 
-    # --- Projection (§6 — recomputed on read, never stored) ---
+    with read_status(store_path, clock_now, query_services) as (conn, comp):
-    try:
+        parts = [render_status_cli(comp)]
-        proj = compute_projection(conn, clock_now)
+        if not comp.store_fault and not comp.newer_schema:
-    except Exception:
+            drive_facts = []
-        proj = None
+            proj = None
-
+            if conn is not None:
-    # --- Journal hint on failure or staleness (§8.8) ---
+                try:
-    journal_hint = None
+                    drive_facts = _query_drive_facts(conn)
-    if query_journal and freshness in ("missed", "stale"):
+                    proj = compute_projection(conn, clock_now)
-        journal_hint = _journalctl_hint()
+                except (sqlite3.Error, ValueError, TypeError):
-
+                    pass
-    # --- Compose output ---
+            parts.append(_format_projection(
-    result = _format_projection(
+                proj, comp.freshness, drive_facts, config_error,
-        proj, freshness, service, drive_facts, config_error,
+                comp.sample_count, comp.day_count,
-        None, None, journal_hint, sample_count, day_count,
+            ))
-    )
+        if query_journal and (
-
+            comp.store_fault or comp.last_collect_ok is False
-    conn.close()
+            or comp.freshness in ("missed", "stale")
-    return result
+        ):
+            hint = _journalctl_hint()
+            if hint:
+                parts.append("Recent collector logs:\n" + hint)
+        return "\n\n".join(part for part in parts if part)
 
 
 def render_status(store_path: Optional[Path] = None, clock_now: Optional[datetime] = None,
@@ -619,65 +547,8 @@ def get_status_composition(
     collecting: bool = False,
     reduced_motion: bool = False,
 ) -> 'StatusComposition':
-    """Get the shared status composition consumed by both TUI and CLI.
+    """Return monitoring status without retaining the read-only snapshot."""
-
+    with read_status(
-    This is the new entry point that centralizes the status lattice.
+        store_path, clock_now, query_services, collecting, reduced_motion,
-    """
+    ) as (_, comp):
-    # Lazy import to avoid circular dependency
+        return comp
-    from .status_composition import (
-        StatusComposition,
-        compose_status,
-    )
-
-    if clock_now is None:
-        clock_now = datetime.now(timezone.utc)
-
-    # --- Configuration (§8.3) ---
-    # Config errors are surfaced through the store fault mechanism
-
-    # --- Service state ---
-    service = {}
-    if query_services:
-        try:
-            service = query_service_state()
-        except Exception:
-            service = None
-
-    # --- Store open ---
-    store_fault = None
-    newer_schema = None
-    conn = None
-
-    if store_path is None:
-        store_path = Path("/var/lib/fenris/observations.db")
-
-    try:
-        conn = open_store_readonly(store_path)
-    except StoreFault as e:
-        store_fault = str(e)
-    except NewerSchema as e:
-        newer_schema = str(e)
-
-    # --- Use shared composition ---
-    if conn is not None:
-        try:
-            comp = compose_status(
-                conn, service, clock_now,
-                store_fault=store_fault,
-                newer_schema=newer_schema,
-                collecting=collecting,
-                reduced_motion=reduced_motion,
-            )
-        finally:
-            conn.close()
-    else:
-        # Store fault or newer schema - compose without store data
-        comp = compose_status(
-            None, service, clock_now,
-            store_fault=store_fault,
-            newer_schema=newer_schema,
-            collecting=collecting,
-            reduced_motion=reduced_motion,
-        )
-
-    return comp

src/fenris/status_composition.py

@@ -188,7 +188,7 @@ def _determine_explanation(
 ) -> str:
     """Determine the explanation line for the status state."""
     if store_fault:
-        return "observation store unreadable — see journal"
+        return "observation store unreadable — see collector logs"
 
     if newer_schema:
         return "observation store written by a newer Fenris — upgrade Fenris"
@@ -200,7 +200,7 @@ def _determine_explanation(
             if last_collect_reason:
                 parts.append("(%s)" % last_collect_reason)
         if freshness_age_s is not None and freshness != "empty":
-            parts.append("· last good sample %s ago" % freshness_age_human(freshness_age_s))
+            parts.append("· last good sample %s" % freshness_age_human(freshness_age_s))
         return " ".join(parts) if parts else "last run failed"
 
     if state == StatusState.INTERRUPTED:
@@ -211,7 +211,7 @@ def _determine_explanation(
 
     if state == StatusState.STALE:
         if freshness_age_s is not None:
-            return "last sample %s ago" % freshness_age_human(freshness_age_s)
+            return "last sample %s" % freshness_age_human(freshness_age_s)
         return "data is stale"
 
     if state == StatusState.WAITING:
@@ -223,7 +223,7 @@ def _determine_explanation(
 
     if state == StatusState.MONITORING:
         if freshness_age_s is not None:
-            return "last sample %s ago" % freshness_age_human(freshness_age_s)
+            return "last sample %s" % freshness_age_human(freshness_age_s)
         return "monitoring active"
 
     if state == StatusState.UNKNOWN:
@@ -257,7 +257,7 @@ def _determine_collecting_overlay(
 
 
 def compose_status(
-    conn: sqlite3.Connection,
+    conn: Optional[sqlite3.Connection],
     service: Optional[Dict[str, Any]],
     clock_now: datetime,
     store_fault: Optional[str] = None,
@@ -269,14 +269,17 @@ def compose_status(
 
     This is the single entry point consumed by both TUI and CLI.
     """
-    service_available = service is not None and len(service) > 0
+    service_available = bool(service) and any(
+        service.get(key) is not None
+        for key in ("boot_enabled", "timer_active")
+    )
 
     # --- Separate facts from service ---
-    boot_enabled = service.get("boot_enabled") if service_available else None
+    boot_enabled = service.get("boot_enabled") if service else None
-    timer_active = service.get("timer_active") if service_available else None
+    timer_active = service.get("timer_active") if service else None
-    last_collect_ok = service.get("last_collect_ok") if service_available else None
+    last_collect_ok = service.get("last_collect_ok") if service else None
-    last_collect_age_s = service.get("last_collect_age_s") if service_available else None
+    last_collect_age_s = service.get("last_collect_age_s") if service else None
-    last_collect_reason = service.get("last_collect_reason") if service_available else None
+    last_collect_reason = service.get("last_collect_reason") if service else None
 
     # --- Freshness from store ---
     freshness = "unknown"
@@ -285,7 +288,10 @@ def compose_status(
     day_count = 0
     deliberately_paused = False
 
-    if store_fault is None and newer_schema is None:
+    if conn is None and store_fault is None and newer_schema is None:
+        freshness = "empty"
+
+    if conn is not None and store_fault is None and newer_schema is None:
         try:
             cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1")
             row = cursor.fetchone()
@@ -301,28 +307,35 @@ def compose_status(
                     freshness_age_s = int((clock_now - ts).total_seconds())
                 except (ValueError, TypeError):
                     pass
-        except sqlite3.Error:
+        except sqlite3.Error as exc:
-            freshness = "unknown"
+            store_fault = str(exc)
 
         try:
             cursor = conn.execute("SELECT COUNT(*) FROM samples")
             sample_count = cursor.fetchone()[0]
             cursor = conn.execute("SELECT COUNT(*) FROM day_aggregates")
             day_count = cursor.fetchone()[0]
-        except sqlite3.Error:
+        except sqlite3.Error as exc:
-            pass
+            store_fault = str(exc)
 
         try:
             svc_for_pause = service if service_available else {}
             deliberately_paused = is_deliberately_paused(conn, svc_for_pause)
-        except sqlite3.Error:
+        except sqlite3.Error as exc:
-            deliberately_paused = False
+            store_fault = str(exc)
+
+    if store_fault or newer_schema:
+        freshness = "unknown"
+        freshness_age_s = None
+        sample_count = day_count = 0
+        deliberately_paused = False
 
     # --- External stop detection ---
     # External stop = timer inactive + boot disabled + NOT deliberately paused
     # + period still open (the timer was stopped but Fenris didn't close the period)
     external_stop_reason = None
-    if not deliberately_paused and timer_active is False and boot_enabled is False:
+    if (conn is not None and not store_fault and not newer_schema
+            and not deliberately_paused and timer_active is False and boot_enabled is False):
         # Check if there's an open monitoring period (external stop left it open)
         try:
             open_period = conn.execute(
@@ -428,10 +441,9 @@ def render_status_cli(comp: StatusComposition) -> str:
 
     # Separate facts
     facts = []
-    if comp.freshness != "unknown":
+    facts.append("freshness: %s" % comp.freshness)
-        facts.append("freshness: %s" % comp.freshness)
+    if comp.freshness_age_s is not None and facts:
-    if comp.freshness_age_s is not None:
+        facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s)
-        facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s) if facts else "freshness: %s" % freshness_age_human(comp.freshness_age_s)
     if comp.last_collect_ok is True:
         facts.append("last collect: ok")
     elif comp.last_collect_ok is False:
@@ -441,10 +453,12 @@ def render_status_cli(comp: StatusComposition) -> str:
         facts.append(collect_str)
     else:
         facts.append("last collect: unknown")
-    if comp.boot_enabled is not None:
+    facts.append("boot: %s" % (
-        facts.append("boot: %s" % ("enabled" if comp.boot_enabled else "disabled"))
+        "unknown" if comp.boot_enabled is None else "enabled" if comp.boot_enabled else "disabled"
-    if comp.timer_active is not None:
+    ))
-        facts.append("timer: %s" % ("active" if comp.timer_active else "inactive"))
+    facts.append("timer: %s" % (
+        "unknown" if comp.timer_active is None else "active" if comp.timer_active else "inactive"
+    ))
 
     if facts:
         lines.append(" · ".join(facts))
@@ -486,29 +500,30 @@ def render_status_tui(comp: StatusComposition) -> str:
 
     # Explanation
     if comp.explanation:
-        lines.append(comp.explanation)
+        lines.append(comp.explanation[:1].upper() + comp.explanation[1:])
 
     lines.append("")
 
     # Separate facts
     facts = []
-    if comp.freshness != "unknown":
+    facts.append("Freshness: %s" % comp.freshness)
-        facts.append("freshness: %s" % comp.freshness)
     if comp.freshness_age_s is not None and facts:
         facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s)
     if comp.last_collect_ok is True:
-        facts.append("last collect: ok")
+        facts.append("Last collect: ok")
     elif comp.last_collect_ok is False:
-        collect_str = "last collect: FAILED"
+        collect_str = "Last collect: failed"
         if comp.last_collect_reason:
             collect_str += " (%s)" % comp.last_collect_reason
         facts.append(collect_str)
     else:
-        facts.append("last collect: unknown")
+        facts.append("Last collect: unknown")
-    if comp.boot_enabled is not None:
+    facts.append("Boot: %s" % (
-        facts.append("boot: %s" % ("enabled" if comp.boot_enabled else "disabled"))
+        "unknown" if comp.boot_enabled is None else "enabled" if comp.boot_enabled else "disabled"
-    if comp.timer_active is not None:
+    ))
-        facts.append("timer: %s" % ("active" if comp.timer_active else "inactive"))
+    facts.append("Timer: %s" % (
+        "unknown" if comp.timer_active is None else "active" if comp.timer_active else "inactive"
+    ))
 
     if facts:
         lines.append(" · ".join(facts))
@@ -516,11 +531,11 @@ def render_status_tui(comp: StatusComposition) -> str:
     # Continuity
     if comp.continuity:
         lines.append("")
-        lines.append("[bold]CONTINUITY[/bold]  %s" % comp.continuity)
+        lines.append("[bold]Continuity[/bold]  %s" % comp.continuity)
 
     # Deliberate pause
     if comp.paused_lines:
         for pl in comp.paused_lines:
-            lines.append(pl)
+            lines.append(pl[:1].upper() + pl[1:])
 
     return "\n".join(lines)

tests/test_acceptance_sweep.py

@@ -461,7 +461,7 @@ class TestCI2Parity:
         }
 
         with patch("fenris.status.query_service_state", return_value=service_state), patch(
-            "fenris.tui.query_service_state", return_value=service_state
+            "fenris.status.query_service_state", return_value=service_state
         ):
             status = get_status(
                 store_path=db, clock_now=_clock(), query_services=True, query_journal=False
@@ -502,18 +502,6 @@ class TestCI2Parity:
                                 query_services=True, query_journal=False)
         assert "no observations yet" in status.lower()
 
-    def test_store_fault_phrase_both_views(self, tmp_path):
-        status_src = (FENRIS_PKG / "status.py").read_text()
-        tui_src = (FENRIS_PKG / "tui.py").read_text()
-        phrase = "observation store unreadable"
-        assert phrase in status_src
-        assert phrase.lower() in tui_src.lower()
-
-    def test_newer_schema_phrase_both_views(self):
-        status_src = (FENRIS_PKG / "status.py").read_text()
-        phrase = "observation store written by a newer Fenris"
-        assert phrase in status_src
-
     def test_status_never_prompts(self):
         status_src = (FENRIS_PKG / "status.py").read_text()
         assert "input(" not in status_src
@@ -603,12 +591,6 @@ class TestCI3ProhibitionSet:
                 table_names.append(m.group(1))
         assert "projection" not in [t.lower() for t in table_names]
 
-    def test_no_partial_newer_schema_interpretation(self):
-        """Readers refuse newer-schema stores. [3.6, 9.5]"""
-        status_src = (FENRIS_PKG / "status.py").read_text()
-        assert "NewerSchema" in status_src
-        assert "upgrade Fenris" in status_src
-
     def test_polkit_authorizes_one_binary(self):
         """Polkit authorizes exactly one binary: fenris-monitor. [8.5]"""
         monitor_src = (FENRIS_PKG / "monitor.py").read_text()
@@ -688,18 +670,6 @@ class TestCI4WordingAndDisclosures:
         proj_src = (FENRIS_PKG / "projection.py").read_text()
         assert phrase in proj_src
 
-    def test_store_fault_phrase(self):
-        phrase = "observation store unreadable"
-        status_src = (FENRIS_PKG / "status.py").read_text()
-        assert phrase in status_src
-        tui_src = (FENRIS_PKG / "tui.py").read_text()
-        assert phrase.lower() in tui_src.lower()
-
-    def test_newer_schema_phrase(self):
-        phrase = "observation store written by a newer Fenris"
-        status_src = (FENRIS_PKG / "status.py").read_text()
-        assert phrase in status_src
-
     def test_no_observations_phrase(self):
         phrase = "no observations yet"
         status_src = (FENRIS_PKG / "status.py").read_text()

tests/test_collector_history_tracer.py

@@ -412,7 +412,7 @@ class TestDisplayStates:
 
     def test_one_sample_awaiting_another_in_tui(self, tmp_path):
         """One sample → TUI shows awaiting state."""
-        from fenris.tui import FenrisTuiApp, _query_service_facts
+        from fenris.tui import FenrisTuiApp
         db = tmp_path / "test.db"
         conn = init_store(db)
         conn.execute(

tests/test_control.py

@@ -0,0 +1,77 @@
+"""The CLI and TUI cross the same terminal-attached action interface."""
+import argparse
+import runpy
+import shlex
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+from fenris.control import MONITOR_HELPER, MonitorError, run_monitor
+from fenris.tui import FenrisTuiApp
+
+
+@pytest.mark.parametrize("uid", [0, 1000])
+@pytest.mark.parametrize("args", [("enable", "--now"), ("disable", "--now"), ("collect",)])
+def test_fixed_helper_and_terminal_attachment(uid, args):
+    with patch("fenris.control.os.geteuid", return_value=uid), \
+         patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], 0)) as run:
+        run_monitor(*args)
+    expected = [MONITOR_HELPER, *args]
+    if uid:
+        expected.insert(0, "pkexec")
+    # Inherited stdin/out/err keep authentication on the user's terminal.
+    # No frontend deadline can cut short a valid 90-second Collection run.
+    run.assert_called_once_with(expected)
+
+
+@pytest.mark.parametrize("code", [1, 126, 127, -15])
+def test_failure_is_not_retried_and_root_equivalent_preserves_arguments(code):
+    args = ("baseline", "set", '{"model": "Drive $(whoami)", "tbw": 100}')
+    with patch("fenris.control.os.geteuid", return_value=1000), \
+         patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], code)) as run:
+        with pytest.raises(MonitorError) as failure:
+            run_monitor(*args)
+    run.assert_called_once()
+    assert failure.value.exit_code == (code if code > 0 else 128 - code)
+    root_command = str(failure.value).split("run in your terminal: ")[1]
+    assert shlex.split(root_command) == ["sudo", MONITOR_HELPER, *args]
+
+
+@pytest.mark.parametrize("uid,missing", [(0, MONITOR_HELPER), (1000, "pkexec")])
+def test_missing_command_is_identified(uid, missing):
+    with patch("fenris.control.os.geteuid", return_value=uid), \
+         patch("fenris.control.subprocess.run", side_effect=FileNotFoundError(2, "Missing", missing)):
+        with pytest.raises(MonitorError, match="Command not found") as failure:
+            run_monitor("collect")
+    assert missing in str(failure.value)
+    assert failure.value.exit_code == 127
+    assert ("sudo" in str(failure.value)) == bool(uid)
+
+
+def test_interrupt_reports_uncertain_outcome():
+    with patch("fenris.control.subprocess.run", side_effect=KeyboardInterrupt):
+        with pytest.raises(MonitorError, match="Check fenris status") as failure:
+            run_monitor("collect")
+    assert failure.value.exit_code == 130
+
+
+def test_cli_and_tui_show_the_same_failure(tmp_path, capsys):
+    # The launcher may prepend an installed runtime while loading; isolate it.
+    with patch.object(sys, "path", sys.path.copy()):
+        cli = runpy.run_path(str(Path(__file__).parent.parent / "scripts" / "fenris"))
+    with patch("fenris.control.os.geteuid", return_value=1000), \
+         patch("fenris.control.subprocess.run", return_value=subprocess.CompletedProcess([], 126)):
+        with pytest.raises(SystemExit) as exit_info:
+            cli["cmd_monitor_resume"](argparse.Namespace())
+        assert exit_info.value.code == 126
+        cli_message = capsys.readouterr().err.strip()
+        app = FenrisTuiApp(store_path=tmp_path / "missing.db")
+        with patch.object(app, "suspend"), patch.object(app, "_refresh") as refresh, \
+             patch.object(app, "notify") as notify:
+            app.action_resume()
+        notify.assert_called_once_with(cli_message, severity="error")
+        refresh.assert_called_once()

tests/test_init_system.py

@@ -283,7 +283,7 @@ class TestSystemdQueryState:
     def test_query_state_disabled_inactive(self):
         """Query state for disabled and inactive timer."""
         with patch("fenris.init_system._systemctl_show") as mock_show:
-            mock_show.return_value = {}
+            mock_show.return_value = {"UnitFileState": "disabled", "ActiveState": "inactive"}
             result = _systemd_query_state()
             assert result["boot_enabled"] is False
             assert result["timer_active"] is False
@@ -688,5 +688,5 @@ class TestEdgeCases:
         with patch("fenris.init_system._systemctl_show") as mock_show:
             mock_show.return_value = {}
             result = _systemd_query_state()
-            assert result["boot_enabled"] is False
+            assert result["boot_enabled"] is None
-            assert result["timer_active"] is False
+            assert result["timer_active"] is None

tests/test_issue_79.py

@@ -260,7 +260,7 @@ class TestPreservedBehavior:
         conn.close()
         app = FenrisTuiApp(store_path=tmp_path / "test.db")
 
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": True, "timer_active": True,
             "last_collect_ok": True, "last_collect_age_s": 60,
             "last_collect_reason": None,
@@ -272,11 +272,11 @@ class TestPreservedBehavior:
 
     @pytest.mark.asyncio
     async def test_quit_rail_preserved(self, tmp_path):
-        """Separate q QUIT TUI rail preserved."""
+        """Separate q Quit TUI rail preserved."""
         app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
         async with app.run_test(size=(120, 24)) as pilot:
             rail = str(app.query_one("#quit-rail").render())
-            assert "q QUIT TUI" in rail
+            assert "q Quit TUI" in rail
 
     @pytest.mark.asyncio
     async def test_auth_banner_preserved(self, tmp_path):
@@ -307,7 +307,7 @@ class TestPreservedBehavior:
         conn.close()
         app = FenrisTuiApp(store_path=db)
 
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": False, "timer_active": False,
             "last_collect_ok": None, "last_collect_age_s": None,
             "last_collect_reason": None,

tests/test_packaging.py

@@ -135,7 +135,7 @@ RPM_TARGETS = [
 ]
 
 XBPS_TARGETS = [
-    ("voidlinux/void-linux:latest", "xbps"),
+    ("ghcr.io/void-linux/void-glibc-full:latest", "xbps"),
 ]
 
 ALL_TARGETS = DEB_TARGETS + RPM_TARGETS + XBPS_TARGETS
@@ -176,9 +176,8 @@ def _build_xbps_dockerfile(image: str, pkg_name: str) -> str:
     """Dockerfile for testing xbps installation."""
     return textwrap.dedent(f"""\
         FROM {image}
-        RUN xbps-install -Suyn void-repo-nonfree && \\
+        RUN xbps-install -Sy python3 smartmontools runit shadow && \\
-            xbps-install -Syun python3 smartmontools runit && \\
+            xbps-remove -O
-            xbps-remove -Oy void-repo-nonfree || true
         COPY dist/{pkg_name} /pkg/{pkg_name}
     """)
 
@@ -1713,7 +1712,7 @@ def test_xbps_dormant_install(skip_no_docker, version):
         check=True,
     )
     (build_dir / "Dockerfile").write_text(
-        _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
+        _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
     )
 
     tag = "fenris-test-xbps"
@@ -1731,8 +1730,17 @@ def test_xbps_dormant_install(skip_no_docker, version):
     )
 
     try:
-        rc, out = _xbps_extract_and_install(container, pkg_name, action="install")
+        # Exercise XBPS itself, including its file ownership and lifecycle hooks.
+        # The temporary local repository does not need release signatures.
+        rc, out = _container_exec(
+            container,
+            f"mkdir -p /tmp/fenris-repo && cp /pkg/{pkg_name} /tmp/fenris-repo/ && "
+            f"xbps-rindex -a /tmp/fenris-repo/{pkg_name} && "
+            "xbps-install -y -R /tmp/fenris-repo fenris",
+        )
         assert rc == 0, f"XBPS install failed: {out}"
+        rc, out = _container_exec(container, "xbps-query -p pkgver fenris")
+        assert rc == 0 and out.strip() == f"fenris-{version}_1"
         _assert_runit_dormant_layout(container, version)
     finally:
         subprocess.run(
@@ -1755,7 +1763,7 @@ def test_xbps_migration_guard(skip_no_docker, version):
         check=True,
     )
     (build_dir / "Dockerfile").write_text(
-        _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
+        _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
     )
 
     tag = "fenris-guard-xbps"
@@ -1803,7 +1811,7 @@ def test_xbps_upgrade_semantics(skip_no_docker, version):
         check=True,
     )
     (build_dir / "Dockerfile").write_text(
-        _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
+        _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
     )
 
     tag = "fenris-upgrade-xbps"
@@ -1889,7 +1897,7 @@ def test_xbps_remove_preserves_config_and_store(skip_no_docker, version):
         check=True,
     )
     (build_dir / "Dockerfile").write_text(
-        _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
+        _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
     )
 
     tag = "fenris-remove-xbps"
@@ -1960,7 +1968,7 @@ def test_xbps_purge_removes_everything(skip_no_docker, version):
         check=True,
     )
     (build_dir / "Dockerfile").write_text(
-        _build_xbps_dockerfile("voidlinux/void-linux:latest", pkg_name)
+        _build_xbps_dockerfile("ghcr.io/void-linux/void-glibc-full:latest", pkg_name)
     )
 
     tag = "fenris-purge-xbps"

tests/test_status.py

@@ -622,6 +622,7 @@ class TestProjectionInStatus:
         from fenris.status import get_status
 
         nonexistent = tmp_path / "nonexistent.db"
+        nonexistent.write_bytes(b"not a SQLite database")
         now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
 
         with patch("fenris.status.query_service_state", return_value={

tests/test_status_composition.py

@@ -587,7 +587,7 @@ class TestCLIStatusRendering:
                               newer_schema=None)
         cli_text = render_status_cli(comp)
         assert "observation store unreadable" in cli_text
-        assert "see journal" in cli_text.lower()
+        assert "see collector logs" in cli_text.lower()
         conn.close()
 
 
@@ -653,7 +653,7 @@ class TestTUIStatusRendering:
         comp = compose_status(conn, svc, now, store_fault="observation store unreadable",
                               newer_schema=None)
         tui_text = render_status_tui(comp)
-        assert "observation store unreadable" in tui_text
+        assert "Observation store unreadable" in tui_text
         conn.close()
 
 
@@ -966,7 +966,7 @@ class TestTUIIntegration:
         conn.close()
 
         app = FenrisTuiApp(store_path=tmp_path / "test.db")
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": True, "timer_active": True,
             "last_collect_ok": True, "last_collect_age_s": 120,
             "last_collect_reason": None,
@@ -992,7 +992,7 @@ class TestTUIIntegration:
         conn.close()
 
         app = FenrisTuiApp(store_path=db)
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": False, "timer_active": False,
             "last_collect_ok": None, "last_collect_age_s": None,
             "last_collect_reason": None,
@@ -1019,7 +1019,7 @@ class TestTUIIntegration:
         conn.close()
 
         app = FenrisTuiApp(store_path=tmp_path / "test.db")
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": True, "timer_active": True,
             "last_collect_ok": False, "last_collect_age_s": 60,
             "last_collect_reason": "exit code 3",

tests/test_status_read.py

@@ -0,0 +1,150 @@
+"""Exercise shared status acquisition through the CLI and running TUI."""
+import sqlite3
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
+
+from fenris.status import get_status, get_status_composition, read_status
+from fenris.status_composition import StatusState
+from fenris.store import init_store, SCHEMA_VERSION
+from fenris.tui import FenrisTuiApp
+
+
+NOW = datetime(2026, 9, 16, 12, tzinfo=timezone.utc)
+ACTIVE = {"boot_enabled": True, "timer_active": True, "last_collect_ok": True}
+DISABLED = {"boot_enabled": False, "timer_active": False, "last_collect_ok": None}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("kind", ["missing", "corrupt", "newer", "incomplete", "denied"])
+@pytest.mark.parametrize("service", [ACTIVE, DISABLED, {}])
+async def test_cli_tui_acquisition_parity(tmp_path, kind, service):
+    path = tmp_path / "observations.db"
+    if kind == "corrupt":
+        path.write_bytes(b"Not a SQLite database")
+    elif kind == "incomplete":
+        sqlite3.connect(path).close()
+    elif kind != "missing":
+        conn = init_store(path)
+        if kind == "newer":
+            conn.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION + 1))
+        conn.close()
+    original_exists = Path.exists
+
+    def exists(target):
+        if kind == "denied" and target == path:
+            raise PermissionError("Observation store access denied")
+        return original_exists(target)
+
+    before = path.read_bytes() if original_exists(path) else None
+    with patch("fenris.status.query_service_state", return_value=service), \
+         patch("fenris.status._journalctl_hint", return_value="[bold]Native collector log[/bold]"), \
+         patch("fenris.tui._journalctl_hint", return_value="[bold]Native collector log[/bold]"), \
+         patch.object(Path, "exists", exists), \
+         patch("fenris.tui.compute_projection") as projection:
+        comp = get_status_composition(path, NOW)
+        cli = get_status(path, NOW).lower()
+        app = FenrisTuiApp(store_path=path)
+        async with app.run_test(size=(80, 24)) as pilot:
+            headline = str(app.query_one("#headline-band").render()).lower()
+            strip = str(app.query_one("#service-strip").render()).lower()
+            if kind == "missing":
+                assert comp.store_fault is None
+                assert comp.freshness == "empty"
+                assert "no observations yet" in cli and "no observations yet" in headline
+            else:
+                assert comp.state == StatusState.ERROR
+                assert comp.freshness == "unknown"
+                phrase = "newer fenris — upgrade fenris" if kind == "newer" else "observation store unreadable"
+                assert phrase in cli and phrase in headline
+                if kind != "newer":
+                    assert str(app.query_one("#drive-health").render()) == "[bold]Native collector log[/bold]"
+                assert not app.query_one("#main-grid").has_class("paused")
+                await pilot.resize_terminal(60, 18)
+                await pilot.pause()
+                assert phrase in str(app.query_one("#constrained-summary").render()).lower()
+            for text in (cli, strip):
+                assert ("boot: enabled" if service == ACTIVE else "boot: disabled" if service == DISABLED else "boot: unknown") in text
+                assert ("timer: active" if service == ACTIVE else "timer: inactive" if service == DISABLED else "timer: unknown") in text
+                if not service:
+                    assert "does not start on next boot" not in text
+            projection.assert_not_called()
+    assert (path.read_bytes() if original_exists(path) else None) == before
+
+
+def test_reader_owns_readonly_snapshot_and_closes_on_error(tmp_path):
+    path = tmp_path / "observations.db"
+    writer = init_store(path)
+    with patch("fenris.status.query_service_state", return_value=ACTIVE) as query:
+        with pytest.raises(RuntimeError, match="renderer failed"):
+            with read_status(path, NOW) as (conn, comp):
+                assert comp.sample_count == 0
+                assert conn.in_transaction
+                with pytest.raises(sqlite3.OperationalError, match="readonly"):
+                    conn.execute("DELETE FROM samples")
+                writer.execute("INSERT INTO monitoring_periods (started_at) VALUES ('2026-09-16')")
+                writer.commit()
+                assert conn.execute("SELECT COUNT(*) FROM monitoring_periods").fetchone()[0] == 0
+                raise RuntimeError("renderer failed")
+        query.assert_called_once()
+    with pytest.raises(sqlite3.ProgrammingError, match="closed"):
+        conn.execute("SELECT 1")
+    writer.close()
+
+
+def test_monitoring_query_failure_is_unknown(tmp_path):
+    path = tmp_path / "observations.db"
+    init_store(path).close()
+    with patch("fenris.status.query_service_state", side_effect=OSError("Unavailable")):
+        comp = get_status_composition(path, NOW)
+        cli = get_status(path, NOW, query_journal=False)
+    assert comp.state == StatusState.UNKNOWN
+    assert comp.boot_enabled is None and comp.timer_active is None
+    assert "boot: unknown" in cli and "timer: unknown" in cli
+    assert "does not start on next boot" not in cli
+
+
+def test_native_systemd_query_failure_is_unknown(tmp_path):
+    from fenris.init_system import InitSystem
+
+    with patch("fenris.init_system.get_init_system", return_value=InitSystem.SYSTEMD), \
+         patch("fenris.init_system._systemctl_show", return_value={}):
+        comp = get_status_composition(tmp_path / "missing.db", NOW)
+    assert comp.state == StatusState.UNKNOWN
+    assert comp.boot_enabled is None and comp.timer_active is None
+
+
+@pytest.mark.asyncio
+async def test_new_store_fault_clears_paused_views_and_can_recover(tmp_path):
+    path = tmp_path / "observations.db"
+    conn = init_store(path)
+    conn.execute(
+        "INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
+        "VALUES ('2026-09-15', '2026-09-16', 'user_disabled')"
+    )
+    conn.commit()
+    conn.close()
+    with patch("fenris.status.query_service_state", return_value=DISABLED):
+        app = FenrisTuiApp(store_path=path)
+        async with app.run_test() as pilot:
+            assert app.query_one("#main-grid").has_class("paused")
+            writer = sqlite3.connect(path)
+            writer.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION + 1))
+            writer.close()
+            app.on_refresh_tick()
+            await pilot.pause()
+            assert not app.query_one("#main-grid").has_class("paused")
+            assert str(app.query_one("#drive-health").render()) == ""
+            assert "upgrade Fenris" in str(app.query_one("#headline-band").render())
+            assert app.query_one("#usage-history")._day_data == []
+            writer = sqlite3.connect(path)
+            writer.execute("PRAGMA user_version = %d" % SCHEMA_VERSION)
+            writer.close()
+            app.on_refresh_tick()
+            await pilot.pause()
+            assert app.query_one("#main-grid").has_class("paused")

tests/test_tui.py

@@ -42,11 +42,7 @@ from fenris.tui import (
     FenrisTuiApp,
     DailyBarGraph,
     _format_remaining,
-    _sparkline,
-    _habit_bar,
-    _query_usage_history,
     _query_drive_health,
-    _query_service_facts,
     _query_daily_graph_data,
     _query_hourly_graph_data,
     _RANGE_OPTIONS,
@@ -142,64 +138,6 @@ class TestFormatRemaining:
         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")
@@ -215,27 +153,6 @@ class TestQueryDriveHealth:
         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
 # ---------------------------------------------------------------------------
@@ -397,10 +314,10 @@ class TestDenseScreen:
         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 "Boot:" in strip
-            assert "timer:" in strip
+            assert "Timer:" in strip
-            assert "last collect:" in strip
+            assert "Last collect:" in strip
-            assert "freshness:" in strip
+            assert "Freshness:" in strip
             # Maker credit now in titlebox only (issue #79)
             assert "by Bongbetic" not in strip
 
@@ -412,7 +329,7 @@ class TestDenseScreen:
         conn.close()
         app = FenrisTuiApp(store_path=tmp_path / "test.db")
 
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": True, "timer_active": True,
             "last_collect_ok": True, "last_collect_age_s": 60,
             "last_collect_reason": None,
@@ -427,7 +344,7 @@ class TestDenseScreen:
                 assert "monitoring: active in background · persists across reboots" in strip
                 assert "r resume — enable monitoring and future boots" in strip
                 assert "q quit" not in strip
-                assert rail == "q QUIT TUI"
+                assert rail == "q Quit TUI"
                 assert usage.region.y < service.region.y < quit_rail.region.y
                 assert usage.region.bottom <= service.region.y
                 assert service.region.bottom <= quit_rail.region.y
@@ -446,11 +363,11 @@ class TestDenseScreen:
         conn.close()
         app = FenrisTuiApp(store_path=db)
 
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": False, "timer_active": False,
             "last_collect_ok": None, "last_collect_age_s": None,
             "last_collect_reason": None,
-        }), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object(
+        }), patch("fenris.control.subprocess.run") as subprocess_run, patch.object(
             app, "_run_helper"
         ) as run_helper:
             async with app.run_test(size=(80, 24)) as pilot:
@@ -472,7 +389,7 @@ class TestDenseScreen:
                 visible_text = " ".join(
                     "".join(ElementTree.fromstring(screenshot).itertext()).split()
                 )
-                assert "paused time is excluded from your usage habit" in visible_text
+                assert "Paused time is excluded from your usage habit" in visible_text
                 assert "resume: fenris monitor resume" in visible_text
                 assert "monitoring: does not start on next boot" in str(
                     app.query_one("#service-strip").render()
@@ -487,7 +404,7 @@ class TestDenseScreen:
                         ElementTree.fromstring(app.export_screenshot()).itertext()
                     ).split()
                 )
-                assert "q QUIT TUI" in footer_text
+                assert "q Quit TUI" in footer_text
                 await pilot.press("q")
                 assert not app.is_running
             subprocess_run.assert_not_called()
@@ -509,11 +426,11 @@ class TestDenseScreen:
         conn.close()
         app = FenrisTuiApp(store_path=db)
 
-        with patch("fenris.tui.query_service_state", return_value={
+        with patch("fenris.status.query_service_state", return_value={
             "boot_enabled": True, "timer_active": True,
             "last_collect_ok": None, "last_collect_age_s": None,
             "last_collect_reason": None,
-        }), patch("fenris.tui.subprocess.run") as subprocess_run, patch.object(
+        }), patch("fenris.control.subprocess.run") as subprocess_run, patch.object(
             app, "_run_helper"
         ) as run_helper:
             async with app.run_test() as pilot:
@@ -539,7 +456,7 @@ class TestDenseScreen:
             store_path=tmp_path / "nonexistent.db",
             refresh_interval_s=0.2,
         )
-        auth_notice = "privileged actions will prompt for authentication (polkit)"
+        auth_notice = "Open with fenris (no sudo). Actions authenticate via polkit. ? Help"
 
         async with app.run_test() as pilot:
             headline = str(app.query_one("#headline-band").render())
@@ -595,12 +512,47 @@ class TestDisclosuresAndGreeting:
 # ---------------------------------------------------------------------------
 
 class TestFirstRun:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("size", [(80, 24), (60, 18)])
+    async def test_sudo_help_remains_available_after_launch(self, tmp_path, size):
+        app = FenrisTuiApp(store_path=tmp_path / "missing.db")
+        with patch.object(app, "_run_helper") as run_helper:
+            async with app.run_test(size=size) as pilot:
+                app.on_refresh_tick()
+                await pilot.press("?")
+                help_text = str(app.screen.query_one("#help-text").render())
+                assert "The dashboard does not need sudo" in help_text
+                assert "sudo fenris monitor resume" in help_text
+                assert "sudo fenris monitor pause" in help_text
+                assert "sudo fenris sample" in help_text
+                assert 'sudo usermod -aG fenris "$USER"' in help_text
+                assert "log out and back in" in help_text
+                await pilot.press("end")
+                visible = " ".join("".join(
+                    ElementTree.fromstring(app.export_screenshot()).itertext()
+                ).split())
+                assert "Esc Close" in visible
+                await pilot.press("escape")
+                assert len(app.screen_stack) == 1
+            run_helper.assert_not_called()
+
+    def test_missing_polkit_identifies_command_and_points_to_help(self, tmp_path):
+        app = FenrisTuiApp(store_path=tmp_path / "missing.db")
+        with patch("fenris.control.os.geteuid", return_value=1000), \
+             patch("fenris.control.subprocess.run", side_effect=FileNotFoundError(2, "Missing", "pkexec")), \
+             patch.object(app, "suspend"), patch.object(app, "_refresh"), \
+             patch.object(app, "notify") as notify:
+            app.action_resume()
+        message = notify.call_args.args[0]
+        assert "Command not found: pkexec" in message
+        assert "sudo /usr/libexec/fenris/fenris-monitor enable --now" in message
+
     def test_unprivileged_resume_uses_polkit(self, tmp_path):
         """TUI controls use the same authenticated path as the CLI."""
         app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
 
-        with patch("fenris.tui.os.geteuid", return_value=1000), \
+        with patch("fenris.control.os.geteuid", return_value=1000), \
-             patch("fenris.tui.subprocess.run") as run, \
+             patch("fenris.control.subprocess.run") as run, \
              patch.object(app, "suspend"), \
              patch.object(app, "_refresh"):
             run.return_value.returncode = 0
@@ -608,7 +560,6 @@ class TestFirstRun:
 
         run.assert_called_once_with(
             ["pkexec", "/usr/libexec/fenris/fenris-monitor", "enable", "--now"],
-            timeout=30,
         )
 
     @pytest.mark.asyncio
@@ -820,12 +771,29 @@ class TestQueryDailyGraphData:
         assert result[0]["is_gap"] is True
         conn.close()
 
+    def test_window_is_consecutive_utc_days_with_explicit_gaps(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        _insert_day(conn, "2026-09-14", bw=500)
+        result = _query_daily_graph_data(
+            conn, datetime(2026, 9, 16, 12, tzinfo=timezone.utc),
+        )
+        assert len(result) == 90
+        assert result[-3]["day"] == "2026-09-14"
+        assert result[-2]["day"] == "2026-09-15"
+        assert result[-2]["is_gap"] is True
+        assert result[-1]["day"] == "2026-09-16"
+        assert result[-1]["is_partial"] is True
+        conn.close()
+
 
 class TestQueryHourlyGraphData:
     def test_empty_day(self, tmp_path):
         conn = init_store(tmp_path / "test.db")
-        result = _query_hourly_graph_data(conn, "2026-09-20")
+        result = _query_hourly_graph_data(
-        assert result == []
+            conn, "2026-09-20", datetime(2026, 9, 21, tzinfo=timezone.utc),
+        )
+        assert len(result) == 24
+        assert all(item["is_gap"] for item in result)
         conn.close()
 
     def test_with_hours(self, tmp_path):
@@ -844,9 +812,25 @@ class TestQueryHourlyGraphData:
     def test_hour_labels(self, tmp_path):
         conn = init_store(tmp_path / "test.db")
         _insert_hour(conn, "2026-09-20T12:00:00+00:00", bw=1000)
-        result = _query_hourly_graph_data(conn, "2026-09-20")
+        result = _query_hourly_graph_data(
-        assert len(result) == 1
+            conn, "2026-09-20", datetime(2026, 9, 21, tzinfo=timezone.utc),
-        assert result[0]["local_label"] == "12"
+        )
+        assert len(result) == 24
+        assert result[12]["local_label"] == "12"
+        assert result[12]["bytes_written"] == 1000
+        conn.close()
+
+    def test_current_hour_is_partial_and_future_hours_are_not_gaps(self, tmp_path):
+        conn = init_store(tmp_path / "test.db")
+        result = _query_hourly_graph_data(
+            conn, "2026-09-20",
+            datetime(2026, 9, 20, 14, 30, tzinfo=timezone.utc),
+        )
+        assert result[13]["is_gap"] is True
+        assert result[14]["is_partial"] is True
+        assert result[14]["is_gap"] is True
+        assert result[15]["is_future"] is True
+        assert result[15]["is_gap"] is False
         conn.close()
 
 
@@ -897,6 +881,17 @@ class TestDailyBarGraph:
         graph._trim_to_range()
         assert len(graph._day_data) == 7
 
+    def test_scale_includes_allocated_and_unallocated_writes(self):
+        graph = DailyBarGraph()
+        graph._all_day_data = [{
+            "day": "2026-09-16",
+            "total_bytes": 800,
+            "allocated_bytes": 500,
+            "unallocated_bytes": 300,
+        }]
+        graph._trim_to_range()
+        assert graph._max_bytes == 800
+
 
 # ---------------------------------------------------------------------------
 # Bar graph headless TUI tests (issue #75)
@@ -924,8 +919,15 @@ class TestBarGraphTUI:
             assert len(graph._day_data) > 0
             # Legend should be visible
             legend = str(app.query_one("#bar-legend").render())
-            assert "alloc" in legend
+            assert "Alloc" in legend
-            assert "zero" in legend
+            assert "Zero" in legend
+            range_label = str(app.query_one("#bar-range").render())
+            assert "14 days" in range_label
+            assert "UTC" in range_label
+            plotted = str(app.query_one("#bar-render").render())
+            assert "Writes (" in plotted
+            assert "Day UTC" in plotted
+            assert "→" in plotted
 
     @pytest.mark.asyncio
     async def test_arrow_selection(self, tmp_path):
@@ -946,20 +948,16 @@ class TestBarGraphTUI:
             await pilot.pause()
             await pilot.pause()
 
-            # Right arrow selects first bar
+            # The newest day is selected by default; left moves backward.
-            await pilot.press("right")
+            initial = graph.selected_index
-            await pilot.pause()
-            assert graph.selected_index == 0
-
-            # Right again moves to second
-            await pilot.press("right")
-            await pilot.pause()
-            assert graph.selected_index == 1
-
-            # Left moves back
             await pilot.press("left")
             await pilot.pause()
-            assert graph.selected_index == 0
+            assert graph.selected_index == initial - 1
+
+            # Right returns to the newest day.
+            await pilot.press("right")
+            await pilot.pause()
+            assert graph.selected_index == initial
 
     @pytest.mark.asyncio
     async def test_range_switching(self, tmp_path):
@@ -990,17 +988,17 @@ class TestBarGraphTUI:
             assert graph.range_days == 7
             assert len(graph._day_data) == 7
 
-            # Press 3 for 28-day range
+            # Press 3 for 30-day range
             await pilot.press("3")
             await pilot.pause()
-            assert graph.range_days == 28
+            assert graph.range_days == 30
-            assert len(graph._day_data) == 28
+            assert len(graph._day_data) == 30
 
-            # Press 4 for 90-day range (only 30 days available)
+            # Exact windows retain gap slots instead of compressing time.
             await pilot.press("4")
             await pilot.pause()
             assert graph.range_days == 90
-            assert len(graph._day_data) == 30
+            assert len(graph._day_data) == 90
 
     @pytest.mark.asyncio
     async def test_readout_updates_on_selection(self, tmp_path):
@@ -1021,12 +1019,13 @@ class TestBarGraphTUI:
             await pilot.pause()
             await pilot.pause()
 
-            # No selection initially
+            # Newest day is selected initially.
             readout = str(app.query_one("#bar-readout").render())
-            assert "select" in readout.lower()
+            assert "UTC" in readout
+            assert "GB" in readout
 
-            # Select first bar
+            # Moving left updates the selected-day readout.
-            await pilot.press("right")
+            await pilot.press("left")
             await pilot.pause()
             readout = str(app.query_one("#bar-readout").render())
             assert "2026-09" in readout
@@ -1056,10 +1055,8 @@ class TestBarGraphTUI:
             await pilot.pause()
             await pilot.pause()
 
-            # Select a day
+            # The newest day is selected automatically.
-            await pilot.press("right")
+            assert graph.selected_index == len(graph._day_data) - 1
-            await pilot.pause()
-            assert graph.selected_index == 0
             assert graph.view_mode == "daily"
 
             # Enter drill-down
@@ -1069,6 +1066,14 @@ class TestBarGraphTUI:
             assert graph.drill_day is not None
             assert len(graph._hour_data) == 24
 
+            # Five-minute data refreshes retain the selected hour.
+            graph._hourly_selected = 12
+            selected_hour = graph._hour_data[12]["hour"]
+            app._refresh()
+            await pilot.pause()
+            assert graph.view_mode == "hourly"
+            assert graph._hour_data[graph._hourly_selected]["hour"] == selected_hour
+
             # Esc returns to daily
             await pilot.press("escape")
             await pilot.pause()
@@ -1091,7 +1096,7 @@ class TestBarGraphTUI:
         app = FenrisTuiApp(store_path=tmp_path / "nonexistent.db")
         async with app.run_test(size=(80, 24)) as pilot:
             graph = app.query_one("#usage-history")
-            assert graph.border_title == "usage history"
+            assert graph.border_title == "Usage history"
 
     @pytest.mark.asyncio
     async def test_glyphs_in_legend(self, tmp_path):
@@ -1172,7 +1177,7 @@ class TestConstrainedLayout:
             # Constrained summary should be visible
             summary = app.query_one("#constrained-summary")
             summary_text = str(summary.render())
-            assert "graph needs ≥80×24" in summary_text
+            assert "Graph needs ≥80×24" in summary_text
             assert "days" in summary_text
             assert "GB total" in summary_text
 
@@ -1196,7 +1201,7 @@ class TestConstrainedLayout:
             # Constrained summary should be visible
             summary = app.query_one("#constrained-summary")
             summary_text = str(summary.render())
-            assert "graph needs ≥80×24" in summary_text
+            assert "Graph needs ≥80×24" in summary_text
 
     @pytest.mark.asyncio
     async def test_resize_from_constrained_to_normal(self, tmp_path):
@@ -1257,9 +1262,7 @@ class TestConstrainedLayout:
             # Focus and select a day
             app.set_focus(graph)
             await pilot.pause()
-            await pilot.press("right")
+            assert graph.selected_index == len(graph._day_data) - 1
-            await pilot.pause()
-            assert graph.selected_index == 0
 
             # Resize to constrained — pilot.resize_terminal changes terminal size
             await pilot.resize_terminal(79, 24)
@@ -1273,7 +1276,7 @@ class TestConstrainedLayout:
             # Constrained summary should show selected day context
             summary = app.query_one("#constrained-summary")
             summary_text = str(summary.render())
-            assert "graph needs ≥80×24" in summary_text
+            assert "Graph needs ≥80×24" in summary_text
             # Selected day context should survive
             assert "2026-09" in summary_text
 
@@ -1285,7 +1288,7 @@ class TestConstrainedLayout:
             assert app._is_constrained_mode
             summary = app.query_one("#constrained-summary")
             summary_text = str(summary.render())
-            assert "graph needs ≥80×24" in summary_text
+            assert "Graph needs ≥80×24" in summary_text
             assert "awaiting" in summary_text.lower()
 
     @pytest.mark.asyncio
@@ -1315,13 +1318,13 @@ class TestConstrainedLayout:
 
             # Service strip should have actions
             strip = str(app.query_one("#service-strip").render())
-            assert "p pause" in strip
+            assert "p Pause" in strip
-            assert "r resume" in strip
+            assert "r Resume" in strip
             assert "q quit" not in strip  # Quit is in quit-rail
 
             # Quit rail should be visible
             rail = str(app.query_one("#quit-rail").render())
-            assert "QUIT" in rail
+            assert "Quit" in rail
 
     @pytest.mark.asyncio
     async def test_constrained_long_reasons_visible(self, tmp_path):