diff --git a/adhdo2/E2E_CHECKLIST.md b/adhdo2/E2E_CHECKLIST.md index a0f2ce3..f155df3 100644 --- a/adhdo2/E2E_CHECKLIST.md +++ b/adhdo2/E2E_CHECKLIST.md @@ -1,5 +1,35 @@ # ADHDo 2.0 P1 Hardware E2E (pi5-hailo) +> How boot catch-up interacts with the dispatcher and fixed-event schedule +> (and why they never double-fire) is documented in +> [docs/catchup-vs-schedule.md](docs/catchup-vs-schedule.md). +> +> Note (P2 quality pass): the heartbeat is no longer a `*/25` cron line — +> `install-cron.sh` now installs the `adhdo-heartbeat.timer` systemd user +> timer (true 25-min cadence). On-device step: re-run +> `~/adhdo2/scripts/install-cron.sh` after the next deploy so the old cron +> heartbeat line is replaced and the timer is enabled. + +## P3 Telegram bridge — on-device setup (human steps) + +1. Create a bot with @BotFather on Telegram; copy the bot token. +2. On the Pi, put the token in `~/adhdo2/config.yaml` under + `telegram.bot_token` (or export `TELEGRAM_BOT_TOKEN` in the service + environment — env overrides config). +3. Message the bot once from your own Telegram account, then find your + numeric chat ID (e.g. via `curl "https://api.telegram.org/bot/getUpdates"` + → `message.chat.id`) and add it to `telegram.chat_id_allowlist` in + `~/adhdo2/config.yaml`. +4. `cp ~/adhdo2/systemd/adhdo-telegram.service ~/.config/systemd/user/ && + systemctl --user daemon-reload && systemctl --user enable --now adhdo-telegram` +5. Verify: + - [ ] Text the bot "hello" → appears in the adhdo tmux pane as + `[telegram chat:] User says: "hello"` and Claude responds. + - [ ] `bin/adhdo-telegram send "test reply"` → arrives in Telegram. + - [ ] Message from a non-allowlisted chat → ignored, `error` row in journal. + - [ ] Send a photo → bot replies "Text messages only"; nothing injected. + - [ ] 7 rapid messages in a minute → later ones get a rate-limit reply. + Run 2026-07-10, automated portion only (no audible verification performed — that requires a human physically listening at the Pi's location). @@ -81,6 +111,12 @@ that requires a human physically listening at the Pi's location). SKIPPED — explicitly instructed not to reboot the Pi during this automated pass. +- [ ] Nightly rollup: after re-running `install-cron.sh` (adds a 03:15 + `adhdo-rollup.sh` entry), verify next morning that `journal rollup` + wrote rows (`sqlite3 ~/adhdo2/data/journal.db "SELECT * FROM rollups"`) + and `~/adhdo2/data/backup/journal.db` exists. Depends on cron being + enabled (human-gated switch below). + - [ ] Full heartbeat observed: cron fires, session runs state, journals a decision SKIPPED — install-cron.sh intentionally NOT run yet (final human-gated switch, see Step 4 notes). No heartbeat cron installed, so nothing to @@ -197,3 +233,13 @@ installed as a safety net per Step 1). `media_controller.status` until idle, max 30s) was NOT implemented — it requires audible hardware observation to tune correctly, which is a human step. +8. P3 dashboard (`bin/adhdo-dashboard`, port 8766): on-device verification — + `systemctl --user enable --now adhdo-dashboard` (unit in + `systemd/adhdo-dashboard.service`), then open + `http://:8766/` from another LAN machine and confirm the page + shows live state (devices, disk, last-event ages) and recent journal rows, + auto-refreshing every 30s. Confirm `curl -X POST http://:8766/` + returns 405 (read-only) and that the service is NOT reachable from outside + the LAN (binds to `dashboard.bind`, defaulting to `lan_ip`). Contract + tests (routing, JSON schemas, secret scrubbing, method rejection) are + covered in `tests/test_dashboard.py`. diff --git a/adhdo2/adhdolib/config.py b/adhdo2/adhdolib/config.py index aed9374..faf4a33 100644 --- a/adhdo2/adhdolib/config.py +++ b/adhdo2/adhdolib/config.py @@ -15,8 +15,13 @@ "high_urgency_events": ["medication", "safety", "user_requested"], }, "disk_warn_pct": 94, + "dashboard": {"bind": None, "port": 8766}, "crisis": {"contacts": ["Lifeline Australia 13 11 14"]}, - "telegram": {"chat_id_allowlist": []}, + "telegram": { + "chat_id_allowlist": [], + "bot_token": None, + "rate_limit_per_minute": 6, + }, } def home() -> Path: diff --git a/adhdo2/adhdolib/envelope.py b/adhdo2/adhdolib/envelope.py index df604de..850dbc9 100644 --- a/adhdo2/adhdolib/envelope.py +++ b/adhdo2/adhdolib/envelope.py @@ -35,9 +35,9 @@ def cli_main(fn): except ToolError as e: print(json.dumps({"error": e.code, "detail": scrub(e.detail)})) sys.exit(1) - except SystemExit: - raise - except BaseException as e: + except Exception as e: + # Deliberately Exception, not BaseException: KeyboardInterrupt and + # SystemExit must propagate so Ctrl-C / exit codes behave normally. tail = "".join(traceback.format_tb(e.__traceback__)[-3:]) print(json.dumps({"error": "crash", "detail": scrub(f"{type(e).__name__}: {e}"), diff --git a/adhdo2/adhdolib/sanitize.py b/adhdo2/adhdolib/sanitize.py index d47f89d..d49fe75 100644 --- a/adhdo2/adhdolib/sanitize.py +++ b/adhdo2/adhdolib/sanitize.py @@ -1,3 +1,11 @@ +def redact(text: str, secret) -> str: + """Replace every occurrence of secret in text with REDACTED. + No-op when secret is falsy (no token configured).""" + if not secret: + return text + return text.replace(secret, "REDACTED") + + def clean(text: str) -> str: text = text.replace("\r\n", "\n").replace("\r", "\n") text = text.replace("\n", "\\n") diff --git a/adhdo2/adhdolib/telegram.py b/adhdo2/adhdolib/telegram.py new file mode 100644 index 0000000..aa0702c --- /dev/null +++ b/adhdo2/adhdolib/telegram.py @@ -0,0 +1,150 @@ +"""Telegram bridge core: allowlist, per-chat rate limit, advisory crisis +screen, journaling, and a stdlib-only (urllib) Telegram Bot API client. + +Free text from Telegram is forwarded into the resident session through +bin/wake's inject() (literal tmux send-keys) — it can never become tmux +commands. The crisis screen is ADVISORY only: matched messages are flagged +in the journal and the forwarded payload, never blocked.""" +import json, os, re, time, urllib.error, urllib.parse, urllib.request +from collections import defaultdict, deque + +from . import db +from .envelope import ToolError +from .sanitize import redact + +# Ported from the legacy src/mcp_server/llm_client.py SafetyMonitor +# crisis_patterns, tightened per the spec: the security_middleware.py +# patterns (crisis|emergency|help me) false-positive heavily on normal +# ADHD talk ("help me focus") and are deliberately NOT included. +CRISIS_PATTERNS = [re.compile(p, re.IGNORECASE) for p in ( + r"\b(want to die|kill myself|end it all|suicide|suicidal)\b", + r"\b(harm myself|hurt myself|self[- ]harm)\b", + r"\b(no point (in )?living|life isn'?t worth|rather be dead)\b", + r"\b(can'?t go on|want to disappear|end the pain)\b", +)] + + +def crisis_screen(text: str) -> bool: + return any(p.search(text) for p in CRISIS_PATTERNS) + + +def neutralize(text: str) -> str: + """Make user text inert inside the injected trusted framing. + + Square brackets are the trusted-marker syntax (e.g. [telegram chat:N], + [crisis_advisory], [event:...]) and the payload wraps user text in double + quotes — so user-supplied brackets become parentheses and double quotes + are backslash-escaped, keeping the text readable while ensuring it can + never be parsed as trusted framing.""" + return text.replace("[", "(").replace("]", ")").replace('"', '\\"') + + +def get_token(cfg: dict) -> str: + token = os.environ.get("TELEGRAM_BOT_TOKEN") or cfg["telegram"].get("bot_token") + if not token: + raise ToolError("no_token", + "set telegram.bot_token in config.yaml or TELEGRAM_BOT_TOKEN") + return token + + +class TelegramAPI: + """Minimal Telegram Bot API client on urllib only.""" + + def __init__(self, token: str, base: str = "https://api.telegram.org"): + self.token = token + self.base = base + + def _call(self, method: str, params: dict, timeout: float = 70): + url = f"{self.base}/bot{self.token}/{method}" + data = urllib.parse.urlencode(params).encode() + try: + with urllib.request.urlopen(url, data=data, timeout=timeout) as resp: + body = json.loads(resp.read().decode()) + except urllib.error.HTTPError as e: + detail = redact(f"{method}: HTTP {e.code} {e.read().decode(errors='replace')[:200]}", + self.token) + raise ToolError("telegram_api", detail) + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as e: + raise ToolError("telegram_api", redact(f"{method}: {type(e).__name__}: {e}", self.token)) + if not body.get("ok"): + raise ToolError("telegram_api", + redact(f"{method}: {body.get('description', 'not ok')}", self.token)) + return body["result"] + + def get_updates(self, offset: int, timeout: int = 50): + return self._call("getUpdates", + {"offset": offset, "timeout": timeout, + "allowed_updates": '["message"]'}, + timeout=timeout + 20) + + def send_message(self, chat_id, text: str): + return self._call("sendMessage", {"chat_id": chat_id, "text": text}, timeout=30) + + +class Bridge: + """Handles inbound updates: allowlist -> text-only -> rate limit -> + crisis screen (advisory) -> journal -> inject into the session.""" + + def __init__(self, cfg: dict, conn, injector, api: TelegramAPI): + tcfg = cfg["telegram"] + self.allowlist = {int(c) for c in tcfg["chat_id_allowlist"]} + self.rate_limit = int(tcfg.get("rate_limit_per_minute", 6)) + self.conn = conn + self.injector = injector # wake.inject: returns "sent" or "queued" + self.api = api + self._recent = defaultdict(deque) # chat_id -> timestamps of forwarded msgs + + def _rate_limited(self, chat_id, now: float) -> bool: + q = self._recent[chat_id] + while q and q[0] <= now - 60: + q.popleft() + if len(q) >= self.rate_limit: + return True + q.append(now) + return False + + def _reply(self, chat_id, text: str): + """Best-effort outbound courtesy reply: a sendMessage failure is + journaled and swallowed so it can never kill the daemon.""" + try: + self.api.send_message(chat_id, text) + except Exception as e: + db.log_event(self.conn, "telegram", "error", json.dumps( + {"reason": "send_failed", "chat_id": chat_id, + "error": type(e).__name__})) + + def handle_update(self, update: dict) -> dict: + msg = update.get("message") + if not msg or "chat" not in msg: + return {"action": "ignored"} + chat_id = msg["chat"]["id"] + if chat_id not in self.allowlist: + db.log_event(self.conn, "telegram", "error", + json.dumps({"reason": "chat_not_allowed", "chat_id": chat_id})) + return {"action": "denied", "chat_id": chat_id} + text = msg.get("text") + if not text: + # non-text messages count against the rate limit too, and the + # courtesy reply is suppressed once limited — a media flood can't + # generate unlimited outbound sendMessage calls. + limited = self._rate_limited(chat_id, time.time()) + db.log_event(self.conn, "telegram", "error", json.dumps( + {"reason": "non_text_refused", "chat_id": chat_id, + "rate_limited": limited})) + if not limited: + self._reply(chat_id, "Text messages only, sorry — media is ignored.") + return {"action": "refused_media", "chat_id": chat_id, + "rate_limited": limited} + if self._rate_limited(chat_id, time.time()): + db.log_event(self.conn, "telegram", "error", + json.dumps({"reason": "rate_limited", "chat_id": chat_id})) + self._reply(chat_id, "Slow down a moment — rate limit hit, try again shortly.") + return {"action": "rate_limited", "chat_id": chat_id} + crisis = crisis_screen(text) + db.log_event(self.conn, "telegram", "wake", json.dumps( + {"chat_id": chat_id, "text": text, "crisis_advisory": crisis})) + flag = " [crisis_advisory]" if crisis else "" + payload = f'[telegram chat:{chat_id}]{flag} User says: "{neutralize(text)}"' + delivery = self.injector(payload) + return {"action": "forwarded", "chat_id": chat_id, + "crisis_advisory": crisis, "delivery": delivery} diff --git a/adhdo2/bin/adhdo-dashboard b/adhdo2/bin/adhdo-dashboard new file mode 100755 index 0000000..35317ea --- /dev/null +++ b/adhdo2/bin/adhdo-dashboard @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +import json, time, traceback +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from adhdolib import db +from adhdolib.binload import load_bin_module +from adhdolib.config import load_config +from adhdolib.envelope import scrub + +JOURNAL_LIMIT = 50 + +def get_state(): + state = load_bin_module("state") + return state.run(["--via", "dashboard"]) + +def get_journal(limit=JOURNAL_LIMIT): + conn = db.connect() + try: + rows = conn.execute( + "SELECT ts, source, type, payload_json FROM events " + "ORDER BY ts DESC LIMIT ?", (limit,)).fetchall() + return {"events": [ + {"ts": r[0], "source": r[1], "type": r[2], + "payload": json.loads(r[3]) if r[3] else None} for r in rows]} + finally: + conn.close() + +PAGE = """ + + + + +ADHDo Dashboard + + + +

