feat: Add shared status composition for truthful monitoring states (issue #78)

src/fenris/status.py

@@ -687,3 +687,79 @@ def render_status(store_path: Optional[Path] = None, clock_now: Optional[datetim
         parts.append("")
         parts.append(format_disclosures())
     return "\n\n".join(parts)
+
+
+
+# ---------------------------------------------------------------------------
+# Shared status composition integration (issue #78)
+# ---------------------------------------------------------------------------
+
+def get_status_composition(
+    store_path: Optional[Path] = None,
+    clock_now: Optional[datetime] = None,
+    query_services: bool = True,
+    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

src/fenris/status_composition.py

@@ -0,0 +1,526 @@
+"""Shared status composition for TUI and CLI (issue #78, TPH-2/3).
+
+Centralizes the monitoring status lattice consumed by both surfaces.
+States: Monitoring, Collecting, Paused, Waiting, Interrupted, Error, Stale, Unknown.
+
+Precedence: Error > Interrupted > Paused > Stale > Waiting > Monitoring > Unknown.
+Collecting overlays every base except store fault.
+
+Freshness grading uses shared constants from status.py.
+"""
+import enum
+import sqlite3
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional
+
+# Status poll interval (AC78-6): lightweight 5s systemctl show poll
+STATUS_POLL_INTERVAL_S = 5
+
+from .status import (
+    FRESH_THRESHOLD_S,
+    STALENESS_THRESHOLD_S,
+    grade_freshness,
+    freshness_age_human,
+    is_deliberately_paused,
+    monitoring_continuity,
+    deliberate_pause_lines,
+    open_store_readonly,
+    StoreFault,
+    NewerSchema,
+)
+
+
+# ---------------------------------------------------------------------------
+# Status state enum with glyph and label
+# ---------------------------------------------------------------------------
+
+class StatusState(enum.Enum):
+    """The eight monitoring status states (TPH-2)."""
+
+    MONITORING = "monitoring"
+    COLLECTING = "collecting"
+    PAUSED = "paused"
+    WAITING = "waiting"
+    INTERRUPTED = "interrupted"
+    ERROR = "error"
+    STALE = "stale"
+    UNKNOWN = "unknown"
+
+    @property
+    def glyph(self) -> str:
+        _glyphs = {
+            "monitoring": "●",
+            "collecting": "◐",
+            "paused": "‖",
+            "waiting": "○",
+            "interrupted": "⊘",
+            "error": "✖",
+            "stale": "◌",
+            "unknown": "?",
+        }
+        return _glyphs[self.value]
+
+    @property
+    def label(self) -> str:
+        return self.value.capitalize()
+
+
+# Precedence order: higher index = higher precedence
+_PRECEDENCE = [
+    StatusState.UNKNOWN,
+    StatusState.MONITORING,
+    StatusState.WAITING,
+    StatusState.STALE,
+    StatusState.PAUSED,
+    StatusState.INTERRUPTED,
+    StatusState.ERROR,
+]
+
+_PRECEDENCE_RANK = {s: i for i, s in enumerate(_PRECEDENCE)}
+
+
+# ---------------------------------------------------------------------------
+# Status composition result
+# ---------------------------------------------------------------------------
+
+@dataclass
+class StatusComposition:
+    """The composed status result shared between TUI and CLI."""
+
+    state: StatusState
+    glyph: str
+    label: str
+    explanation: str
+
+    # Separate facts (never folded into the status word)
+    freshness: str = "unknown"
+    freshness_age_s: Optional[int] = None
+    last_collect_ok: Optional[bool] = None
+    last_collect_age_s: Optional[int] = None
+    last_collect_reason: Optional[str] = None
+    boot_enabled: Optional[bool] = None
+    timer_active: Optional[bool] = None
+    deliberately_paused: bool = False
+    pause_age_s: Optional[int] = None
+    external_stop_reason: Optional[str] = None
+
+    # Store fault / newer schema (suppress store-dependent views)
+    store_fault: Optional[str] = None
+    newer_schema: Optional[str] = None
+
+    # Collecting overlay
+    overlay_base: Optional[StatusState] = None
+
+    # Sample counts for waiting explanations
+    sample_count: int = 0
+    day_count: int = 0
+
+    # Whether the status dot should blink (only Monitoring)
+    should_blink: bool = False
+
+    # Continuity line
+    continuity: str = ""
+
+    # Paused lines (for TUI banner)
+    paused_lines: List[str] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# Status composition logic
+# ---------------------------------------------------------------------------
+
+def _determine_base_state(
+    freshness: str,
+    sample_count: int,
+    day_count: int,
+    boot_enabled: Optional[bool],
+    timer_active: Optional[bool],
+    last_collect_ok: Optional[bool],
+    deliberately_paused: bool,
+    external_stop_reason: Optional[str],
+    service_available: bool,
+) -> StatusState:
+    """Determine the base status state from facts.
+
+    Precedence: Error > Interrupted > Paused > Stale > Waiting > Monitoring > Unknown.
+    """
+    # Unknown: service query failed and no store-derived fact places us higher
+    if not service_available:
+        return StatusState.UNKNOWN
+
+    # Error: last collect failed
+    if last_collect_ok is False:
+        return StatusState.ERROR
+
+    # Interrupted: external stop (not user_disabled)
+    if external_stop_reason is not None:
+        return StatusState.INTERRUPTED
+
+    # Paused: deliberate disable
+    if deliberately_paused:
+        return StatusState.PAUSED
+
+    # Stale: timer active, no failure, but data ≥ 48h old
+    if freshness == "stale" and timer_active and last_collect_ok is not False:
+        return StatusState.STALE
+
+    # Waiting: empty store, single sample, or fresh data but not yet enough evidence
+    if sample_count == 0:
+        return StatusState.WAITING
+    if sample_count <= 1 and day_count == 0:
+        return StatusState.WAITING
+
+    # Monitoring: everything is fine
+    return StatusState.MONITORING
+
+
+def _determine_explanation(
+    state: StatusState,
+    freshness: str,
+    freshness_age_s: Optional[int],
+    last_collect_ok: Optional[bool],
+    last_collect_reason: Optional[str],
+    sample_count: int,
+    deliberately_paused: bool,
+    store_fault: Optional[str],
+    newer_schema: Optional[str],
+) -> str:
+    """Determine the explanation line for the status state."""
+    if store_fault:
+        return "observation store unreadable — see journal"
+
+    if newer_schema:
+        return "observation store written by a newer Fenris — upgrade Fenris"
+
+    if state == StatusState.ERROR:
+        parts = []
+        if last_collect_ok is False:
+            parts.append("last run failed")
+            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))
+        return " ".join(parts) if parts else "last run failed"
+
+    if state == StatusState.INTERRUPTED:
+        return "collection stopped outside Fenris — monitoring period still open"
+
+    if state == StatusState.PAUSED:
+        return "monitoring paused — paused time excluded from your usage habit"
+
+    if state == StatusState.STALE:
+        if freshness_age_s is not None:
+            return "last sample %s ago" % freshness_age_human(freshness_age_s)
+        return "data is stale"
+
+    if state == StatusState.WAITING:
+        if sample_count == 0:
+            return "awaiting first sample"
+        if sample_count <= 1:
+            return "awaiting another sample"
+        return "waiting for data"
+
+    if state == StatusState.MONITORING:
+        if freshness_age_s is not None:
+            return "last sample %s ago" % freshness_age_human(freshness_age_s)
+        return "monitoring active"
+
+    if state == StatusState.UNKNOWN:
+        return "service state unavailable"
+
+    return ""
+
+
+def _determine_collecting_overlay(
+    base_state: StatusState,
+    last_collect_ok: Optional[bool],
+    deliberately_paused: bool,
+    store_fault: Optional[str],
+    newer_schema: Optional[str],
+) -> str:
+    """Determine the explanation line when Collecting overlays a base state."""
+    if store_fault or newer_schema:
+        return "run in flight — store fault"
+
+    if base_state == StatusState.PAUSED:
+        return "run in flight — paused"
+    if base_state == StatusState.INTERRUPTED:
+        return "run in flight — interrupted"
+    if base_state == StatusState.ERROR:
+        return "run in flight — retry"
+    if base_state == StatusState.STALE:
+        return "run in flight — stale data"
+    if base_state == StatusState.WAITING:
+        return "run in flight"
+    return "run in flight"
+
+
+def compose_status(
+    conn: sqlite3.Connection,
+    service: Optional[Dict[str, Any]],
+    clock_now: datetime,
+    store_fault: Optional[str] = None,
+    newer_schema: Optional[str] = None,
+    collecting: bool = False,
+    reduced_motion: bool = False,
+) -> StatusComposition:
+    """Compose the shared status from service state and store data.
+
+    This is the single entry point consumed by both TUI and CLI.
+    """
+    service_available = service is not None and len(service) > 0
+
+    # --- 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
+
+    # --- Freshness from store ---
+    freshness = "unknown"
+    freshness_age_s = None
+    sample_count = 0
+    day_count = 0
+    deliberately_paused = False
+
+    if 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()
+            newest_ts = row[0] if row else None
+            freshness = grade_freshness(newest_ts, clock_now)
+            if newest_ts:
+                try:
+                    ts = datetime.fromisoformat(newest_ts)
+                    if ts.tzinfo is None:
+                        ts = ts.replace(tzinfo=timezone.utc)
+                    else:
+                        ts = ts.astimezone(timezone.utc)
+                    freshness_age_s = int((clock_now - ts).total_seconds())
+                except (ValueError, TypeError):
+                    pass
+        except sqlite3.Error:
+            freshness = "unknown"
+
+        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
+
+        try:
+            svc_for_pause = service if service_available else {}
+            deliberately_paused = is_deliberately_paused(conn, svc_for_pause)
+        except sqlite3.Error:
+            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:
+        # Check if there's an open monitoring period (external stop left it open)
+        try:
+            open_period = conn.execute(
+                "SELECT 1 FROM monitoring_periods WHERE ended_at IS NULL LIMIT 1"
+            ).fetchone()
+            if open_period is not None:
+                external_stop_reason = "external_stop"
+        except sqlite3.Error:
+            pass
+
+    # --- Determine base state ---
+    # Store fault and newer schema always override to ERROR
+    if store_fault is not None or newer_schema is not None:
+        base_state = StatusState.ERROR
+    else:
+        base_state = _determine_base_state(
+            freshness=freshness,
+            sample_count=sample_count,
+            day_count=day_count,
+            boot_enabled=boot_enabled,
+            timer_active=timer_active,
+            last_collect_ok=last_collect_ok,
+            deliberately_paused=deliberately_paused,
+            external_stop_reason=external_stop_reason,
+            service_available=service_available,
+        )
+
+    # --- Apply Collecting overlay ---
+    state = base_state
+    overlay_base = None
+    explanation = ""
+
+    if collecting and store_fault is None and newer_schema is None:
+        # Collecting overlays every base except store fault
+        overlay_base = base_state
+        state = StatusState.COLLECTING
+        explanation = _determine_collecting_overlay(
+            base_state, last_collect_ok, deliberately_paused,
+            store_fault, newer_schema,
+        )
+    else:
+        explanation = _determine_explanation(
+            base_state, freshness, freshness_age_s,
+            last_collect_ok, last_collect_reason,
+            sample_count, deliberately_paused,
+            store_fault, newer_schema,
+        )
+
+    # --- Continuity line ---
+    continuity = ""
+    if service_available:
+        continuity = monitoring_continuity(service)
+
+    # --- Paused lines ---
+    paused_lines = []
+    if deliberately_paused:
+        paused_lines = deliberate_pause_lines()
+
+    # Determine blink flag: only Monitoring dot blinks, never text or other states
+    should_blink = (state == StatusState.MONITORING and not reduced_motion)
+
+    return StatusComposition(
+        state=state,
+        glyph=state.glyph,
+        label=state.label,
+        explanation=explanation,
+        should_blink=should_blink,
+        freshness=freshness,
+        freshness_age_s=freshness_age_s,
+        last_collect_ok=last_collect_ok,
+        last_collect_age_s=last_collect_age_s,
+        last_collect_reason=last_collect_reason,
+        boot_enabled=boot_enabled,
+        timer_active=timer_active,
+        deliberately_paused=deliberately_paused,
+        external_stop_reason=external_stop_reason,
+        store_fault=store_fault,
+        newer_schema=newer_schema,
+        overlay_base=overlay_base,
+        sample_count=sample_count,
+        day_count=day_count,
+        continuity=continuity,
+        paused_lines=paused_lines,
+    )
+
+
+# ---------------------------------------------------------------------------
+# CLI rendering (static, no styling)
+# ---------------------------------------------------------------------------
+
+def render_status_cli(comp: StatusComposition) -> str:
+    """Render the status composition as static CLI text."""
+    lines = []
+
+    # Status line
+    lines.append("%s %s" % (comp.glyph, comp.label))
+
+    # Explanation
+    if comp.explanation:
+        lines.append(comp.explanation)
+
+    lines.append("")
+
+    # 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)
+    if comp.last_collect_ok is True:
+        facts.append("last collect: ok")
+    elif comp.last_collect_ok is False:
+        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"))
+
+    if facts:
+        lines.append(" · ".join(facts))
+
+    # Continuity
+    if comp.continuity:
+        lines.append("")
+        lines.append("CONTINUITY: %s" % comp.continuity)
+
+    # Deliberate pause
+    if comp.paused_lines:
+        for pl in comp.paused_lines:
+            lines.append(pl)
+
+    return "\n".join(lines)
+
+
+# ---------------------------------------------------------------------------
+# TUI rendering (with styling tokens)
+# ---------------------------------------------------------------------------
+
+def render_status_tui(comp: StatusComposition) -> str:
+    """Render the status composition as TUI text with Textual markup."""
+    lines = []
+
+    # Status line with color
+    color = {
+        StatusState.MONITORING: "green",
+        StatusState.COLLECTING: "green",
+        StatusState.PAUSED: "yellow",
+        StatusState.WAITING: "yellow",
+        StatusState.INTERRUPTED: "red",
+        StatusState.ERROR: "red",
+        StatusState.STALE: "red",
+        StatusState.UNKNOWN: "dim",
+    }[comp.state]
+
+    lines.append("[%s]%s %s[/%s]" % (color, comp.glyph, comp.label, color))
+
+    # Explanation
+    if comp.explanation:
+        lines.append(comp.explanation)
+
+    lines.append("")
+
+    # Separate facts
+    facts = []
+    if comp.freshness != "unknown":
+        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:
+        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"))
+
+    if facts:
+        lines.append(" · ".join(facts))
+
+    # Continuity
+    if comp.continuity:
+        lines.append("")
+        lines.append("[bold]CONTINUITY[/bold]  %s" % comp.continuity)
+
+    # Deliberate pause
+    if comp.paused_lines:
+        for pl in comp.paused_lines:
+            lines.append(pl)
+
+    return "\n".join(lines)

src/fenris/tui.py

@@ -52,6 +52,13 @@ from .status import (
     read_config,
 )
 from .monitoring_periods import get_open_period
+from .status_composition import (
+    StatusState,
+    StatusComposition,
+    compose_status,
+    render_status_tui,
+    STATUS_POLL_INTERVAL_S,
+)
 
 
 # ---------------------------------------------------------------------------
@@ -1111,22 +1118,23 @@ class FenrisTuiApp(App):
         # --- Service strip (§7.2 bottom) ---
         try:
             svc = _query_service_facts(conn, self._clock_now)
-            boot = "enabled" if svc.get("boot_enabled") else "disabled"
+            # Use shared status composition
-            activity = "active" if svc.get("timer_active") else "inactive"
+            comp = compose_status(
-            collect = "ok" if svc.get("last_collect_ok") else "FAILED"
+                conn, svc, self._clock_now,
-            freshness = svc.get("freshness", "unknown")
+                store_fault=None,
+                newer_schema=None,
+                reduced_motion=False,
+            )
+            # Render using the shared composition with TUI styling
+            status_text = render_status_tui(comp)
+            # Add TUI-only branding and actions
             self.query_one("#service-strip").update(
-                "boot: %s · timer: %s · last collect: %s · freshness: %s · "
-                "[dim]by Bongbetic[/dim]\n"
-                "[bold]CONTINUITY[/bold]  %s\n"
                 "%s\n"
+                "[dim]by Bongbetic[/dim]\n"
                 "p pause · r resume · c collect · d disclosures"
-                % (
+                % status_text
-                    boot, activity, collect, freshness,
-                    monitoring_continuity(svc), svc.get("period", ""),
             )
-            )
+            self._render_paused_banner(comp)
-            self._render_paused_banner(svc)
         except Exception:
             self._hide_paused_banner()
             self.query_one("#service-strip").update(
@@ -1135,13 +1143,24 @@ class FenrisTuiApp(App):
                 "p pause · r resume · c collect · d disclosures"
             )
 
-    def _render_paused_banner(self, service: Dict[str, Any]) -> None:
+    def _render_paused_banner(self, comp) -> None:
-        """Show the high-contrast Deliberate disable block only when sanctioned."""
+        """Show the high-contrast Deliberate disable block only when sanctioned.
+
+        Accepts either a StatusComposition or a dict for backward compatibility.
+        """
         banner = self.query_one("#paused-banner")
-        if service.get("deliberately_paused"):
+        # Support both StatusComposition and legacy dict
+        if hasattr(comp, 'deliberately_paused'):
+            is_paused = comp.deliberately_paused
+            paused_lines = comp.paused_lines
+        else:
+            is_paused = comp.get("deliberately_paused", False)
+            paused_lines = deliberate_pause_lines() if is_paused else []
+
+        if is_paused and paused_lines:
             banner.update(
                 "[bold black on red]%s[/bold black on red]\n%s"
-                % tuple(deliberate_pause_lines())
+                % tuple(paused_lines)
             )
             banner.styles.display = "block"
             main_grid = self.query_one("#main-grid")