bu5hm4nn/vault-dash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(CORE-001D3B): surface alert history degraded state

Browse Source
This commit is contained in:
Bu5hm4nn
2026-03-26 15:12:04 +01:00
parent 09e03f96a8
commit ff76e326b1
8 changed files with 138 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
app/models/alerts.py
View File

@@ -8,6 +8,12 @@ from pathlib import Path
from typing import Any
class AlertHistoryLoadError(RuntimeError):
def __init__(self, history_path: Path, message: str) -> None:
super().__init__(message)
self.history_path = history_path
@dataclass
class AlertEvent:
severity: str
@@ -36,6 +42,8 @@ class AlertStatus:
critical_threshold: float
email_alerts_enabled: bool
history: list[AlertEvent]
history_unavailable: bool = False
history_notice: str | None = None
class AlertHistoryRepository:
@@ -53,10 +61,12 @@ class AlertHistoryRepository:
try:
with self.history_path.open() as f:
data = json.load(f)
except (json.JSONDecodeError, OSError):
return []
except json.JSONDecodeError as exc:
raise AlertHistoryLoadError(self.history_path, f"Alert history is not valid JSON: {exc}") from exc
except OSError as exc:
raise AlertHistoryLoadError(self.history_path, f"Alert history could not be read: {exc}") from exc
if not isinstance(data, list):
return []
raise AlertHistoryLoadError(self.history_path, "Alert history payload must be a list")
return [AlertEvent.from_dict(item) for item in data if isinstance(item, dict)]
def save(self, events: list[AlertEvent]) -> None:

8
app/pages/overview.py
View File