ADHDo read-only dashboard

+

+

State

+
+

Devices

+
+

Recent journal

+
+ + + +""" + +class DashboardHandler(BaseHTTPRequestHandler): + server_version = "adhdo-dashboard" + + def _send(self, code, body, ctype): + data = body.encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", ctype) + self.send_header("Content-Length", str(len(data))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + if self.command != "HEAD": + self.wfile.write(data) + + def _send_json(self, code, obj): + self._send(code, scrub(json.dumps(obj, default=str)), "application/json") + + def do_GET(self): + path = self.path.split("?", 1)[0] + if path == "/": + self._send(200, PAGE, "text/html; charset=utf-8") + elif path == "/api/state": + self._api(get_state) + elif path == "/api/journal": + self._api(get_journal) + else: + self._send_json(404, {"error": "not_found", "detail": path}) + + def do_HEAD(self): + # Same routing as GET; _send suppresses the body for HEAD. + self.do_GET() + + def _api(self, fn): + try: + self._send_json(200, fn()) + except Exception: + # Full detail to the server log only; generic error to the client. + print(f"[dashboard] {self.path} failed:\n{traceback.format_exc()}", + file=sys.stderr, flush=True) + self._send_json(500, {"error": "internal", + "detail": "internal error (see server log)"}) + + def _reject(self): + self._send_json(405, {"error": "method_not_allowed", + "detail": "dashboard is read-only (GET only)"}) + + do_POST = do_PUT = do_DELETE = do_PATCH = do_OPTIONS = _reject + +BIND_RETRIES = 6 +BIND_RETRY_DELAY = 5.0 + +def _bind_server(bind, port, retries=BIND_RETRIES, delay=BIND_RETRY_DELAY): + """Bind with a short bounded retry: at boot the static lan_ip may not be + assigned yet, so give the network a little time before giving up.""" + for attempt in range(retries): + try: + return ThreadingHTTPServer((bind, port), DashboardHandler) + except OSError as e: + if attempt == retries - 1: + raise + print(f"[dashboard] bind {bind}:{port} failed ({e}), " + f"retry {attempt + 1}/{retries - 1} in {delay}s", + file=sys.stderr, flush=True) + time.sleep(delay) + +def main(): + cfg = load_config() + bind = cfg["dashboard"]["bind"] or cfg["lan_ip"] + httpd = _bind_server(bind, cfg["dashboard"]["port"]) + httpd.serve_forever() + +if __name__ == "__main__": + main() diff --git a/adhdo2/bin/adhdo-telegram b/adhdo2/bin/adhdo-telegram new file mode 100755 index 0000000..b6a46f4 --- /dev/null +++ b/adhdo2/bin/adhdo-telegram @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +import sys, pathlib +sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) + +import json, time +from adhdolib import db, telegram as tg +from adhdolib.binload import load_bin_module +from adhdolib.config import home, load_config +from adhdolib.envelope import ToolArgumentParser, ToolError, cli_main, write_atomic +from adhdolib.sanitize import redact + + +def read_offset(path) -> int: + try: + return int(path.read_text().strip()) + except (FileNotFoundError, ValueError): + return 0 + + +def poll_once(bridge, api, offset_path) -> int: + """One getUpdates long-poll cycle. Each update is handled inside its own + try/except so a broken injector (e.g. tmux session down) can never crash + the daemon: the failure is journaled and polling continues. The offset is + persisted only AFTER an update is handled (success or handled failure), + so a hard crash re-delivers the in-flight update instead of dropping it. + Returns the number of updates processed.""" + offset = read_offset(offset_path) + updates = api.get_updates(offset) + for upd in updates: + try: + bridge.handle_update(upd) + except Exception as e: + chat_id = (upd.get("message") or {}).get("chat", {}).get("id") + db.log_event(bridge.conn, "telegram", "error", json.dumps( + {"reason": "handle_failed", "chat_id": chat_id, + "error": redact(f"{type(e).__name__}: {e}", + getattr(api, "token", None))})) + write_atomic(offset_path, str(upd["update_id"] + 1)) + return len(updates) + + +def daemon(cfg, conn, api, once=False): + data = home() / "data" + data.mkdir(parents=True, exist_ok=True) + wake = load_bin_module("wake") + bridge = tg.Bridge(cfg, conn, wake.inject, api) + offset_path = data / "telegram.offset" + n = 0 + while True: + try: + n += poll_once(bridge, api, offset_path) + except ToolError as e: + # transient API/network failure: journal once, back off, keep polling + db.log_event(conn, "telegram", "error", f"poll: {e.code}: {e.detail}") + if not once: + time.sleep(10) + if once: + return {"polled": True, "handled": n} + + +def run(argv, _test_conn=None): + ap = ToolArgumentParser(prog="adhdo-telegram") + sub = ap.add_subparsers(dest="cmd") + runp = sub.add_parser("run") + runp.add_argument("--once", action="store_true") + sendp = sub.add_parser("send") + sendp.add_argument("chat_id", type=int) + sendp.add_argument("text") + args = ap.parse_args(argv) + if not args.cmd: + raise ToolError("usage", "adhdo-telegram run [--once] | send ") + cfg = load_config() + conn = _test_conn or db.connect() + api = tg.TelegramAPI(tg.get_token(cfg)) + try: + if args.cmd == "send": + api.send_message(args.chat_id, args.text) + db.audit(conn, "adhdo-telegram", argv, True, "sent") + return {"sent": True, "chat_id": args.chat_id} + return daemon(cfg, conn, api, once=args.once) + except ToolError as e: + db.audit(conn, "adhdo-telegram", argv, False, e.code) + raise + except Exception as e: + db.audit(conn, "adhdo-telegram", argv, False, f"crash: {type(e).__name__}") + raise + + +if __name__ == "__main__": + cli_main(lambda: run(sys.argv[1:])) diff --git a/adhdo2/bin/cast b/adhdo2/bin/cast index 393023c..c83a1d7 100755 --- a/adhdo2/bin/cast +++ b/adhdo2/bin/cast @@ -32,12 +32,26 @@ def resolve_mood(mood_or_url: str, cfg: dict): return random.choice(streams), "stream" raise ToolError("unknown_mood", f"mood '{mood_or_url}' has no playable source") -def play_on_device(url: str, device_name: str): +def device_not_found(device: str, real: str) -> ToolError: + """Uniform device-not-found error across all cast subcommands. + Reports the logical (configured) name, plus the real cast name if + they differ.""" + detail = (f"device '{device}' ('{real}') not found" + if real != device else f"device '{device}' not found") + return ToolError("no_devices", detail) + +def resolve_device(requested, cfg): + """Map a logical device name (or None for the default) to its real + discovered/cast name via the config alias table.""" + device = requested or cfg["default_device"] + return device, cfg["devices"].get(device, device) + +def play_on_device(url: str, device: str, real: str): import pychromecast - casts, browser = pychromecast.get_listed_chromecasts(friendly_names=[device_name]) + casts, browser = pychromecast.get_listed_chromecasts(friendly_names=[real]) try: if not casts: - raise ToolError("no_devices", f"{device_name} not found") + raise device_not_found(device, real) cc = casts[0]; cc.wait() cc.media_controller.play_media(url, "audio/mpeg") cc.media_controller.block_until_active(timeout=15) @@ -46,7 +60,8 @@ def play_on_device(url: str, device_name: str): def device_status(cfg): import pychromecast - names = list(cfg["devices"].values()) + aliases = cfg["devices"] # logical name -> real cast name + names = list(aliases.values()) casts, browser = pychromecast.get_listed_chromecasts(friendly_names=names) try: now_playing = {} @@ -57,9 +72,9 @@ def device_status(cfg): now_playing[cc.name] = s.title if s and s.player_is_playing else None except Exception: pass # this device errored; treat as offline below - return [{"name": n, "online": n in now_playing, - "now_playing": now_playing.get(n)} - for n in names] + return [{"device": logical, "name": real, "online": real in now_playing, + "now_playing": now_playing.get(real)} + for logical, real in aliases.items()] finally: browser.stop_discovery() @@ -77,44 +92,59 @@ def run(argv, _test_conn=None, _test_cfg=None): if not args.arg: raise ToolError("usage", "cast play ") url, source = resolve_mood(args.arg, cfg) - device = args.device or cfg["default_device"] - real = cfg["devices"].get(device, device) - play_on_device(url, real) + device, real = resolve_device(args.device, cfg) + play_on_device(url, device, real) np_path.parent.mkdir(parents=True, exist_ok=True) write_atomic(np_path, json.dumps( {"url": url, "device": real, "mood": args.arg, "started_ts": time.time()})) os.chmod(np_path, 0o600) db.log_event(conn, "cast", "cast", - json.dumps({"mood": args.arg, "device": device})) - out = {"playing": scrub(url), "device": device, "source": source} + json.dumps({"action": "play", "mood": args.arg, + "device": device})) + out = {"playing": scrub(url), "device": device, + "cast_name": real, "source": source} elif args.cmd == "stop": - device = args.device or cfg["default_device"] - real = cfg["devices"].get(device, device) + device, real = resolve_device(args.device, cfg) import pychromecast casts, browser = pychromecast.get_listed_chromecasts(friendly_names=[real]) try: - if casts: - casts[0].wait(); casts[0].media_controller.stop() + if not casts: + # The user asked to stop: clear now_playing.json even + # though the device is unreachable, so a later nudge's + # resume doesn't restart music they tried to stop. + if np_path.exists(): + np_path.unlink(missing_ok=True) + db.log_event(conn, "cast", "cast", json.dumps( + {"action": "stop_cleared_now_playing", + "device": device})) + raise device_not_found(device, real) + casts[0].wait(); casts[0].media_controller.stop() finally: browser.stop_discovery() np_path.unlink(missing_ok=True) - out = {"stopped": True} + db.log_event(conn, "cast", "cast", + json.dumps({"action": "stop", "device": device})) + out = {"stopped": True, "device": device, "cast_name": real} elif args.cmd == "volume": - level = float(args.arg) + if args.arg is None: + raise ToolError("usage", "cast volume <0..1>") + try: + level = float(args.arg) + except ValueError: + raise ToolError("usage", "volume must be 0..1") if not 0 <= level <= 1: raise ToolError("usage", "volume must be 0..1") - device = args.device or cfg["default_device"] - real = cfg["devices"].get(device, device) + device, real = resolve_device(args.device, cfg) import pychromecast casts, browser = pychromecast.get_listed_chromecasts(friendly_names=[real]) try: if not casts: - raise ToolError("no_devices", f"{real} not found") + raise device_not_found(device, real) casts[0].wait(); casts[0].set_volume(level) finally: browser.stop_discovery() - out = {"volume": level} + out = {"volume": level, "device": device, "cast_name": real} else: out = {"devices": device_status(cfg)} db.audit(conn, "cast", argv, True, "") diff --git a/adhdo2/bin/journal b/adhdo2/bin/journal index ed9543f..ca70ef4 100755 --- a/adhdo2/bin/journal +++ b/adhdo2/bin/journal @@ -15,6 +15,21 @@ def cmd_log(conn, args): db.log_event(conn, "session", type_, text) return {"logged": type_} +OUTCOMES = frozenset({"worked", "partial", "ignored", "backfired"}) + +def cmd_outcome(conn, args): + if len(args) < 2: + raise ToolError( + "usage", "journal outcome [feedback...]") + intervention, outcome = args[0], args[1] + if outcome not in OUTCOMES: + raise ToolError("bad_outcome", f"outcome must be one of {sorted(OUTCOMES)}") + payload = {"intervention": intervention, "outcome": outcome} + if len(args) > 2: + payload["feedback"] = " ".join(args[2:]) + db.log_event(conn, "session", "outcome", json.dumps(payload)) + return {"logged": "outcome", "intervention": intervention, "outcome": outcome} + def cmd_recent(conn, args): n = int(args[0]) if args else 20 rows = conn.execute( @@ -58,15 +73,78 @@ def cmd_patterns(conn, args): "FROM events WHERE type IN ('med','meal','break') AND ts>? GROUP BY 1", (cutoff,)): streaks[t] = n + outcome_by_intervention, outcome_by_daypart = {}, {} + for (intervention, outcome, n) in conn.execute( + "SELECT COALESCE(json_extract(payload_json,'$.intervention'),'unknown'), " + "COALESCE(json_extract(payload_json,'$.outcome'),'unknown'), COUNT(*) " + "FROM events WHERE type='outcome' AND ts>? GROUP BY 1, 2", (cutoff,)): + bucket = outcome_by_intervention.setdefault(intervention, {}) + bucket[outcome] = bucket.get(outcome, 0) + n + for (h, outcome, n) in conn.execute( + "SELECT strftime('%H', ts, 'unixepoch', 'localtime'), " + "COALESCE(json_extract(payload_json,'$.outcome'),'unknown'), COUNT(*) " + "FROM events WHERE type='outcome' AND ts>? GROUP BY 1, 2", (cutoff,)): + dp = _daypart(int(h)) + bucket = outcome_by_daypart.setdefault(dp, {}) + bucket[outcome] = bucket.get(outcome, 0) + n + success_rate_by_intervention = {} + for intervention, counts in outcome_by_intervention.items(): + total = sum(counts.values()) + good = counts.get("worked", 0) + counts.get("partial", 0) * 0.5 + success_rate_by_intervention[intervention] = round(good / total, 3) if total else None return {"window_days": 30, "nudge_response_rate_by_hour": by_hour, "nudge_response_rate_by_type": by_type, - "mood_by_daypart": moods, "event_streaks": streaks} + "mood_by_daypart": moods, "event_streaks": streaks, + "outcome_by_intervention": outcome_by_intervention, + "outcome_by_daypart": outcome_by_daypart, + "success_rate_by_intervention": success_rate_by_intervention} + +RETENTION_DAYS = 90 + +def cmd_rollup(conn, args): + """Aggregate complete days of events into rollups, then prune raw + events/audit rows older than RETENTION_DAYS. Idempotent.""" + today = time.strftime("%Y-%m-%d", time.localtime()) + days = [r[0] for r in conn.execute( + "SELECT DISTINCT date(ts,'unixepoch','localtime') FROM events " + "WHERE date(ts,'unixepoch','localtime') < ? " + "AND date(ts,'unixepoch','localtime') NOT IN " + "(SELECT day FROM rollups WHERE metric='events_by_type')", (today,))] + for day in days: + by_type = {t: n for (t, n) in conn.execute( + "SELECT type, COUNT(*) FROM events " + "WHERE date(ts,'unixepoch','localtime')=? GROUP BY 1", (day,))} + outcomes = {} + for (intervention, outcome, n) in conn.execute( + "SELECT COALESCE(json_extract(payload_json,'$.intervention'),'unknown'), " + "COALESCE(json_extract(payload_json,'$.outcome'),'unknown'), COUNT(*) " + "FROM events WHERE type='outcome' " + "AND date(ts,'unixepoch','localtime')=? GROUP BY 1, 2", (day,)): + outcomes[f"{intervention}:{outcome}"] = n + nudges_by_hour = {h: n for (h, n) in conn.execute( + "SELECT strftime('%H', ts, 'unixepoch', 'localtime'), COUNT(*) " + "FROM events WHERE type='nudge' " + "AND date(ts,'unixepoch','localtime')=? GROUP BY 1", (day,))} + for metric, value in (("events_by_type", by_type), + ("outcomes", outcomes), + ("nudges_by_hour", nudges_by_hour)): + conn.execute("INSERT OR REPLACE INTO rollups(day, metric, value_json) " + "VALUES(?,?,?)", (day, metric, json.dumps(value))) + prune_cutoff = time.time() - RETENTION_DAYS * 86400 + pruned_events = conn.execute( + "DELETE FROM events WHERE ts ...") + raise ToolError("usage", "journal ...") conn = db.connect() - cmd = {"log": cmd_log, "recent": cmd_recent, "patterns": cmd_patterns}.get(sys.argv[1]) + cmd = {"log": cmd_log, "outcome": cmd_outcome, "recent": cmd_recent, + "patterns": cmd_patterns, "rollup": cmd_rollup}.get(sys.argv[1]) if not cmd: raise ToolError("usage", f"unknown subcommand {sys.argv[1]}") return cmd(conn, sys.argv[2:]) diff --git a/adhdo2/bin/nudge b/adhdo2/bin/nudge index b456cb8..54d4439 100755 --- a/adhdo2/bin/nudge +++ b/adhdo2/bin/nudge @@ -10,6 +10,25 @@ from adhdolib.config import load_config, home from adhdolib.envelope import ToolArgumentParser, ToolError, cli_main from adhdolib.nudgelimits import check_allowed +def prune_cache(cache_dir: Path, max_age: float = 86400) -> int: + """Delete cached mp3s older than max_age seconds. Runs on every nudge + invocation (not just on cache misses) so the cache can't grow unbounded. + Returns the number of files removed.""" + if not cache_dir.is_dir(): + return 0 + cutoff = time.time() - max_age + removed = 0 + for f in cache_dir.glob("*.mp3"): + # A concurrent nudge may prune the same file between glob and + # stat/unlink; tolerate the race instead of crashing. + try: + if f.stat().st_mtime < cutoff: + f.unlink(missing_ok=True) + removed += 1 + except FileNotFoundError: + continue + return removed + def synthesize(text: str, cache_dir: Path) -> Path: cache_dir.mkdir(parents=True, exist_ok=True) out = cache_dir / (hashlib.sha1(text.encode()).hexdigest()[:16] + ".mp3") @@ -29,11 +48,6 @@ def synthesize(text: str, cache_dir: Path) -> Path: subprocess.run(["ffmpeg", "-y", "-loglevel", "error", "-i", str(wav), str(out)], check=True, timeout=60) wav.unlink(missing_ok=True) - # prune cache >24h - cutoff = time.time() - 86400 - for f in cache_dir.glob("*.mp3"): - if f.stat().st_mtime < cutoff and f != out: - f.unlink() return out def play_url_on_device(url: str, device_name: str, cfg: dict): @@ -104,7 +118,9 @@ def run(argv, _test_conn=None): if not device: raise ToolError("no_device_configured", "set default_device in config") real_name = cfg["devices"].get(device, device) - mp3 = synthesize(args.text, home() / "tts-cache") + cache_dir = home() / "tts-cache" + prune_cache(cache_dir) + mp3 = synthesize(args.text, cache_dir) url = f"http://{cfg['lan_ip']}:{cfg['httpd_port']}/{mp3.name}" play_url_on_device(url, real_name, cfg) wait_until_idle(real_name, cfg) diff --git a/adhdo2/bin/wake b/adhdo2/bin/wake index 4773e47..eca83ac 100755 --- a/adhdo2/bin/wake +++ b/adhdo2/bin/wake @@ -39,13 +39,22 @@ def inject(text: str) -> str: return "sent" def _parse_at(s: str) -> float: - if ":" in s and len(s) <= 5: - h, m = map(int, s.split(":")) - due = datetime.now().replace(hour=h, minute=m, second=0, microsecond=0) - if due.timestamp() <= time.time(): - due = due + timedelta(days=1) - return due.timestamp() - return datetime.fromisoformat(s).timestamp() + try: + if ":" in s and "-" not in s and "T" not in s and " " not in s: + # time-of-day: HH:MM or HH:MM:SS (next occurrence, today or tomorrow) + parts = s.split(":") + if len(parts) not in (2, 3): + raise ValueError(s) + h, m = int(parts[0]), int(parts[1]) + sec = int(parts[2]) if len(parts) == 3 else 0 + due = datetime.now().replace(hour=h, minute=m, second=sec, microsecond=0) + if due.timestamp() <= time.time(): + due = due + timedelta(days=1) + return due.timestamp() + # ISO-8601 datetime or date (date-only means midnight) + return datetime.fromisoformat(s).timestamp() + except ValueError: + raise ToolError("usage", f"invalid --at value: {s!r} (use HH:MM[:SS] or ISO-8601)") def run(argv, _test_conn=None): ap = ToolArgumentParser(prog="wake") diff --git a/adhdo2/config.example.yaml b/adhdo2/config.example.yaml index c380256..71cdf41 100644 --- a/adhdo2/config.example.yaml +++ b/adhdo2/config.example.yaml @@ -13,6 +13,12 @@ nudge: quiet_hours: ["22:00", "07:00"] high_urgency_events: [medication, safety, user_requested] disk_warn_pct: 94 +dashboard: # P3 read-only LAN dashboard (separate service) + bind: null # null → falls back to lan_ip + port: 8766 crisis: contacts: ["Lifeline Australia 13 11 14"] -telegram: {chat_id_allowlist: []} # P3 +telegram: # P3 bridge + chat_id_allowlist: [] # numeric Telegram chat IDs allowed to talk to the session + bot_token: "" # from @BotFather; env TELEGRAM_BOT_TOKEN overrides + rate_limit_per_minute: 6 # per-chat inbound message cap diff --git a/adhdo2/docs/catchup-vs-schedule.md b/adhdo2/docs/catchup-vs-schedule.md new file mode 100644 index 0000000..cccebf1 --- /dev/null +++ b/adhdo2/docs/catchup-vs-schedule.md @@ -0,0 +1,40 @@ +# Catch-up vs. schedule: how missed events are handled + +Two mechanisms deliver time-based wakes, and they must not double-fire. + +## Normal operation (Pi is up) + +- **Fixed events** (meds 08:00, bedtime 22:30) fire from cron entries + installed by `scripts/install-cron.sh` (`bin/wake --event ...`). +- **Heartbeat** fires from the `adhdo-heartbeat.timer` systemd user timer + every 25 minutes (a true 25-min cadence; cron's `*/25` leaves a 10-minute + gap at each hour boundary, which is why this is a timer, not a cron line). +- **One-off follow-ups** (`bin/wake --at