Compare commits

...
13 Commits
Author SHA1 Message Date
xavierk 714e69be52 Expose XBPS tools during CI setup
Release / release (push) Successful in 1m11s
2026-09-16 08:18:49 +05:30
xavierk ba16413363 Provision XBPS tools in release CI
Release / release (push) Failing after 40s
2026-09-16 08:17:31 +05:30
xavierk dfe6a6a2d0 Fix 0.3.6 release changelog
Release / release (push) Failing after 53s
2026-09-16 08:13:14 +05:30
xavierk 44c57b70dd Release Fenris 0.3.6
Release / release (push) Failing after 6s
2026-09-16 08:05:28 +05:30
xavierk f06424f3b8 Merge remote-tracking branch 'origin/main'
# Conflicts:
#	README.md
#	scripts/xbps-publish.sh
2026-09-15 21:19:33 +05:30
xavierk fae72bb07b Document XBPS cache-bypass refresh 2026-09-15 19:53:50 +05:30
xavierk 9d22525403 Fix XBPS repository verification 2026-09-15 19:48:08 +05:30
xavierk a3e6cc3b3c Fix Void package acceptance gaps 2026-09-15 19:43:00 +05:30
xavierk 95c75badd5 Route TUI controls through polkit 2026-09-15 15:44:10 +05:30
xavierk 70dbae65fb Clarify TUI monitoring activation 2026-09-15 15:40:29 +05:30
xavierk d119a09b1f Fix Void package lifecycle on host 2026-09-15 15:30:51 +05:30
xavierk fca0724fb4 docs: add Void installation and operations guide 2026-09-15 11:00:02 +05:30
xavierk 1c3037c2f8 Implement independent format gates for release workflow (issue #86)
Add XBPS build and sign steps to CI workflow
Add XBPS publication as independent gate (requires manual trigger)
Track format availability in release notes
Update release-footer.md with XBPS install instructions
Add --available/--withheld arguments to extract_changelog.py
Attach XBPS artifacts to Gitea release
Clean up XBPS signing key material after use
2026-09-15 10:50:38 +05:30
36 changed files with 904 additions and 839 deletions
+14
View File
@@ -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
+5
View File
@@ -1,5 +1,10 @@
## Agent skills
## Commit messages
Do not add `Co-authored-by: CommandCodeBot <noreply@commandcode.ai>` or other
CommandCodeBot attribution trailers to commits.
### Issue tracker
Issues are tracked in Gitea using the authenticated `tea` CLI. See `docs/agents/issue-tracker.md`.
+14
View File
@@ -9,6 +9,20 @@ backfill releases from before this changelog.
## [Unreleased]
## [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
+21
View File
@@ -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.
+12 -1
View File
@@ -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)
@@ -95,6 +95,7 @@ install: check-python check-smartctl dist/fenris-*.whl
@echo "=== Installing runit service files (dormant — not enabled) ==="
@sudo install -d -m 0755 /etc/sv/fenris-collect/log
@sudo install -d -o root -g fenris -m 2770 /var/log/fenris-collect
@sudo install -m 0755 units/runit/fenris-collect/run /etc/sv/fenris-collect/run
@sudo install -m 0755 units/runit/fenris-collect/log/run /etc/sv/fenris-collect/log/run
@sudo touch /etc/sv/fenris-collect/down
@@ -111,10 +112,14 @@ install: check-python check-smartctl dist/fenris-*.whl
@echo "$(LIBEXEC_DIR)/fenris-collect" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(UNIT_DIR)/fenris-collect.timer" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(UNIT_DIR)/fenris-collect.service" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/etc/sv/fenris-collect/run" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/etc/sv/fenris-collect/log/run" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/etc/sv/fenris-collect/down" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(POLKIT_DIR)/com.bongbetic.fenris.monitor.policy" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(VENV_DIR)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(DATA_DIR)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(CONF_DIR)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/var/log/fenris-collect" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(MANIFEST)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "=== Install complete ==="
@@ -150,6 +155,8 @@ upgrade: dist/fenris-*.whl
@sudo install -m 0644 units/fenris-collect.timer $(UNIT_DIR)/
@sudo install -m 0644 units/fenris-collect.service $(UNIT_DIR)/
@sudo install -d -m 0755 /etc/sv/fenris-collect/log
@sudo groupadd -f fenris
@sudo install -d -o root -g fenris -m 2770 /var/log/fenris-collect
@sudo install -m 0755 units/runit/fenris-collect/run /etc/sv/fenris-collect/run
@sudo install -m 0755 units/runit/fenris-collect/log/run /etc/sv/fenris-collect/log/run
@sudo install -m 0644 polkit/com.bongbetic.fenris.monitor.policy $(POLKIT_DIR)/
@@ -167,10 +174,14 @@ upgrade: dist/fenris-*.whl
@echo "$(LIBEXEC_DIR)/fenris-collect" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(UNIT_DIR)/fenris-collect.timer" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(UNIT_DIR)/fenris-collect.service" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/etc/sv/fenris-collect/run" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/etc/sv/fenris-collect/log/run" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/etc/sv/fenris-collect/down" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(POLKIT_DIR)/com.bongbetic.fenris.monitor.policy" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(VENV_DIR)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(DATA_DIR)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(CONF_DIR)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "/var/log/fenris-collect" | sudo tee -a $(MANIFEST) > /dev/null
@echo "$(MANIFEST)" | sudo tee -a $(MANIFEST) > /dev/null
@echo "=== Restarting timer only if contents changed and active (IN-5) ==="
+32 -8
View File
@@ -68,16 +68,15 @@ TLS).
### Void Linux (XBPS)
Void x86_64 with glibc and runit is the native target. Its XBPS channel is
withheld until the host-acceptance gate passes; do not install proof artifacts
from it. Once a Fenris Release lists XBPS as available, add the permanent
signed repository, refresh its metadata, and install the package:
Void x86_64 with glibc and runit is the native target. Its signed XBPS channel
is available from the permanent repository below. Add it, refresh its
metadata, and install the released package:
```bash
sudo install -d -m 0755 /etc/xbps.d
echo 'repository=https://git.bongbetic.com/xavierk/Fenris-xbps/raw/branch/stable/x86_64' \
| sudo tee /etc/xbps.d/fenris.conf
sudo xbps-install -S fenris
sudo xbps-install -M -S fenris
```
XBPS requires remote repositories to be signed. On the first refresh it
@@ -89,9 +88,12 @@ available at
For later updates, always refresh first so XBPS fetches the current index:
```bash
sudo xbps-install -Syu
sudo xbps-install -M -Syu
```
The `-M` flag bypasses XBPS's on-disk repodata cache. It is required when
checking for a newly published package through Gitea's cached raw-file URL.
The runit service remains dormant after installation. `fenris monitor resume`
creates `/var/service/fenris-collect`; pause removes that link and records a
deliberate disable in the observation history.
@@ -250,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.
+4 -4
View File
@@ -1,14 +1,14 @@
# 8. Native Void Linux support and XBPS delivery
Status: Accepted scope and hosting direction; implementation and acceptance proof pending.
Status: Accepted — implementation and host acceptance completed; evidence is recorded in [issue #87](https://git.bongbetic.com/xavierk/Fenris/issues/87).
Fenris will support Void Linux natively with runit and full application feature parity, while retaining its existing Debian/RPM and systemd support. This extends the platform boundary in [ADR 0003](0003-service-lifecycle-and-sanctioned-toggle.md) and the delivery scope in [ADR 0007](0007-package-delivery-amends-0004.md): requiring Void users to replace their init system would not meet the native-support goal.
Delivery will include a Fenris-maintained, signed XBPS repository that users configure once for subsequent installation and updates through XBPS, plus versioned release artifacts and notes on Gitea. All downloads must be served directly by Gitea itself; a separate static HTTP repository, even alongside Gitea, does not satisfy this requirement.
Use a dedicated public Gitea repository, provisionally `xavierk/Fenris-xbps`, with a permanent `stable` branch. Its raw-file URL serves the XBPS index, versioned packages, and package signatures as ordinary Git blobs without LFS. Keeping binaries in a separate repository avoids increasing application source-clone size. This accepts growth in distribution-repository Git history in exchange for publishing index and artifacts together through one branch update, without the generic registry's delete-and-upload index replacement gap. The proposed repository has not yet been created.
Use the dedicated public Gitea repository `xavierk/Fenris-xbps` with its permanent `stable` branch. Its raw-file URL serves the XBPS index, versioned packages, and package signatures as ordinary Git blobs without LFS. Keeping binaries in a separate repository avoids increasing application source-clone size. This accepts growth in distribution-repository Git history in exchange for publishing index and artifacts together through one branch update, without the generic registry's delete-and-upload index replacement gap.
Serialize publication, commit the signed index and its new artifacts together, and retain older versioned artifacts in the current tree so clients with cached older indexes can still download them. Native XBPS installation and update tests against the actual endpoint are required before release validation. Verify binary delivery limits and index freshness: the inspected Gitea raw endpoint advertised six-hour HTTP caching, so immediate update visibility has not been established.
Serialize publication, commit the signed index and its new artifacts together, and retain older versioned artifacts in the current tree so clients with cached older indexes can still download them. Native XBPS installation and update tests against the actual endpoint are required before release validation. The first release acceptance recorded in issue #87 verified direct artifact delivery, signed metadata, retained packages, and immediate discovery after an explicit memory-synchronized refresh. The Gitea raw endpoint advertises six-hour HTTP caching; users should use `xbps-install -M -S` when looking for updates so XBPS bypasses its on-disk repodata cache.
Immediate availability is required: after successful XBPS publication, an explicit repository refresh against the permanent URL must discover the newly published version without a cache-expiry wait or a URL change. This does not promise automatic installation on client machines. Acceptance must exercise a client that fetched the previous index before publication and verify that refresh retrieves the new index and its signed package afterward. Resolve and document actual client and intermediary cache behavior; if the selected Gitea route cannot meet this requirement, hold XBPS publication and revisit its delivery mechanics rather than silently accepting delayed availability.
@@ -18,4 +18,4 @@ The first supported Void target is x86_64 with glibc, matching the inspected dev
Package formats have independent publication gates: publish each validated format, and hold only formats that have not passed release validation. A failure or pending validation in XBPS must not prevent a validated Debian or RPM package from shipping, and vice versa. Release notes must identify available formats and those still withheld; publication must not imply validation of a missing format.
After successful native acceptance testing, leave the released XBPS package installed on this machine and monitoring the selected NVMe drive. Preserve the observation history collected during testing. Coordinate the reboot test with the user so it can occur at a suitable interruption point.
After successful native acceptance testing, leave the released XBPS package installed on this machine and monitoring the selected NVMe drive. Preserve the observation history collected during testing. Coordinate the reboot test with the user so it can occur at a suitable interruption point. Issue #87 records that this final state, including reboot persistence, was achieved for the first supported release.
+8
View File
@@ -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.
+1 -1
View File
@@ -63,4 +63,4 @@ Musl, other architectures, additional init systems, official Void repository inc
## Further Notes
The design decisions are recorded in ADR 0008. Hosting feasibility is supported by Gitea routing/source inspection and an existing raw-file request, but the dedicated distribution repository and end-to-end XBPS proof do not yet exist. Current host inspection found Void x86_64/glibc with runit, polkit support and an NVMe controller; Fenris and smartmontools were absent. Verify these facts again before host changes.
The design decisions are recorded in ADR 0008. The dedicated distribution repository and end-to-end XBPS proof are complete; the host acceptance record is [issue #87](https://git.bongbetic.com/xavierk/Fenris/issues/87). The accepted target is Void x86_64/glibc with runit, with the released package installed and monitoring the selected NVMe drive after the coordinated reboot and lifecycle checks.
+5 -5
View File
@@ -68,8 +68,8 @@ Cache behavior:
- Conditional requests with old ETag return 200 (full content), not 304
xbps-install behavior:
- Always fetches fresh repodata with `-S` flag
- Respects ETag changes for immediate discovery
- `-M -S` fetches fresh repodata while bypassing the on-disk cache
- The ordinary `-S` path can reuse a cached repodata archive
- No 6-hour delay observed in practice
Binary size limits:
@@ -77,8 +77,9 @@ Binary size limits:
- Real packages expected to be <10MB (vendored pure-Python)
- Gitea serves any file size without LFS
**Caveat**: Clients using `xbps-install -Su` without `-S` may use cached repodata.
The `-S` flag forces a fresh fetch. Users should always use `-Syu` for updates.
**Caveat**: Clients using `xbps-install -Su` or `-Syu` without `-M` may use
cached repodata. Users should use `-M -Syu` for updates when immediate
publication visibility matters.
### 5. Publish index/artifacts together, preserve older artifacts, safe failure recovery
@@ -117,7 +118,6 @@ xavierk/Fenris-xbps (stable branch)
fenris-<version>_1.x86_64.xbps # Package archives
fenris-<version>_1.x86_64.xbps.sig2 # Package signatures
x86_64-repodata # Repository index (zstd-compressed tar)
x86_64-repodata.sig2 # Repository metadata signature
keys/
fenris-xbps-signing.pub # Public signing key
README.md
+2 -2
View File
@@ -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:
+3 -3
View File
@@ -63,14 +63,14 @@ 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
fi
# Ensure log directory exists
mkdir -p /var/log/fenris-collect
chown fenris:fenris /var/log/fenris-collect 2>/dev/null || true
install -d -o root -g fenris -m 2770 /var/log/fenris-collect 2>/dev/null || true
fi
;;
abort-upgrade|abort-install|disappear)
+9 -3
View File
@@ -24,14 +24,14 @@ 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
fi
# Ensure log directory exists
mkdir -p /var/log/fenris-collect
chown fenris:fenris /var/log/fenris-collect 2>/dev/null || true
install -d -o root -g fenris -m 2770 /var/log/fenris-collect 2>/dev/null || true
fi
elif [ "$1" -ge 2 ]; then
# Upgrade — snapshot, migration, init-system-aware reload
@@ -67,5 +67,11 @@ print(f'Fenris migration: {n} step(s) applied') if n else None
done
# Re-apply placement modes (store dir group access, issue #54)
systemd-tmpfiles --create || true
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
+3
View File
@@ -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"
+7 -5
View File
@@ -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
@@ -48,19 +52,17 @@ print(f'Fenris migration: {n} step(s) applied') if n else None
" 2>&1 || echo "Fenris: migration skipped (store not yet initialized)"
fi
# Ensure log directory exists on upgrade
mkdir -p /var/log/fenris-collect
chown fenris:fenris /var/log/fenris-collect 2>/dev/null || true
groupadd -f fenris
install -d -o root -g fenris -m 2770 /var/log/fenris-collect 2>/dev/null || true
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
fi
# Ensure log directory exists
mkdir -p /var/log/fenris-collect
chown fenris:fenris /var/log/fenris-collect 2>/dev/null || true
install -d -o root -g fenris -m 2770 /var/log/fenris-collect 2>/dev/null || true
fi
;;
esac
+3 -2
View File
@@ -1,8 +1,9 @@
[project]
name = "fenris"
version = "0.3.5"
version = "0.3.6"
description = "NVMe wear monitor with persistent TUI"
requires-python = ">=3.9"
requires-python = ">=3.10"
license = {file = "LICENSE"}
dependencies = [
"textual>=0.40.0",
]
+1
View File
@@ -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
+9 -30
View File
@@ -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.
"""
monitor_cmd = "/usr/libexec/fenris/fenris-monitor"
if is_root():
result = subprocess.run([monitor_cmd] + list(args))
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)
try:
invoke_monitor(*args)
except MonitorError as exc:
print(str(exc), file=sys.stderr)
sys.exit(exc.exit_code)
sys.exit(0)
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")
+30 -4
View File
@@ -42,7 +42,7 @@ for arg in "$@"; do
echo " --publish Execute the full publication flow"
echo ""
echo "Environment:"
echo " XBPS_SIGNING_KEY Path to SSH RSA private key (default: ~/.ssh/id_rsa)"
echo " XBPS_SIGNING_KEY Path to SSH RSA private key (default: ~/.ssh/id_xbps)"
echo " SIGNED_BY Signature identity (default: Fenris Packaging <packaging@bongbetic.com>)"
exit 0
;;
@@ -98,7 +98,7 @@ XBPS_PATH="${SOURCE_DIR}/${XBPS_FILE}"
GIT_USER_NAME=$(git config user.name || true)
GIT_USER_EMAIL=$(git config user.email || true)
if [[ -z "$GIT_USER_NAME" || -z "$GIT_USER_EMAIL" ]]; then
if [[ -z "${GIT_USER_NAME}" || -z "${GIT_USER_EMAIL}" ]]; then
echo "ERROR: Configure git user.name and user.email in the source repository before publishing" >&2
exit 1
fi
@@ -154,8 +154,34 @@ echo ""
# ── Step 6: Verify publication ───────────────────────────────────────────
echo "--- Verify publication ---"
RAW_URL="${GITEA_URL}/${GITEA_OWNER}/${GITEA_REPO}/raw/branch/${GITEA_BRANCH}/${ARCH}/x86_64-repodata"
_run "curl -sI '${RAW_URL}' | head -5"
RAW_BASE="${GITEA_URL}/${GITEA_OWNER}/${GITEA_REPO}/raw/branch/${GITEA_BRANCH}/${ARCH}"
if $PUBLISH; then
# Compare every served byte with the artifact that was indexed and pushed.
# A successful HEAD request alone can still hide a stale or incomplete
# publication behind the raw endpoint's cache.
# Repository signatures are embedded in x86_64-repodata by xbps-rindex;
# only package signatures are separate .sig2 files.
for artifact in "${XBPS_FILE}" "${XBPS_FILE}.sig2" "x86_64-repodata"; do
case "${artifact}" in
"${XBPS_FILE}") local_path="${XBPS_PATH}" ;;
"${XBPS_FILE}.sig2") local_path="${XBPS_PATH}.sig2" ;;
*) local_path="${WORK_DIR}/${ARCH}/${artifact}" ;;
esac
downloaded="${WORK_DIR}/.${artifact}.download"
curl --fail --silent --show-error --location \
--output "${downloaded}" "${RAW_BASE}/${artifact}"
cmp -- "${local_path}" "${downloaded}"
rm -f "${downloaded}"
done
else
_run "curl --fail --silent --show-error --location --output ${WORK_DIR}/.${XBPS_FILE}.download ${RAW_BASE}/${XBPS_FILE}"
_run "cmp -- ${XBPS_PATH} ${WORK_DIR}/.${XBPS_FILE}.download"
_run "curl --fail --silent --show-error --location --output ${WORK_DIR}/.${XBPS_FILE}.sig2.download ${RAW_BASE}/${XBPS_FILE}.sig2"
_run "cmp -- ${XBPS_PATH}.sig2 ${WORK_DIR}/.${XBPS_FILE}.sig2.download"
_run "curl --fail --silent --show-error --location --output ${WORK_DIR}/.x86_64-repodata.download ${RAW_BASE}/x86_64-repodata"
_run "cmp -- ${WORK_DIR}/${ARCH}/x86_64-repodata ${WORK_DIR}/.x86_64-repodata.download"
_run "rm -f ${WORK_DIR}/.${XBPS_FILE}.download ${WORK_DIR}/.${XBPS_FILE}.sig2.download ${WORK_DIR}/.x86_64-repodata.download"
fi
echo ""
# ── Cleanup ──────────────────────────────────────────────────────────────
+1 -1
View File
@@ -1,2 +1,2 @@
"""Fenris: NVMe wear monitor with persistent TUI."""
__version__ = "0.3.1"
__version__ = "0.3.6"
+56
View File
@@ -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,
)
+4 -4
View File
@@ -166,11 +166,11 @@ def _systemd_query_state() -> Dict[str, Any]:
"ActiveState", "ExecMainStatus", "ExecMainExitTimestamp",
)
boot_enabled_str = timer_props.get("UnitFileState", "")
boot_enabled = boot_enabled_str == "enabled"
boot_enabled_str = timer_props.get("UnitFileState")
boot_enabled = boot_enabled_str == "enabled" if boot_enabled_str else None
active_state = timer_props.get("ActiveState", "inactive")
timer_active = active_state == "active"
active_state = timer_props.get("ActiveState")
timer_active = active_state == "active" if active_state else None
last_collect_ok = None
last_collect_age_s = None
+100 -229
View File
@@ -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],
drive_facts: List[str], config_error: Optional[str],
store_fault: Optional[str], newer_schema: Optional[str],
journal_hint: Optional[str],
def _format_projection(proj, freshness: str, drive_facts: List[str],
config_error: Optional[str],
sample_count: int = 0, day_count: int = 0) -> str:
"""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,
query_services: bool = True, query_journal: bool = True) -> str:
"""Render the complete read-only status (§8.8, LC-9).
@contextmanager
def read_status(
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,
never prompts, and never writes to the store.
Own acquisition, fault classification, and connection lifetime for both
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:
clock_now = datetime.now(timezone.utc)
from .status_composition import compose_status
# --- Configuration (§8.3) ---
config_error = None
device = None
try:
config = read_config()
device = config["device"]
except ConfigError as e:
config_error = str(e)
# --- Service state ---
service = {}
clock_now = clock_now or datetime.now(timezone.utc)
service = None
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)
if ts.tzinfo is None:
ts = ts.replace(tzinfo=timezone.utc)
freshness_age_s = int((clock_now - ts).total_seconds())
except (ValueError, TypeError):
service = query_service_state()
except (OSError, RuntimeError):
pass
service["freshness"] = freshness
service["freshness_age_s"] = freshness_age_s
conn = None
store_fault = newer_schema = None
try:
service["deliberately_paused"] = is_deliberately_paused(conn, service)
except sqlite3.Error:
service["deliberately_paused"] = False
try:
conn = open_store_readonly(store_path or Path("/var/lib/fenris/observations.db"))
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) ---
drive_facts = []
comp = compose_status(
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)
except sqlite3.Error:
pass
read_config()
except ConfigError as exc:
config_error = str(exc)
# --- Projection (§6 — recomputed on read, never stored) ---
try:
proj = compute_projection(conn, clock_now)
except Exception:
proj = None
# --- Journal hint on failure or staleness (§8.8) ---
journal_hint = None
if query_journal and freshness in ("missed", "stale"):
journal_hint = _journalctl_hint()
# --- Compose output ---
result = _format_projection(
proj, freshness, service, drive_facts, config_error,
None, None, journal_hint, sample_count, day_count,
)
conn.close()
return result
with read_status(store_path, clock_now, query_services) as (conn, comp):
parts = [render_status_cli(comp)]
if not comp.store_fault and not comp.newer_schema:
drive_facts = []
proj = None
if conn is not None:
try:
drive_facts = _query_drive_facts(conn)
proj = compute_projection(conn, clock_now)
except (sqlite3.Error, ValueError, TypeError):
pass
parts.append(_format_projection(
proj, comp.freshness, drive_facts, config_error,
comp.sample_count, comp.day_count,
))
if query_journal and (
comp.store_fault or comp.last_collect_ok is False
or comp.freshness in ("missed", "stale")
):
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.
This is the new entry point that centralizes the status lattice.
"""
# Lazy import to avoid circular dependency
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
"""Return monitoring status without retaining the read-only snapshot."""
with read_status(
store_path, clock_now, query_services, collecting, reduced_motion,
) as (_, comp):
return comp
+54 -39
View File
@@ -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
timer_active = service.get("timer_active") if service_available else None
last_collect_ok = service.get("last_collect_ok") if service_available else None
last_collect_age_s = service.get("last_collect_age_s") if service_available else None
last_collect_reason = service.get("last_collect_reason") if service_available else None
boot_enabled = service.get("boot_enabled") if service else None
timer_active = service.get("timer_active") if service 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 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:
freshness = "unknown"
except sqlite3.Error as exc:
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:
pass
except sqlite3.Error as exc:
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:
deliberately_paused = False
except sqlite3.Error as exc:
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)
if comp.freshness_age_s is not None:
facts[-1] += " (%s)" % freshness_age_human(comp.freshness_age_s) if facts else "freshness: %s" % freshness_age_human(comp.freshness_age_s)
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")
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" % ("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("boot: %s" % (
"unknown" if comp.boot_enabled is None else "enabled" if comp.boot_enabled else "disabled"
))
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")
if comp.boot_enabled is not None:
facts.append("boot: %s" % ("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("Last collect: unknown")
facts.append("Boot: %s" % (
"unknown" if comp.boot_enabled is None else "enabled" if comp.boot_enabled else "disabled"
))
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)
+159 -334
View File
@@ -14,14 +14,13 @@ Criteria: TUI-1, TUI-2, TUI-4, CI-1, CI-2, CI-4, IN-3, LC-6, LC-8.
"""
from __future__ import annotations
import os
import sqlite3
import subprocess
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from rich.text import Text
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import Container, Horizontal, VerticalScroll
@@ -39,27 +38,19 @@ from .status import (
CADENCE_DEFAULT_S,
FRESH_THRESHOLD_S,
STALENESS_THRESHOLD_S,
ConfigError,
NewerSchema,
StoreFault,
format_disclosures,
freshness_age_human,
grade_freshness,
deliberate_pause_lines,
is_deliberately_paused,
monitoring_continuity,
open_store_readonly,
query_service_state,
read_config,
read_status,
_journalctl_hint,
)
from .monitoring_periods import get_open_period
from .status_composition import (
StatusState,
StatusComposition,
compose_status,
render_status_tui,
STATUS_POLL_INTERVAL_S,
)
from .control import MONITOR_HELPER, MonitorError, run_monitor
from .preferences import load_preferences, save_preferences
from .themes import THEMES, THEME_NAMES, get_theme, get_graph_colors
@@ -68,9 +59,10 @@ from .themes import THEMES, THEME_NAMES, get_theme, get_graph_colors
# Helpers
# ---------------------------------------------------------------------------
_RESUME_HINT = "r resume — enable monitoring and future boots"
_RESUME_HINT = "r Resume — enable monitoring and future boots"
_ACTION_LEGEND = (
"p pause · " + _RESUME_HINT + " · c collect · t theme · m motion · d disclosures"
"p Pause · " + _RESUME_HINT + "\n"
"c Collect now · t Theme · m Motion · d Disclosures · ? Help"
)
def _format_remaining(seconds: float) -> str:
@@ -91,43 +83,6 @@ def _format_remaining(seconds: float) -> str:
return " ".join(parts)
def _sparkline(values: List[float], width: int = 40) -> str:
"""Render a sparkline from daily bytes-written values."""
if not values:
return ""
mx = max(values) or 1.0
blocks = " ▁▂▃▄▅▆▇█"
step = max(1, len(values) // width or 1)
picked = values[-width * step :][::step][-width:]
return "".join(
blocks[min(len(blocks) - 1, int(v / mx * (len(blocks) - 1)) + (1 if v > 0 else 0))]
for v in picked
)
def _habit_bar(a: float, i: float, o: float, u: float, width: int = 40) -> str:
"""Render the habit-split bar with legend."""
total = a + i + o + u or 1.0
segs = [
("a", a, "#33ff33"),
("i", i, "#ffff33"),
("o", o, "#33ffff"),
("?", u, "#ff33ff"),
]
parts = []
for label, v, _color in segs:
n = max(1 if v > 0 else 0, round(v / total * width))
parts.append(label * n)
bar = "".join(parts)
legend = " active %d%% · idle %d%% · powered-off %d%% · unknown %d%%" % (
round(a / total * 100),
round(i / total * 100),
round(o / total * 100),
round(u / total * 100),
)
return bar + "\n" + legend
# ---------------------------------------------------------------------------
# Identity and wolf glyph detection (issue #79)
# ---------------------------------------------------------------------------
@@ -274,8 +229,8 @@ class DailyBarGraph(Widget):
self._render_readout()
def _show_empty(self) -> None:
self.query_one("#bar-range").update("[dim]usage history[/dim]")
self.query_one("#bar-render").update("[dim]awaiting first sample[/dim]")
self.query_one("#bar-range").update("[dim]Usage history[/dim]")
self.query_one("#bar-render").update("[dim]Awaiting first sample[/dim]")
self.query_one("#bar-legend").update("")
self.query_one("#bar-readout").update("")
@@ -296,9 +251,9 @@ class DailyBarGraph(Widget):
def _show_constrained_summary(self) -> None:
"""Textual fallback for terminals below 80×24."""
self.query_one("#bar-range").update("[dim]usage history[/dim]")
self.query_one("#bar-range").update("[dim]Usage history[/dim]")
if not self._day_data:
self.query_one("#bar-render").update("[dim]graph needs ≥80×24[/dim]")
self.query_one("#bar-render").update("[dim]Graph needs ≥80×24[/dim]")
self.query_one("#bar-legend").update("")
self.query_one("#bar-readout").update("")
return
@@ -310,7 +265,7 @@ class DailyBarGraph(Widget):
first = self._day_data[0].get("local_label", "")
last = self._day_data[-1].get("local_label", "")
self.query_one("#bar-render").update(
"[dim]graph needs ≥80×24[/dim]\n"
"[dim]Graph needs ≥80×24[/dim]\n"
" %d days · %d with writes · %.3f GB total\n"
" %s → %s" % (n, days_with_data, total_bytes / 1e9, first, last)
)
@@ -324,7 +279,7 @@ class DailyBarGraph(Widget):
% (day.get("local_label", ""), day.get("total_bytes", 0) / 1e9)
)
else:
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select[/dim]")
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select[/dim]")
def _show_constrained_hourly_summary(self) -> None:
"""Textual fallback for hourly view when terminal is too small."""
@@ -334,7 +289,7 @@ class DailyBarGraph(Widget):
)
n = len(self._hour_data)
self.query_one("#bar-render").update(
"[dim]graph needs ≥80×24[/dim]\n"
"[dim]Graph needs ≥80×24[/dim]\n"
" %d hours · %d with writes · %.3f GB total" % (n, hours_with_data, total_bytes / 1e9)
)
self.query_one("#bar-legend").update("")
@@ -346,7 +301,7 @@ class DailyBarGraph(Widget):
% (h.get("local_label", ""), h.get("bytes_written", 0) / 1e9)
)
else:
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]")
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]")
def on_resize(self) -> None:
"""Re-render when terminal size changes."""
@@ -362,9 +317,9 @@ class DailyBarGraph(Widget):
parts.append("[bold]%d[/bold]" % r)
else:
parts.append(str(r))
label = "range: " + " / ".join(parts)
label = "Range: " + " / ".join(parts)
if self.view_mode == "hourly":
label += " \u00b7 [bold]%s[/bold] \u00b7 esc back" % (self.drill_day or "")
label += " \u00b7 [bold]%s[/bold] \u00b7 Esc Back" % (self.drill_day or "")
self.query_one("#bar-range").update(label)
def _render_bars(self) -> None:
@@ -436,7 +391,7 @@ class DailyBarGraph(Widget):
def _render_legend(self) -> None:
legend = (
"%s alloc %s unalloc %s gap %s zero %s partial"
"%s Alloc %s Unalloc %s Gap %s Zero %s Partial"
% (
_GLYPH_ALLOCATED,
_GLYPH_UNALLOCATED,
@@ -450,7 +405,7 @@ class DailyBarGraph(Widget):
def _render_readout(self) -> None:
if self.selected_index < 0 or self.selected_index >= len(self._day_data):
self.query_one("#bar-readout").update(
"[dim]\u2190 \u2192 select \u00b7 1-4 range \u00b7 Enter drill[/dim]"
"[dim]\u2190 \u2192 Select \u00b7 1-4 Range \u00b7 Enter Hourly view[/dim]"
)
return
@@ -473,7 +428,7 @@ class DailyBarGraph(Widget):
]
if unallocated > 0:
parts.append(
" allocated %.3f GB \u00b7 unallocated %.3f GB"
" Allocated %.3f GB \u00b7 unallocated %.3f GB"
% (allocated / 1e9, unallocated / 1e9)
)
self.query_one("#bar-readout").update("\n".join(parts))
@@ -482,7 +437,7 @@ class DailyBarGraph(Widget):
def _refresh_hourly(self) -> None:
if not self._hour_data:
self.query_one("#bar-render").update("[dim]no hourly data[/dim]")
self.query_one("#bar-render").update("[dim]No hourly data[/dim]")
return
self._render_range()
@@ -533,7 +488,7 @@ class DailyBarGraph(Widget):
self.query_one("#bar-render").update("\n".join(lines))
self.query_one("#bar-legend").update(
"%s writes \u00b7 %s zero" % (_GLYPH_ALLOCATED, _GLYPH_ZERO)
"%s Writes \u00b7 %s Zero" % (_GLYPH_ALLOCATED, _GLYPH_ZERO)
)
# Hourly readout
@@ -548,7 +503,7 @@ class DailyBarGraph(Widget):
)
)
else:
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 select hour[/dim]")
self.query_one("#bar-readout").update("[dim]\u2190 \u2192 Select hour[/dim]")
# -- Event handling --
@@ -754,51 +709,6 @@ def _query_hourly_graph_data(
# Data queries for TUI regions
# ---------------------------------------------------------------------------
def _query_usage_history(conn: sqlite3.Connection) -> Dict[str, Any]:
"""Query usage-history data for the left pane."""
cursor = conn.execute(
"SELECT day, active_seconds, idle_seconds, powered_off_seconds, "
"unknown_seconds, bytes_written_delta, coverage "
"FROM day_aggregates ORDER BY day"
)
days = cursor.fetchall()
if not days:
return {
"sparkline": "",
"habit_bar": "",
"num_days": 0,
"min_gb": 0,
"max_gb": 0,
"habit_change": False,
"gap_days": [],
"day_labels": [],
}
bw_values = [d[5] for d in days]
total_a = sum(d[1] for d in days)
total_i = sum(d[2] for d in days)
total_o = sum(d[3] for d in days)
total_u = sum(d[4] for d in days)
total = total_a + total_i + total_o + total_u or 1
spark = _sparkline([b / 1e9 for b in bw_values]) # Convert to GB for display
bar = _habit_bar(total_a / total, total_i / total, total_o / total, total_u / total)
min_gb = min(bw_values) / 1e9 if bw_values else 0
max_gb = max(bw_values) / 1e9 if bw_values else 0
return {
"sparkline": spark,
"habit_bar": bar,
"num_days": len(days),
"min_gb": min_gb,
"max_gb": max_gb,
"habit_change": False,
"gap_days": [],
"day_labels": [d[0] for d in days],
}
def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
"""Query drive health data for the right pane."""
cursor = conn.execute(
@@ -840,41 +750,6 @@ def _query_drive_health(conn: sqlite3.Connection) -> Dict[str, Any]:
}
def _query_service_facts(conn: sqlite3.Connection, clock_now: datetime) -> Dict[str, Any]:
"""Query service facts for the bottom strip."""
# Get freshness
cursor = conn.execute("SELECT ts FROM samples ORDER BY id DESC LIMIT 1")
row = cursor.fetchone()
newest_ts = row[0] if row else None
freshness = grade_freshness(newest_ts, clock_now)
# Get monitoring period
period = get_open_period(conn)
period_info = "no monitoring period"
if period:
period_info = "open since %s" % period["started_at"][:10]
# Get service state
try:
svc = query_service_state()
except Exception:
svc = {
"boot_enabled": False,
"timer_active": False,
"last_collect_ok": None,
"last_collect_age_s": None,
"last_collect_reason": None,
}
return {
"freshness": freshness,
"freshness_age_s": None,
"period": period_info,
"deliberately_paused": is_deliberately_paused(conn, svc),
**svc,
}
# ---------------------------------------------------------------------------
# Modal screens
# ---------------------------------------------------------------------------
@@ -894,7 +769,7 @@ class ConfirmPause(ModalScreen[bool]):
"This closes the current monitoring period.\n"
"Paused time is [bold]excluded[/bold] from your usage habit\n"
"(powered-off time would still count).\n\n"
"[dim]y pause · n cancel[/dim]",
"[dim]y Pause · n Cancel[/dim]",
id="confirm-text",
)
@@ -915,7 +790,44 @@ class DisclosuresScreen(ModalScreen[None]):
def compose(self) -> ComposeResult:
text = format_disclosures()
yield Static(text + "\n\n[dim]esc to close[/dim]", id="disc-text")
with VerticalScroll():
yield Static(text + "\n\n[dim]Esc Close[/dim]", id="disc-text")
def action_close(self) -> None:
self.dismiss()
class HelpScreen(ModalScreen[None]):
"""Persistent keyboard and privilege guidance."""
BINDINGS = [Binding("escape", "close", "Close"), Binding("?", "close", "Close")]
def compose(self) -> ComposeResult:
with VerticalScroll():
yield Static(
"[bold]Using Fenris[/bold]\n\n"
"Open the dashboard with [bold]fenris[/bold] as your normal user.\n"
"The dashboard does not need sudo. Pause, resume, and collect now\n"
"authenticate through polkit when needed.\n\n"
"[bold]When to use sudo[/bold]\n"
"Use sudo for installation and system configuration.\n"
"For observation store access, an administrator can run:\n"
' sudo usermod -aG fenris "$USER"\n'
"Then log out and back in before opening Fenris.\n\n"
"If polkit authentication is unavailable, quit with q and run\n"
"only the required action in your terminal:\n"
" sudo fenris monitor resume\n"
" sudo fenris monitor pause\n"
" sudo fenris sample\n"
"Then reopen the dashboard with fenris.\n\n"
"[bold]Keyboard controls[/bold]\n"
+ _ACTION_LEGEND + "\n"
"q Quit the dashboard — monitoring continues\n"
"Tab Focus the graph · ← → Select · 1-4 Change range\n"
"Enter Show hours · Esc Return to days\n\n"
"[dim]↑ ↓ Scroll · Esc Close[/dim]",
id="help-text",
)
def action_close(self) -> None:
self.dismiss()
@@ -936,12 +848,12 @@ class FenrisTuiApp(App):
layout: grid;
grid-size: 2 4;
grid-columns: 3fr 2fr;
grid-rows: 8 10 7 3;
grid-rows: auto 10 auto 3;
height: auto;
}
#main-grid.paused {
grid-size: 2 5;
grid-rows: 8 5 10 7 3;
grid-rows: auto auto 10 auto 3;
}
#main-grid.constrained {
grid-size: 1 4;
@@ -966,15 +878,16 @@ class FenrisTuiApp(App):
min-height: 3;
}
#dashboard-scroll { height: 1fr; }
#headline-band { column-span: 2; }
#headline-band { column-span: 2; height: auto; min-height: 8; }
#paused-banner {
column-span: 2;
display: none;
background: $error 20%;
color: $text;
height: 100%;
height: auto;
min-height: 5;
}
#service-strip { column-span: 2; height: 100%; }
#service-strip { column-span: 2; height: auto; }
#quit-rail {
column-span: 2;
border: heavy $accent;
@@ -984,15 +897,17 @@ class FenrisTuiApp(App):
.pane { border: round #555555; padding: 0 1; height: 100%; }
#confirm-text { padding: 1 2; }
#disc-text { padding: 1 2; }
#help-text { padding: 1 2; }
"""
BINDINGS = [
Binding("p", "pause", "Pause", show=False),
Binding("r", "resume", "Resume", show=False),
Binding("c", "collect", "Collect Now", show=False),
Binding("c", "collect", "Collect now", show=False),
Binding("d", "disclose", "Disclosures", show=False),
Binding("t", "toggle_theme", "Theme", show=False),
Binding("m", "toggle_motion", "Motion", show=False),
Binding("?", "help", "Help", show=False),
Binding("q", "quit", "Quit", show=False),
]
@@ -1012,10 +927,9 @@ class FenrisTuiApp(App):
super().__init__(**kwargs)
self.store_path = store_path or Path("/var/lib/fenris/observations.db")
self.config_path = config_path
self.helper_path = helper_path or "/usr/libexec/fenris/fenris-monitor"
self.helper_path = helper_path or MONITOR_HELPER
self.refresh_interval_s = refresh_interval_s
self._show_auth_notice = True
self._conn: Optional[sqlite3.Connection] = None
self._clock_now = datetime.now(timezone.utc)
# Register all Fenris themes
@@ -1034,7 +948,7 @@ class FenrisTuiApp(App):
yield Static("", id="constrained-summary", classes="pane")
yield Static("", id="drive-health", classes="pane")
yield Static("", id="service-strip", classes="pane")
yield Static("q QUIT TUI", id="quit-rail")
yield Static("q Quit TUI", id="quit-rail")
def on_mount(self) -> None:
"""Set border titles, apply theme, and render initial state."""
@@ -1042,34 +956,19 @@ class FenrisTuiApp(App):
# Preference uses underscores (high_contrast); Textual theme uses dashes (fenris-high-contrast)
self.theme = "fenris-%s" % self._current_theme_name.replace("_", "-")
self.query_one("#headline-band").border_title = "headline"
self.query_one("#usage-history").border_title = "usage history"
self.query_one("#constrained-summary").border_title = "usage history"
self.query_one("#drive-health").border_title = "drive"
self.query_one("#service-strip").border_title = "service + actions"
self.query_one("#headline-band").border_title = "Headline"
self.query_one("#usage-history").border_title = "Usage history"
self.query_one("#constrained-summary").border_title = "Usage history"
self.query_one("#drive-health").border_title = "Drive"
self.query_one("#service-strip").border_title = "Monitoring and actions"
self._refresh_timer = self.set_interval(
self.refresh_interval_s, self.on_refresh_tick
)
self._refresh()
def on_resize(self) -> None:
"""Handle terminal resize to switch between constrained and normal layout (issue #81)."""
width, height = self.size
was_constrained = self._is_constrained_mode
self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT
main_grid = self.query_one("#main-grid")
if self._is_constrained_mode:
main_grid.add_class("constrained")
else:
main_grid.remove_class("constrained")
# Re-render constrained summary when entering constrained mode
if self._is_constrained_mode and not was_constrained:
self._update_constrained_summary()
# Re-render full layout when leaving constrained mode
elif not self._is_constrained_mode and was_constrained:
self._refresh()
"""Re-render with the new terminal size after Textual applies it."""
self.call_after_refresh(self._refresh)
def _update_constrained_summary(self) -> None:
"""Update the textual summary shown when terminal is too small for graph."""
@@ -1077,7 +976,7 @@ class FenrisTuiApp(App):
summary = self.query_one("#constrained-summary")
if not graph._day_data:
summary.update("[dim]graph needs ≥80×24[/dim]\nawaiting first sample")
summary.update("[dim]Graph needs ≥80×24[/dim]\nAwaiting first sample")
return
# Summarise visible days as text (issue #81 AC2)
@@ -1087,7 +986,7 @@ class FenrisTuiApp(App):
first = graph._day_data[0].get("local_label", "")
last = graph._day_data[-1].get("local_label", "")
text = (
"[dim]graph needs ≥80×24[/dim]\n"
"[dim]Graph needs ≥80×24[/dim]\n"
" %d days · %d with writes · %.3f GB total\n"
" %s → %s" % (n, days_with_data, total_bytes / 1e9, first, last)
)
@@ -1116,7 +1015,7 @@ class FenrisTuiApp(App):
identity = _IDENTITY_FALLBACK
lines = ["[bold]%s[/bold]" % identity]
if self._show_auth_notice:
lines.append("[dim]privileged actions will prompt for authentication (polkit)[/dim]")
lines.append("[dim]Open with fenris (no sudo). Actions authenticate via polkit. ? Help[/dim]")
return "\n".join(lines)
def _render_headline(self, body: str = "") -> None:
@@ -1126,94 +1025,67 @@ class FenrisTuiApp(App):
text += "\n\n" + body
self.query_one("#headline-band").update(text)
def _open_store(self) -> Optional[sqlite3.Connection]:
"""Open store read-only, handling faults."""
try:
return open_store_readonly(self.store_path)
except (StoreFault, NewerSchema):
return None
def _refresh(self) -> None:
"""Refresh all four regions from store data."""
self._clock_now = datetime.now(timezone.utc)
# Also update constrained state on each refresh (issue #81)
try:
width, height = self.size
was_constrained = self._is_constrained_mode
self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT
main_grid = self.query_one("#main-grid")
if self._is_constrained_mode:
main_grid.add_class("constrained")
else:
main_grid.remove_class("constrained")
if self._is_constrained_mode and not was_constrained:
self._update_constrained_summary()
except Exception:
pass
conn = self._open_store()
if conn is None:
self._render_empty_or_fault()
return
try:
self._conn = conn
self._render_all_regions(conn)
finally:
conn.close()
self._conn = None
def _render_empty_or_fault(self) -> None:
"""Render empty store greeting or store fault."""
self._hide_paused_banner()
if not self.store_path.exists():
# Empty store — greeting with enable hint (IN-3)
self._render_headline(
"[bold]No observations yet[/bold]\n\n"
"[bold]%s[/bold]" % _RESUME_HINT
)
self.query_one("#usage-history").set_data([])
if self._is_constrained_mode:
self.query_one("#constrained-summary").update(
"[dim]graph needs ≥80×24[/dim]\nawaiting first sample"
)
self.query_one("#drive-health").update("")
self.query_one("#service-strip").update(
(
"boot: disabled · timer: inactive · last collect: unknown · freshness: empty\n"
"[bold]CONTINUITY[/bold] %s\n" + _ACTION_LEGEND
) % monitoring_continuity({"boot_enabled": False})
)
width, height = self.size
self._is_constrained_mode = width < _MIN_WIDTH or height < _MIN_HEIGHT
main_grid = self.query_one("#main-grid")
if self._is_constrained_mode:
main_grid.add_class("constrained")
else:
# Store fault (FL-4)
self._render_headline(
"[bold red]Observation store unreadable[/bold red]\n"
"Check journalctl -u fenris-collect.service"
)
self.query_one("#usage-history").set_data([])
if self._is_constrained_mode:
self.query_one("#constrained-summary").update(
"[dim]graph needs ≥80×24[/dim]\nobservation store unreadable"
)
self.query_one("#drive-health").update("")
main_grid.remove_class("constrained")
with read_status(
self.store_path, self._clock_now, reduced_motion=self._reduced_motion,
) as (conn, comp):
if conn is None:
self._render_empty_or_fault(comp)
else:
self._render_all_regions(conn, comp)
self.query_one("#service-strip").update(
_ACTION_LEGEND
"%s\n%s" % (render_status_tui(comp), _ACTION_LEGEND)
)
self._render_paused_banner(comp)
def _render_empty_or_fault(self, comp: StatusComposition) -> None:
"""Render empty history or the shared store-fault classification."""
if comp.newer_schema or comp.store_fault:
message = comp.explanation[:1].upper() + comp.explanation[1:]
self._render_headline("[bold red]%s[/bold red]" % message)
if comp.store_fault:
hint = _journalctl_hint()
if hint:
self.query_one("#drive-health").update(Text(hint))
else:
self.query_one("#drive-health").update(
"Check observation store permissions and collector logs. ? Help"
)
else:
self.query_one("#drive-health").update("")
else:
message = "Awaiting first sample"
self._render_headline(
"[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT
)
self.query_one("#drive-health").update("")
graph = self.query_one("#usage-history")
graph.set_data([])
if comp.store_fault or comp.newer_schema:
self.query_one("#bar-render").update(message)
if self._is_constrained_mode:
self.query_one("#constrained-summary").update(
"[dim]Graph needs ≥80×24[/dim]\n" + message
)
def _render_all_regions(self, conn: sqlite3.Connection) -> None:
def _render_all_regions(self, conn: sqlite3.Connection, comp: StatusComposition) -> None:
"""Render all four regions from live store data."""
# --- Sample count for single-sample state (issue #73 AC3) ---
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 Exception:
sample_count = 0
day_count = 0
# --- Headline band (§7.2) ---
if sample_count <= 1 and day_count == 0:
if comp.sample_count == 0 and comp.day_count == 0:
self._render_headline(
"[bold]No observations yet[/bold]\n\n[bold]%s[/bold]" % _RESUME_HINT
)
elif comp.sample_count <= 1 and comp.day_count == 0:
# Single sample: awaiting another sample
self._render_headline(
"[bold]Awaiting another sample[/bold]\n\n"
@@ -1253,10 +1125,10 @@ class FenrisTuiApp(App):
# Vendor wear grouped under Drive health with context
health_text = (
"[bold]Drive health[/bold] · %s\n"
" temperature %d°C · spare %d%%\n"
" media errors %d · unsafe shutdowns %d\n"
" power-on %d h · %d cycles · %s\n"
" vendor wear: %d%% used · %.1f TB written"
" Temperature %d°C · spare %d%%\n"
" Media errors %d · unsafe shutdowns %d\n"
" Power-on %d h · %d cycles · %s\n"
" Vendor wear: %d%% used · %.1f TB written"
) % (
health["model"],
health["temp"],
@@ -1271,30 +1143,6 @@ class FenrisTuiApp(App):
)
self.query_one("#drive-health").update(health_text)
# --- Service strip (§7.2 bottom) ---
try:
svc = _query_service_facts(conn, self._clock_now)
# Use shared status composition
comp = compose_status(
conn, svc, self._clock_now,
store_fault=None,
newer_schema=None,
reduced_motion=self._reduced_motion,
)
# Render using the shared composition with TUI styling
status_text = render_status_tui(comp)
# Add TUI-only actions
self.query_one("#service-strip").update(
"%s\n%s" % (status_text, _ACTION_LEGEND)
)
self._render_paused_banner(comp)
except Exception:
self._hide_paused_banner()
self.query_one("#service-strip").update(
"boot: unknown · timer: unknown · last collect: unknown · freshness: unknown\n"
+ _ACTION_LEGEND
)
def _render_paused_banner(self, comp) -> None:
"""Show the high-contrast Deliberate disable block only when sanctioned.
@@ -1312,7 +1160,7 @@ class FenrisTuiApp(App):
if is_paused and paused_lines:
banner.update(
"[bold black on red]%s[/bold black on red]\n%s"
% tuple(paused_lines)
% tuple(line[:1].upper() + line[1:] for line in paused_lines)
)
banner.styles.display = "block"
main_grid = self.query_one("#main-grid")
@@ -1331,16 +1179,12 @@ class FenrisTuiApp(App):
def _on_graph_drill(self, day: str) -> None:
"""Load hourly data when the graph enters drill-down mode."""
graph = self.query_one("#usage-history")
try:
conn = self._open_store()
if conn is not None:
try:
hour_data = _query_hourly_graph_data(conn, day)
graph.set_hour_data(hour_data)
finally:
conn.close()
except Exception:
graph.set_hour_data([])
with read_status(
self.store_path, self._clock_now, query_services=False,
) as (conn, _):
graph.set_hour_data(
_query_hourly_graph_data(conn, day) if conn is not None else []
)
def _format_headline(self, proj: ProjectionResult) -> str:
"""Format the lifespan headline (spec §6.11)."""
@@ -1357,7 +1201,7 @@ class FenrisTuiApp(App):
regime = " · sustained regime: %d days" % proj.regime_days
return (
"[bold]Usage-adjusted theoretical lifespan: [white]%s remaining[/white][/bold]"
"\n if current habits continue%s" % (remaining, regime)
"\n If current habits continue%s" % (remaining, regime)
)
def _format_confidence(self, proj: ProjectionResult) -> str:
@@ -1377,7 +1221,7 @@ class FenrisTuiApp(App):
color,
proj.confidence_state.value,
color,
facts,
facts[:1].upper() + facts[1:],
)
def _format_scenario(self, proj: ProjectionResult) -> str:
@@ -1436,48 +1280,29 @@ class FenrisTuiApp(App):
def action_collect(self) -> None:
"""Collect now — synchronous outcome (spec §8.7, LC-8)."""
self._run_helper("collect", blocking=True)
self._run_helper("collect")
def action_disclose(self) -> None:
"""Show disclosures (spec §6.11, CI-4)."""
self.push_screen(DisclosuresScreen())
def action_help(self) -> None:
"""Show keyboard controls and sudo guidance."""
self.push_screen(HelpScreen())
def _run_helper(
self,
operation: str,
extra_args: Optional[List[str]] = None,
blocking: bool = False,
) -> None:
"""Run fenris-monitor as terminal-attached subprocess (LC-6, LC-8).
The TUI suspends, polkit agent prompts on real terminal, control returns.
"""
cmd = [self.helper_path, operation]
if extra_args:
cmd.extend(extra_args)
if os.geteuid() != 0:
cmd.insert(0, "pkexec")
"""Suspend the TUI while the shared control module owns the action."""
try:
with self.suspend():
proc = subprocess.run(cmd, timeout=30)
if proc.returncode != 0:
self.notify(
"Operation failed (exit %d)" % proc.returncode,
severity="error",
)
except FileNotFoundError:
self.notify(
"Helper not found: %s" % self.helper_path,
severity="error",
)
except subprocess.TimeoutExpired:
self.notify("Operation timed out", severity="error")
except Exception as e:
self.notify("Error: %s" % e, severity="error")
# Refresh after action
self._refresh()
run_monitor(operation, *(extra_args or []), helper_path=self.helper_path)
except MonitorError as exc:
self.notify(str(exc), severity="error")
finally:
self._refresh()
def run_tui(
+1 -31
View File
@@ -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()
+1 -1
View File
@@ -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(
+77
View File
@@ -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()
+3 -3
View File
@@ -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["timer_active"] is False
assert result["boot_enabled"] is None
assert result["timer_active"] is None
+4 -4
View File
@@ -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,
+41 -10
View File
@@ -96,6 +96,29 @@ def _find_package(fmt: str) -> Path:
return candidate
def test_runit_logger_uses_accounts_shipped_by_package():
"""The runit logger must use the package's group-only identity.
The package declares a ``fenris`` group for store/log access, not a
``fenris`` service user. Starting the logger with ``fenris:fenris``
therefore leaves the diagnostics supervisor down on a real Void host;
Void's standard ``nobody`` account supplies the unprivileged uid.
"""
logger = (REPO_ROOT / "units" / "runit" / "fenris-collect" / "log" / "run").read_text()
sysusers = (REPO_ROOT / "packaging" / "sysusers.d" / "fenris.conf").read_text()
assert "g fenris -" in sysusers
assert "chpst -u nobody:fenris" in logger
def test_xbps_publisher_verifies_embedded_repository_signature():
"""Publication verification must match XBPS's repository layout."""
publisher = (REPO_ROOT / "scripts" / "xbps-publish.sh").read_text()
assert '"x86_64-repodata"' in publisher
assert '"x86_64-repodata.sig2"' not in publisher
# ---------------------------------------------------------------------------
# Matrix definitions
# ---------------------------------------------------------------------------
@@ -112,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
@@ -153,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 && \\
xbps-install -Syun python3 smartmontools runit && \\
xbps-remove -Oy void-repo-nonfree || true
RUN xbps-install -Sy python3 smartmontools runit shadow && \\
xbps-remove -O
COPY dist/{pkg_name} /pkg/{pkg_name}
""")
@@ -1690,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"
@@ -1708,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(
@@ -1732,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"
@@ -1780,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"
@@ -1866,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"
@@ -1937,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"
+1
View File
@@ -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={
+5 -5
View File
@@ -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",
+150
View File
@@ -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")
+60 -109
View File
@@ -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 "timer:" in strip
assert "last collect:" in strip
assert "freshness:" in strip
assert "Boot:" in strip
assert "Timer:" in strip
assert "Last collect:" 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), \
patch("fenris.tui.subprocess.run") as run, \
with patch("fenris.control.os.geteuid", return_value=1000), \
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
@@ -924,8 +875,8 @@ 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 "zero" in legend
assert "Alloc" in legend
assert "Zero" in legend
@pytest.mark.asyncio
async def test_arrow_selection(self, tmp_path):
@@ -1091,7 +1042,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 +1123,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 +1147,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):
@@ -1273,7 +1224,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 +1236,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 +1266,13 @@ class TestConstrainedLayout:
# Service strip should have actions
strip = str(app.query_one("#service-strip").render())
assert "p pause" in strip
assert "r resume" in strip
assert "p Pause" 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):
+4 -1
View File
@@ -5,5 +5,8 @@
# The logger writes to runit's log directory for diagnostics.
#
# Spec: §8.8 (actionable native diagnostics)
exec chpst -u fenris:fenris \
# The package creates the fenris group for shared diagnostics access; it does
# not create a service user. Use Void's standard unprivileged account while
# giving the logger the declared group identity.
exec chpst -u nobody:fenris \
svlogd -tt /var/log/fenris-collect/