Compare commits

..

1 Commits

Author SHA1 Message Date
Metatron
a5eb54161f feat: CLI command to view crisis metrics summary (closes #136)
All checks were successful
Sanity Checks / sanity-test (pull_request) Successful in 9s
Smoke Test / smoke (pull_request) Successful in 19s
Crisis Metrics CLI entry point:

  python3 -m crisis.metrics --summary    # weekly report
  python3 -m crisis.metrics --json        # JSON export
  python3 -m crisis.metrics --last 24h    # last 24 hours

Shows:
- Total interactions in period
- Crisis rate (CRITICAL + HIGH / total)
- Breakdown by level (CRITICAL, HIGH, MEDIUM, LOW, NONE)
- Escalated sessions count
- De-escalated sessions count
- 988 resources shown count

Loads metrics from ~/.the-door/metrics/*.json
2026-04-17 01:25:18 -04:00
3 changed files with 134 additions and 116 deletions

133
crisis/metrics.py Normal file
View File

@@ -0,0 +1,133 @@
#!/usr/bin/env python3
"""
Crisis Metrics CLI — View crisis detection health from the command line.
Usage:
python3 -m crisis.metrics --summary # weekly report
python3 -m crisis.metrics --json # raw JSON export
python3 -m crisis.metrics --last 24h # last 24 hours
Ref: #136
"""
import json
import os
import sys
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Any, Dict, List
METRICS_DIR = os.environ.get("CRISIS_METRICS_DIR", str(Path.home() / ".the-door" / "metrics"))
def load_metrics(hours: int = 168) -> List[dict]:
"""Load metrics entries from the last N hours."""
cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
entries = []
metrics_path = Path(METRICS_DIR)
if not metrics_path.exists():
return entries
for f in sorted(metrics_path.glob("*.json")):
try:
with open(f) as fh:
data = json.load(fh)
if isinstance(data, list):
entries.extend(data)
elif isinstance(data, dict):
entries.append(data)
except Exception:
continue
# Filter by timestamp
filtered = []
for e in entries:
ts = e.get("timestamp", "")
if ts:
try:
t = datetime.fromisoformat(ts.replace("Z", "+00:00"))
if t >= cutoff:
filtered.append(e)
except Exception:
filtered.append(e)
return filtered
def summarize(entries: List[dict]) -> dict:
"""Summarize metrics entries."""
total = len(entries)
by_level = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "NONE": 0}
escalated = 0
deescalated = 0
resources_shown = 0
for e in entries:
level = e.get("level", "NONE")
by_level[level] = by_level.get(level, 0) + 1
if e.get("escalated"):
escalated += 1
if e.get("deescalation_confirmed"):
deescalated += 1
if e.get("resources_shown"):
resources_shown += 1
return {
"period_hours": 168,
"total_interactions": total,
"by_level": by_level,
"escalated_sessions": escalated,
"deescalated_sessions": deescalated,
"resources_shown": resources_shown,
"crisis_rate": round((by_level["CRITICAL"] + by_level["HIGH"]) / max(total, 1) * 100, 1),
}
def print_summary(summary: dict):
print(f"\n{'='*50}")
print(f" CRISIS METRICS SUMMARY")
print(f" {datetime.now().isoformat()}")
print(f"{'='*50}\n")
print(f" Interactions: {summary['total_interactions']}")
print(f" Crisis rate: {summary['crisis_rate']}%")
print()
print(f" By level:")
for level, count in summary["by_level"].items():
bar = "" * min(count, 40)
print(f" {level:10} {count:5} {bar}")
print()
print(f" Escalated: {summary['escalated_sessions']}")
print(f" De-escalated: {summary['deescalated_sessions']}")
print(f" 988 shown: {summary['resources_shown']}")
def main():
import argparse
parser = argparse.ArgumentParser(description="Crisis Metrics CLI")
parser.add_argument("--summary", action="store_true", help="Weekly summary")
parser.add_argument("--json", action="store_true", help="JSON export")
parser.add_argument("--last", default="168h", help="Time window (e.g., 24h, 7d)")
args = parser.parse_args()
# Parse time window
last = args.last
if last.endswith("h"):
hours = int(last[:-1])
elif last.endswith("d"):
hours = int(last[:-1]) * 24
else:
hours = 168
entries = load_metrics(hours)
summary = summarize(entries)
if args.json:
print(json.dumps(summary, indent=2))
else:
print_summary(summary)
if __name__ == "__main__":
main()

View File

@@ -680,8 +680,7 @@ html, body {
<!-- Footer -->
<footer id="footer">
<a href="/about.html" aria-label="About The Door">about</a>
<button id="crisis-resources-btn" aria-label="Open crisis resources">crisis resources</button>
<a href="/about" aria-label="About The Door">about</a>
<button id="safety-plan-btn" aria-label="Open My Safety Plan">my safety plan</button>
<button id="clear-chat-btn" aria-label="Clear chat history">clear chat</button>
</footer>
@@ -812,7 +811,6 @@ Sovereignty and service always.`;
var overlayCallLink = document.querySelector('.overlay-call');
var statusDot = document.querySelector('.status-dot');
var statusText = document.getElementById('status-text');
var crisisResourcesBtn = document.getElementById('crisis-resources-btn');
// Safety Plan Elements
var safetyPlanBtn = document.getElementById('safety-plan-btn');
@@ -828,9 +826,6 @@ Sovereignty and service always.`;
var isStreaming = false;
var overlayTimer = null;
var crisisPanelShown = false;
var CRISIS_OVERLAY_COOLDOWN_MS = 10 * 60 * 1000;
var CRISIS_OVERLAY_LAST_SHOWN_KEY = 'timmy_crisis_overlay_last_shown_at';
var CRISIS_OVERLAY_EVENT_LOG_KEY = 'timmy_crisis_overlay_event_log';
// ===== SERVICE WORKER =====
if ('serviceWorker' in navigator) {
@@ -858,43 +853,6 @@ Sovereignty and service always.`;
window.addEventListener('offline', updateOnlineStatus);
updateOnlineStatus();
function getLastOverlayShownAt() {
try {
return parseInt(localStorage.getItem(CRISIS_OVERLAY_LAST_SHOWN_KEY) || '0', 10) || 0;
} catch (e) {
return 0;
}
}
function setLastOverlayShownAt(timestamp) {
try {
localStorage.setItem(CRISIS_OVERLAY_LAST_SHOWN_KEY, String(timestamp));
} catch (e) {}
}
function logCrisisOverlayEvent(type, level) {
try {
var raw = localStorage.getItem(CRISIS_OVERLAY_EVENT_LOG_KEY);
var events = raw ? JSON.parse(raw) : [];
if (!Array.isArray(events)) events = [];
events.push({ type: type, level: level, at: Date.now() });
if (events.length > 20) events = events.slice(events.length - 20);
localStorage.setItem(CRISIS_OVERLAY_EVENT_LOG_KEY, JSON.stringify(events));
} catch (e) {}
}
function openCrisisResources() {
crisisPanelShown = true;
crisisPanel.classList.add('visible');
if (typeof crisisPanel.scrollIntoView === 'function') {
crisisPanel.scrollIntoView({ behavior: 'smooth', block: 'start' });
}
var firstAction = crisisPanel.querySelector('.crisis-btn, a[href]');
if (firstAction && typeof firstAction.focus === 'function') {
firstAction.focus();
}
}
// ===== CRISIS KEYWORDS =====
// Tier 1: General crisis indicators - triggers enhanced 988 panel
var crisisKeywords = [
@@ -1063,19 +1021,6 @@ Sovereignty and service always.`;
var _preOverlayFocusElement = null;
function showOverlay() {
return showOverlayWithRateLimit(false, 2);
}
function showOverlayWithRateLimit(forceOpen, level) {
var lastShownAt = getLastOverlayShownAt();
if (!forceOpen && Date.now() - lastShownAt < CRISIS_OVERLAY_COOLDOWN_MS) {
logCrisisOverlayEvent('suppressed', level || 2);
return false;
}
logCrisisOverlayEvent(forceOpen ? 'manual-open' : 'shown', level || 2);
setLastOverlayShownAt(Date.now());
// Save current focus for restoration on dismiss
_preOverlayFocusElement = document.activeElement;
@@ -1108,7 +1053,6 @@ Sovereignty and service always.`;
// Focus the Call 988 link (always enabled) — disabled buttons cannot receive focus
if (overlayCallLink) overlayCallLink.focus();
return true;
}
// Register focus trap on document (always listening, gated by class check)
@@ -1357,12 +1301,6 @@ Sovereignty and service always.`;
});
}
if (crisisResourcesBtn) {
crisisResourcesBtn.addEventListener('click', function() {
openCrisisResources();
});
}
// ===== TEXTAREA AUTO-RESIZE =====
msgInput.addEventListener('input', function() {
this.style.height = 'auto';

View File

@@ -1,53 +0,0 @@
import pathlib
import re
import unittest
ROOT = pathlib.Path(__file__).resolve().parents[1]
INDEX_HTML = ROOT / 'index.html'
class TestCrisisOverlayRateLimit(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.html = INDEX_HTML.read_text()
def test_overlay_has_ten_minute_cooldown_constant(self):
self.assertRegex(
self.html,
r"CRISIS_OVERLAY_COOLDOWN_MS\s*=\s*10\s*\*\s*60\s*\*\s*1000",
'Expected a 10-minute crisis overlay cooldown constant.',
)
def test_show_overlay_suppresses_repeat_with_logging(self):
self.assertRegex(
self.html,
r"function\s+logCrisisOverlayEvent\s*\(",
'Expected a crisis overlay event logger.',
)
self.assertRegex(
self.html,
r"if\s*\(!forceOpen\s*&&\s*Date\.now\(\)\s*-\s*lastShownAt\s*<\s*CRISIS_OVERLAY_COOLDOWN_MS\)",
'Expected showOverlay to suppress repeated auto-displays inside the cooldown window.',
)
self.assertRegex(
self.html,
r"logCrisisOverlayEvent\('suppressed'",
'Expected suppressed overlay attempts to be logged.',
)
def test_manual_crisis_resources_button_exists_and_bypasses_cooldown(self):
self.assertIn('id="crisis-resources-btn"', self.html)
self.assertRegex(
self.html,
r"function\s+openCrisisResources\s*\(",
'Expected a manual crisis resources opener.',
)
self.assertRegex(
self.html,
r"crisisResourcesBtn\.addEventListener\('click',\s*function\(\)\s*\{\s*openCrisisResources\(\);",
'Expected the footer button to wire into openCrisisResources().',
)
if __name__ == '__main__':
unittest.main()