@@ -170,6 +170,8 @@ async def overview_page(workspace_id: str) -> None:
f"Warning at {alert_status.warning_threshold:.0%} · Critical at {alert_status.critical_threshold:.0%} · "
f"Email alerts {'enabled' if alert_status.email_alerts_enabled else 'disabled'}"
).classes("text-sm text-slate-500 dark:text-slate-400")
if alert_status.history_notice:
ui.label(alert_status.history_notice).classes("text-sm text-amber-700 dark:text-amber-300")
if alert_status.history:
latest = alert_status.history[0]
ui.label(
@@ -211,9 +213,7 @@ async def overview_page(workspace_id: str) -> None:
ui.label(value).classes("text-2xl font-bold text-slate-900 dark:text-slate-50")
ui.label(caption).classes("text-sm text-slate-500 dark:text-slate-400")
with ui.card().classes(
f"w-full rounded-2xl border shadow-sm {recommendation_style('info')}"
):
with ui.card().classes(f"w-full rounded-2xl border shadow-sm {recommendation_style('info')}"):
ui.label("Quick Strategy Recommendations").classes(
"text-lg font-semibold text-slate-900 dark:text-slate-100"
)
@@ -265,6 +265,8 @@ async def overview_page(workspace_id: str) -> None:
f"Logged {_format_timestamp(event.updated_at)} · Spot ${event.spot_price:,.2f} · LTV {event.ltv_ratio:.1%}"
).classes("text-xs text-slate-500 dark:text-slate-400")
ui.label(event.severity.upper()).classes(_alert_badge_classes(event.severity))
elif alert_status.history_notice:
ui.label(alert_status.history_notice).classes("text-sm text-amber-700 dark:text-amber-300")
else:
ui.label(
"No alert history yet. Alerts will be logged once the warning threshold is crossed."

5
app/pages/settings.py
View File

@@ -352,7 +352,10 @@ def settings_page(workspace_id: str) -> None:
alert_message.set_text(alert_status.message)
status.set_text(_save_card_status_text(last_saved_config, preview_config=preview_config))
alert_history_column.clear()
if alert_status.history:
if alert_status.history_notice:
with alert_history_column:
ui.label(alert_status.history_notice).classes("text-sm text-amber-700 dark:text-amber-300")
elif alert_status.history:
for event in alert_status.history[:5]:
with alert_history_column:
with ui.row().classes(

37
app/services/alerts.py
View File

@@ -2,15 +2,18 @@
from __future__ import annotations
import logging
from dataclasses import dataclass
from decimal import Decimal
from typing import Mapping
from app.domain.portfolio_math import build_alert_context
from app.models.alerts import AlertEvent, AlertHistoryRepository, AlertStatus
from app.models.alerts import AlertEvent, AlertHistoryLoadError, AlertHistoryRepository, AlertStatus
from app.models.portfolio import PortfolioConfig
from app.services.boundary_values import boundary_decimal
logger = logging.getLogger(__name__)
@dataclass(frozen=True, slots=True)
class AlertEvaluationInput:
@@ -74,7 +77,18 @@ class AlertService:
def evaluate(
self, config: PortfolioConfig, portfolio: Mapping[str, object], *, persist: bool = True
) -> AlertStatus:
history = self.repository.load() if persist else []
history: list[AlertEvent] = []
history_unavailable = False
history_notice: str | None = None
try:
history = self.repository.load()
except AlertHistoryLoadError as exc:
history_unavailable = True
history_notice = (
"Alert history is temporarily unavailable due to a storage error. New alerts are not being recorded."
)
logger.warning("Alert history unavailable at %s: %s", exc.history_path, exc)
evaluation = _normalize_alert_evaluation_input(config, portfolio)
if evaluation.ltv_ratio >= evaluation.critical_threshold:
@@ -106,12 +120,21 @@ class AlertService:
email_alerts_enabled=evaluation.email_alerts_enabled,
)
if persist:
if self._should_record(history, event):
if not history_unavailable and self._should_record(history, event):
history.append(event)
self.repository.save(history)
else:
preview_history = [event]
if not persist:
resolved_history = preview_history
elif history_unavailable:
resolved_history = []
elif severity != "ok":
resolved_history = list(reversed(self.repository.load()))
else:
resolved_history = history
return AlertStatus(
severity=severity,
message=message,
@@ -119,11 +142,9 @@ class AlertService:
warning_threshold=float(evaluation.warning_threshold),
critical_threshold=float(evaluation.critical_threshold),
email_alerts_enabled=evaluation.email_alerts_enabled,
history=(
preview_history
if not persist
else list(reversed(self.repository.load() if severity != "ok" else history))
),
history=resolved_history,
history_unavailable=history_unavailable,
history_notice=history_notice,
)
@staticmethod

2
docs/roadmap/ROADMAP.yaml
View File

@@ -25,6 +25,7 @@ priority_queue:
- OPS-001
- BT-003
recently_completed:
- CORE-001D3B
- CORE-001D3A
- UX-001
- CORE-002
@@ -66,6 +67,7 @@ states:
- CORE-001D2A
- CORE-001D2B
- CORE-001D3A
- CORE-001D3B
- CORE-002
- CORE-002A
- CORE-002B

1
docs/roadmap/backlog/CORE-001D-decimal-boundary-cleanup.yaml
View File

@@ -18,6 +18,7 @@ technical_notes:
- `CORE-001D2A` is complete: DataService quote/provider cache normalization is now a named boundary adapter with explicit symbol mismatch rejection and GLD quote-unit repair.
- `CORE-001D2B` is complete: option expirations and options-chain payloads now use explicit normalization boundaries with malformed cached payload discard/retry behavior.
- `CORE-001D3A` is complete: alert evaluation and settings save-status entrypoints now normalize float-heavy boundary values through explicit named adapters.
- `CORE-001D3B` is complete: corrupt alert-history storage now surfaces as an explicit degraded state with logging and route-visible notices instead of silently appearing as empty history.
- Remaining focus is the rest of `CORE-001D2` provider/cache normalization plus follow-on `CORE-001D3` service entrypoint tightening.
- Pre-launch policy: unit-aware schema changes may be breaking until persistence is considered live; old flat payloads may fail loudly instead of being migrated.
- See `docs/CORE-001D_BOUNDARY_CLEANUP_PLAN.md` for the current hotspot inventory and proposed sub-slices.

19
docs/roadmap/done/CORE-001D3B-alert-history-degraded-state.yaml Normal file
View File

@@ -0,0 +1,19 @@
id: CORE-001D3B
title: Alert History Degraded State Handling
status: done
priority: P2
effort: S
depends_on:
- CORE-001D3A
tags:
- core
- alerts
- persistence
- ux
summary: Corrupt or unreadable alert-history storage now surfaces as an explicit degraded state instead of silently appearing as empty history.
completed_notes:
- Added `AlertHistoryLoadError` in `app/models/alerts.py` so corrupt or unreadable alert-history storage is an explicit failure mode.
- `AlertService.evaluate(...)` now logs history load failures and returns `history_unavailable` / `history_notice` metadata instead of silently treating corruption as an empty history list.
- `/overview` and `/{workspace_id}/settings` now render a visible degraded-history notice when alert history storage is unavailable.
- Added focused regression coverage in `tests/test_alerts.py` for corrupt-history load failures in both persisted and preview paths.
- Validated with focused pytest coverage, local Docker, and browser-driven checks on overview and settings with an intentionally corrupted `/app/data/alert_history.json`.

65
tests/test_alerts.py
View File

@@ -5,6 +5,7 @@ from pathlib import Path
import pytest
from app.models.alerts import AlertHistoryLoadError, AlertHistoryRepository
from app.models.portfolio import PortfolioConfig
from app.services.alerts import AlertService, _normalize_alert_evaluation_input, build_portfolio_alert_context
@@ -169,6 +170,70 @@ def test_alert_service_escalates_to_critical_and_keeps_history(alert_service: Al
assert critical_status.history[0].ltv_ratio == pytest.approx(0.7525, rel=1e-6)
def test_alert_history_repository_raises_on_corrupt_history_file(tmp_path: Path) -> None:
history_path = tmp_path / "alert_history.json"
history_path.write_text("{not valid json", encoding="utf-8")
repository = AlertHistoryRepository(history_path=history_path)
with pytest.raises(AlertHistoryLoadError, match="Alert history is not valid JSON"):
repository.load()
def test_alert_service_marks_history_unavailable_on_corrupt_storage(alert_service: AlertService) -> None:
alert_service.repository.history_path.write_text("{not valid json", encoding="utf-8")
config = PortfolioConfig(
gold_value=215_000.0,
entry_price=215.0,
loan_amount=151_000.0,
ltv_warning=0.70,
margin_threshold=0.75,
)
portfolio = build_portfolio_alert_context(
config,
spot_price=215.0,
source="test",
updated_at="2026-03-24T12:00:00Z",
)
status = alert_service.evaluate(config, portfolio)
assert status.severity == "warning"
assert status.history == []
assert status.history_unavailable is True
assert (
status.history_notice
== "Alert history is temporarily unavailable due to a storage error. New alerts are not being recorded."
)
assert alert_service.repository.history_path.read_text(encoding="utf-8") == "{not valid json"
def test_alert_service_preview_marks_history_unavailable_on_corrupt_storage(alert_service: AlertService) -> None:
alert_service.repository.history_path.write_text("{not valid json", encoding="utf-8")
config = PortfolioConfig(
gold_value=215_000.0,
entry_price=215.0,
loan_amount=151_000.0,
ltv_warning=0.70,
margin_threshold=0.75,
)
portfolio = build_portfolio_alert_context(
config,
spot_price=215.0,
source="settings-preview",
updated_at="",
)
status = alert_service.evaluate(config, portfolio, persist=False)
assert status.severity == "warning"
assert [event.severity for event in status.history] == ["warning"]
assert status.history_unavailable is True
assert (
status.history_notice
== "Alert history is temporarily unavailable due to a storage error. New alerts are not being recorded."
)
def test_alert_service_preserves_persisted_history_during_ok_evaluation(alert_service: AlertService) -> None:
warning_config = PortfolioConfig(
gold_value=215_000.0,