feat(tui): clarify monitoring continuity and quitting

This commit is contained in:
xavierk
2026-09-10 19:43:32 +05:30
parent 7bbe5cede7
commit d01df6468f
6 changed files with 462 additions and 18 deletions
+2
View File
@@ -193,6 +193,8 @@ No interval key exists in `/etc/fenris/fenris.conf`. Cadence is a systemd concer
`fenris` opens the TUI dashboard.
- **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.
## Retired menu options
+54 -1
View File
@@ -449,7 +449,7 @@ def _format_headline(proj) -> str:
def _append_service_facts(lines: List[str], service: Dict[str, Any]) -> None:
"""Append the four separate service facts (§7.3, LC-9)."""
"""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"
@@ -473,6 +473,55 @@ def _append_service_facts(lines: List[str], service: Dict[str, Any]) -> None:
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)
# ---------------------------------------------------------------------------
_CONTINUITY_ACTIVE = "monitoring: active in background · persists across reboots"
_CONTINUITY_DISABLED = "monitoring: does not start on next boot"
_PAUSED_TITLE = "monitoring: paused — deliberate disable"
_PAUSED_CONSEQUENCE = (
"paused time is excluded from your usage habit · resume: fenris monitor resume"
)
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
def deliberate_pause_lines() -> List[str]:
"""Return the exact CLI/TUI presentation for a sanctioned pause."""
return [_PAUSED_TITLE, _PAUSED_CONSEQUENCE]
def is_deliberately_paused(conn: sqlite3.Connection, service: Dict[str, Any]) -> bool:
"""Whether the latest closed period was ended by Fenris's own pause path.
Raw systemd operations have no `user_disabled` row, so they must never be
presented as a Deliberate disable. A live enabled timer also wins over a
stale period marker, keeping the presentation consistent with service facts.
"""
if service.get("boot_enabled") or service.get("timer_active"):
return False
open_period = conn.execute(
"SELECT 1 FROM monitoring_periods WHERE ended_at IS NULL LIMIT 1"
).fetchone()
if open_period is not None:
return False
row = conn.execute(
"SELECT end_cause FROM monitoring_periods "
"WHERE ended_at IS NOT NULL "
"ORDER BY ended_at DESC, id DESC LIMIT 1"
).fetchone()
return row is not None and row[0] == "user_disabled"
def format_disclosures() -> str:
@@ -562,6 +611,10 @@ def get_status(store_path: Optional[Path] = None, clock_now: Optional[datetime]
service["freshness"] = freshness
service["freshness_age_s"] = freshness_age_s
try:
service["deliberately_paused"] = is_deliberately_paused(conn, service)
except sqlite3.Error:
service["deliberately_paused"] = False
# --- Drive anomalies (§9.7, FL-7) ---
drive_facts = []
+68 -12
View File
@@ -23,7 +23,7 @@ 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.containers import Container, Horizontal, VerticalScroll
from textual.screen import ModalScreen
from textual.widgets import Static
@@ -43,6 +43,9 @@ from .status import (
format_disclosures,
freshness_age_human,
grade_freshness,
deliberate_pause_lines,
is_deliberately_paused,
monitoring_continuity,
open_store_readonly,
query_service_state,
read_config,
@@ -229,6 +232,7 @@ def _query_service_facts(conn: sqlite3.Connection, clock_now: datetime) -> Dict[
"freshness": freshness,
"freshness_age_s": None,
"period": period_info,
"deliberately_paused": is_deliberately_paused(conn, svc),
**svc,
}
@@ -292,14 +296,32 @@ class FenrisTuiApp(App):
CSS = """
#main-grid {
layout: grid;
grid-size: 2 3;
grid-size: 2 4;
grid-columns: 3fr 2fr;
grid-rows: 8 1fr 7;
height: 1fr;
grid-rows: 8 10 7 3;
height: auto;
}
#main-grid.paused {
grid-size: 2 5;
grid-rows: 8 5 10 7 3;
}
#dashboard-scroll { height: 1fr; }
#headline-band { column-span: 2; }
#service-strip { column-span: 2; }
.pane { border: round #555555; padding: 0 1; }
#paused-banner {
column-span: 2;
display: none;
background: $error 20%;
color: $text;
height: 100%;
}
#service-strip { column-span: 2; height: 100%; }
#quit-rail {
column-span: 2;
border: heavy $accent;
content-align: center middle;
height: 100%;
}
.pane { border: round #555555; padding: 0 1; height: 100%; }
#confirm-text { padding: 1 2; }
#disc-text { padding: 1 2; }
"""
@@ -330,11 +352,14 @@ class FenrisTuiApp(App):
self._clock_now = datetime.now(timezone.utc)
def compose(self) -> ComposeResult:
with Vertical(id="main-grid"):
with VerticalScroll(id="dashboard-scroll"):
with Container(id="main-grid"):
yield Static("", id="headline-band", classes="pane")
yield Static("", id="paused-banner")
yield Static("", id="usage-history", classes="pane")
yield Static("", id="drive-health", classes="pane")
yield Static("", id="service-strip", classes="pane")
yield Static("q QUIT TUI", id="quit-rail")
def on_mount(self) -> None:
"""Set border titles and render initial state."""
@@ -391,6 +416,7 @@ class FenrisTuiApp(App):
def _render_empty_or_fault(self) -> None:
"""Render empty store greeting or store fault."""
self._hide_paused_banner()
if not self.store_path.exists():
# Empty store — greeting with enable hint (IN-3)
self._render_headline(
@@ -402,7 +428,9 @@ class FenrisTuiApp(App):
self.query_one("#service-strip").update(
"boot: disabled · timer: inactive · last collect: unknown · freshness: empty · "
"[dim]by Bongbetic[/dim]\n"
"p pause · r resume · c collect · d disclosures · q quit"
"[bold]CONTINUITY[/bold] %s\n"
"p pause · r resume · c collect · d disclosures"
% monitoring_continuity({"boot_enabled": False})
)
else:
# Store fault (FL-4)
@@ -414,7 +442,7 @@ class FenrisTuiApp(App):
self.query_one("#drive-health").update("")
self.query_one("#service-strip").update(
"[dim]by Bongbetic[/dim]\n"
"p pause · r resume · c collect · d disclosures · q quit"
"p pause · r resume · c collect · d disclosures"
)
def _render_all_regions(self, conn: sqlite3.Connection) -> None:
@@ -473,17 +501,45 @@ class FenrisTuiApp(App):
self.query_one("#service-strip").update(
"boot: %s · timer: %s · last collect: %s · freshness: %s · "
"[dim]by Bongbetic[/dim]\n"
"[bold]CONTINUITY[/bold] %s\n"
"%s\n"
"p pause · r resume · c collect · d disclosures · q quit"
% (boot, activity, collect, freshness, svc.get("period", ""))
"p pause · r resume · c collect · d disclosures"
% (
boot, activity, collect, freshness,
monitoring_continuity(svc), svc.get("period", ""),
)
)
self._render_paused_banner(svc)
except Exception:
self._hide_paused_banner()
self.query_one("#service-strip").update(
"boot: unknown · timer: unknown · last collect: unknown · freshness: unknown · "
"[dim]by Bongbetic[/dim]\n"
"p pause · r resume · c collect · d disclosures · q quit"
"p pause · r resume · c collect · d disclosures"
)
def _render_paused_banner(self, service: Dict[str, Any]) -> None:
"""Show the high-contrast Deliberate disable block only when sanctioned."""
banner = self.query_one("#paused-banner")
if service.get("deliberately_paused"):
banner.update(
"[bold black on red]%s[/bold black on red]\n%s"
% tuple(deliberate_pause_lines())
)
banner.styles.display = "block"
main_grid = self.query_one("#main-grid")
main_grid.add_class("paused")
main_grid.refresh(layout=True)
else:
self._hide_paused_banner()
def _hide_paused_banner(self) -> None:
"""Ensure an unavailable store cannot retain a stale paused presentation."""
self.query_one("#paused-banner").styles.display = "none"
main_grid = self.query_one("#main-grid")
main_grid.remove_class("paused")
main_grid.refresh(layout=True)
def _format_headline(self, proj: ProjectionResult) -> str:
"""Format the lifespan headline (spec §6.11)."""
if proj.headline_remaining_seconds is None:
+83
View File
@@ -399,6 +399,89 @@ class TestCI2Parity:
assert "last collect:" in status
assert "freshness:" in status
def test_dashboard_clarity_parity_strings_have_one_status_source(self):
"""DC-2/DC-3 wording originates in status and the TUI imports it."""
status_src = (FENRIS_PKG / "status.py").read_text()
tui_src = (FENRIS_PKG / "tui.py").read_text()
for wording in (
"monitoring: active in background · persists across reboots",
"monitoring: does not start on next boot",
"monitoring: paused — deliberate disable",
"paused time is excluded from your usage habit · resume: fenris monitor resume",
):
assert status_src.count(wording) == 1
assert wording not in tui_src
@pytest.mark.asyncio
@pytest.mark.parametrize(
("state", "service", "expected_lines"),
[
(
"active_enabled",
{"boot_enabled": True, "timer_active": True},
["monitoring: active in background · persists across reboots"],
),
(
"boot_disabled",
{"boot_enabled": False, "timer_active": False},
["monitoring: does not start on next boot"],
),
(
"deliberately_paused",
{"boot_enabled": False, "timer_active": False},
[
"monitoring: does not start on next boot",
"monitoring: paused — deliberate disable",
"paused time is excluded from your usage habit · resume: fenris monitor resume",
],
),
],
)
async def test_dashboard_clarity_monitoring_lines_match_both_views(
self, tmp_path, state, service, expected_lines
):
"""CI-2 synthetic-store sweep covers active, disabled, and paused states."""
db = tmp_path / (state + ".db")
conn = init_store(db)
if state == "active_enabled":
ensure_period_open(conn, _clock())
elif state == "deliberately_paused":
conn.execute(
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
"VALUES (?, ?, ?)",
("2026-09-30T09:00:00+00:00", "2026-09-30T10:00:00+00:00", "user_disabled"),
)
conn.commit()
conn.close()
service_state = {
**service,
"last_collect_ok": None,
"last_collect_age_s": None,
"last_collect_reason": None,
}
with patch("fenris.status.query_service_state", return_value=service_state), patch(
"fenris.tui.query_service_state", return_value=service_state
):
status = get_status(
store_path=db, clock_now=_clock(), query_services=True, query_journal=False
).lower()
app = FenrisTuiApp(store_path=db)
async with app.run_test(size=(100, 40)):
tui_text = "\n".join(
(
str(app.query_one("#service-strip").render()),
str(app.query_one("#paused-banner").render()),
)
).lower()
for expected in expected_lines:
assert expected in status
assert expected in tui_text
if state != "deliberately_paused":
assert "monitoring: paused — deliberate disable" not in status
assert "monitoring: paused — deliberate disable" not in tui_text
def test_pause_resume_action_names(self):
tui_keys = {b.key for b in FenrisTuiApp.BINDINGS}
assert "p" in tui_keys
+125
View File
@@ -415,6 +415,131 @@ class TestServiceFacts:
assert "last collect:" in result
assert "freshness:" in result
def test_continuity_reports_boot_enabled_independently_of_runtime(self, tmp_path):
"""Status names reboot continuity while retaining the timer fact."""
from fenris.status import get_status
db = tmp_path / "observations.db"
init_store(db)
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
with patch("fenris.status.query_service_state", return_value={
"boot_enabled": True, "timer_active": False,
"last_collect_ok": None, "last_collect_age_s": None,
"last_collect_reason": None,
}):
result = get_status(store_path=db, clock_now=now,
query_services=True, query_journal=False)
assert "monitoring: active in background · persists across reboots" in result
assert "timer: inactive" in result
def test_continuity_and_deliberate_pause_are_reported_separately(self, tmp_path):
"""Only a sanctioned user_disabled period renders the paused wording."""
from fenris.status import get_status
db = tmp_path / "observations.db"
conn = init_store(db)
conn.execute(
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
"VALUES (?, ?, ?)",
("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "user_disabled"),
)
conn.commit()
conn.close()
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
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,
}):
result = get_status(store_path=db, clock_now=now,
query_services=True, query_journal=False)
assert "monitoring: does not start on next boot" in result
assert "monitoring: paused — deliberate disable" in result
assert "paused time is excluded from your usage habit · resume: fenris monitor resume" in result
def test_raw_system_state_without_user_disabled_is_not_a_deliberate_pause(self, tmp_path):
"""A non-sanctioned stop never acquires the deliberate-disable label."""
from fenris.status import get_status
db = tmp_path / "observations.db"
conn = init_store(db)
conn.execute(
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
"VALUES (?, ?, ?)",
("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "migrated"),
)
conn.commit()
conn.close()
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
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,
}):
result = get_status(store_path=db, clock_now=now,
query_services=True, query_journal=False)
assert "monitoring: paused — deliberate disable" not in result
def test_resumed_open_period_clears_a_previous_deliberate_pause(self, tmp_path):
"""A later sanctioned resume takes precedence over an older pause."""
from fenris.status import get_status
db = tmp_path / "observations.db"
conn = init_store(db)
conn.execute(
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
"VALUES (?, ?, ?)",
("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "user_disabled"),
)
conn.execute(
"INSERT INTO monitoring_periods (started_at) VALUES (?)",
("2026-09-01T11:00:00+00:00",),
)
conn.commit()
conn.close()
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
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,
}):
result = get_status(store_path=db, clock_now=now,
query_services=True, query_journal=False)
assert "monitoring: paused — deliberate disable" not in result
def test_live_enabled_service_suppresses_a_stale_pause_marker(self, tmp_path):
"""A raw re-enable cannot leave a contradictory paused presentation."""
from fenris.status import get_status
db = tmp_path / "observations.db"
conn = init_store(db)
conn.execute(
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
"VALUES (?, ?, ?)",
("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "user_disabled"),
)
conn.commit()
conn.close()
now = datetime(2026, 9, 1, 12, 0, 0, tzinfo=timezone.utc)
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,
}):
result = get_status(store_path=db, clock_now=now,
query_services=True, query_journal=False)
assert "monitoring: paused — deliberate disable" not in result
# ---------------------------------------------------------------------------
# Status output structure (§8.8, LC-9)
+126 -1
View File
@@ -10,6 +10,7 @@ Covers:
Criteria: TUI-1, TUI-4, CI-1, CI-4, IN-3.
"""
import sqlite3
from xml.etree import ElementTree
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock
@@ -386,6 +387,131 @@ class TestDenseScreen:
assert "freshness:" in strip
assert "by Bongbetic" in strip
@pytest.mark.asyncio
async def test_service_strip_shows_continuity_and_separate_quit_rail(self, tmp_path):
"""The visible action footer excludes quit because the rail owns it."""
conn = init_store(tmp_path / "test.db")
_open_period(conn)
conn.close()
app = FenrisTuiApp(store_path=tmp_path / "test.db")
with patch("fenris.tui.query_service_state", return_value={
"boot_enabled": True, "timer_active": True,
"last_collect_ok": True, "last_collect_age_s": 60,
"last_collect_reason": None,
}):
async with app.run_test():
strip = str(app.query_one("#service-strip").render()).lower()
rail = str(app.query_one("#quit-rail").render())
usage = app.query_one("#usage-history")
service = app.query_one("#service-strip")
quit_rail = app.query_one("#quit-rail")
assert "continuity" in strip
assert "monitoring: active in background · persists across reboots" in strip
assert "p pause · r resume · c collect · d disclosures" in strip
assert "q quit" not in strip
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
@pytest.mark.asyncio
async def test_deliberate_pause_banner_is_visible_and_quit_preserves_periods(self, tmp_path):
"""The Pilot sees the paused block; q leaves persisted monitoring state alone."""
db = tmp_path / "test.db"
conn = init_store(db)
conn.execute(
"INSERT INTO monitoring_periods (started_at, ended_at, end_cause) "
"VALUES (?, ?, ?)",
("2026-09-01T09:00:00+00:00", "2026-09-01T10:00:00+00:00", "user_disabled"),
)
conn.commit()
conn.close()
app = FenrisTuiApp(store_path=db)
with patch("fenris.tui.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(
app, "_run_helper"
) as run_helper:
async with app.run_test(size=(80, 24)) as pilot:
await pilot.pause()
paused_banner = app.query_one("#paused-banner")
main_grid = app.query_one("#main-grid")
assert str(main_grid.styles.layout) == "<grid>"
assert main_grid.has_class("paused")
assert len(main_grid.styles.grid_rows) == 5
banner = str(paused_banner.render()).lower()
assert "monitoring: paused — deliberate disable" in banner
assert "paused time is excluded from your usage habit · resume: fenris monitor resume" in banner
assert paused_banner.region.height >= 5
assert paused_banner.region.y < app.query_one("#usage-history").region.y
assert app.query_one("#usage-history").region.bottom <= app.query_one(
"#service-strip"
).region.y
screenshot = app.export_screenshot()
visible_text = " ".join(
"".join(ElementTree.fromstring(screenshot).itertext()).split()
)
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()
).lower()
dashboard_scroll = app.query_one("#dashboard-scroll")
assert dashboard_scroll.max_scroll_y > 0
dashboard_scroll.focus()
await pilot.press("end")
assert dashboard_scroll.scroll_y == dashboard_scroll.max_scroll_y
footer_text = " ".join(
"".join(
ElementTree.fromstring(app.export_screenshot()).itertext()
).split()
)
assert "q QUIT TUI" in footer_text
await pilot.press("q")
assert not app.is_running
subprocess_run.assert_not_called()
run_helper.assert_not_called()
conn = sqlite3.connect(db)
row = conn.execute(
"SELECT ended_at, end_cause FROM monitoring_periods"
).fetchone()
conn.close()
assert row == ("2026-09-01T10:00:00+00:00", "user_disabled")
@pytest.mark.asyncio
async def test_quit_preserves_an_active_monitoring_period(self, tmp_path):
"""Quitting an active dashboard never closes or mutates its period."""
db = tmp_path / "test.db"
conn = init_store(db)
_open_period(conn, "2026-09-01T09:00:00+00:00")
conn.close()
app = FenrisTuiApp(store_path=db)
with patch("fenris.tui.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(
app, "_run_helper"
) as run_helper:
async with app.run_test() as pilot:
await pilot.press("q")
assert not app.is_running
subprocess_run.assert_not_called()
run_helper.assert_not_called()
conn = sqlite3.connect(db)
row = conn.execute(
"SELECT started_at, ended_at, end_cause FROM monitoring_periods"
).fetchone()
conn.close()
assert row == ("2026-09-01T09:00:00+00:00", None, None)
@pytest.mark.asyncio
async def test_branding_and_one_time_auth_banner(self, tmp_path):
"""Identity is visible at launch; auth notice clears once per session."""
@@ -553,4 +679,3 @@ class TestStateMatrixCombinations:
# Incomplete provenance → UNVERIFIED tier
assert proj.baseline_tier.value == "unverified_override"
conn.close()