diff --git a/nerve/channels/telegram.py b/nerve/channels/telegram.py index d1d0515..cd4dde3 100644 --- a/nerve/channels/telegram.py +++ b/nerve/channels/telegram.py @@ -1515,11 +1515,22 @@ async def _handle_callback_query(self, update: Update, context: Any) -> None: ) if success: - await query.answer(f"Answered: {answer}") + status_line = f"\u2705 Answered: {answer}" + toast = f"Answered: {answer}" + # A snooze keeps the row pending with redeliver_at stamped \u2014 + # confirm on the card that it will come back, instead of the + # generic answered state (which read as "handled, gone"). + snoozed_until = await self._get_snoozed_until(notification_id) + if snoozed_until: + status_line = ( + f"\U0001F4A4 Snoozed until {snoozed_until} \u2014 will resurface" + ) + toast = f"Snoozed until {snoozed_until}" + await query.answer(toast) try: original = query.message.text or "" await query.edit_message_text( - text=f"{original}\n\n\u2705 Answered: {answer}", + text=f"{original}\n\n{status_line}", reply_markup=None, ) except Exception: @@ -1527,6 +1538,30 @@ async def _handle_callback_query(self, update: Update, context: Any) -> None: else: await query.answer("Already answered or expired", show_alert=True) + async def _get_snoozed_until(self, notification_id: str) -> str | None: + """Return a human-readable re-delivery time if the row was snoozed. + + After ``handle_answer`` succeeds, a snoozed approval is the only + outcome that leaves the row ``pending`` with ``redeliver_at`` + set. Rendered in the host's local timezone. None when the answer + was a final decision (or anything fails \u2014 this is cosmetic). + """ + try: + notif = await self._notification_service.db.get_notification( + notification_id, + ) + if ( + not notif + or notif.get("status") != "pending" + or not notif.get("redeliver_at") + ): + return None + from datetime import datetime + dt = datetime.fromisoformat(notif["redeliver_at"]) + return dt.astimezone().strftime("%Y-%m-%d %H:%M %Z") + except Exception: + return None + async def _handle_reply(self, update: Update, context: Any) -> None: """Handle /reply ā€” answer the most recent pending question.""" self._touch() diff --git a/nerve/config.py b/nerve/config.py index 1488733..9096e56 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -666,6 +666,7 @@ class NotificationsConfig: channels: list[str] = field(default_factory=lambda: ["web", "telegram"]) telegram_chat_id: int | None = None # Target chat; falls back to first allowed_user default_expiry_hours: int = 48 # Auto-expire unanswered questions + max_redeliveries: int = 3 # Per-row cap on snooze/re-delivery cycles priority_prefixes: dict[str, str] = field(default_factory=lambda: { "high": "āš ļø ", "urgent": "šŸšØ ", @@ -677,6 +678,7 @@ def from_dict(cls, d: dict) -> NotificationsConfig: channels=d.get("channels", ["web", "telegram"]), telegram_chat_id=d.get("telegram_chat_id"), default_expiry_hours=d.get("default_expiry_hours", 48), + max_redeliveries=d.get("max_redeliveries", 3), priority_prefixes=d.get("priority_prefixes", { "high": "āš ļø ", "urgent": "šŸšØ ", diff --git a/nerve/db/migrations/v037_notification_redelivery.py b/nerve/db/migrations/v037_notification_redelivery.py new file mode 100644 index 0000000..c1fd56c --- /dev/null +++ b/nerve/db/migrations/v037_notification_redelivery.py @@ -0,0 +1,46 @@ +"""V37: Add re-delivery columns to notifications. + +Completes the snooze lifecycle for ``question``/``approval`` rows. Until +now, snoozing only advanced ``expires_at`` ā€” the row sat pending and +invisible until the expiry sweep silently killed it ("delayed silent +decline"). The two new columns let the periodic maintenance tick +re-surface snoozed rows with a fresh fanout: + +- ``redeliver_at`` TEXT NULL: when set on a ``pending`` row, the + maintenance tick re-fans-out the notification at/after this time + (fresh Telegram card + web broadcast). NULL = no re-delivery queued. +- ``redelivery_count`` INTEGER NOT NULL DEFAULT 0: how many times the + row has been re-delivered. Capped by + ``config.notifications.max_redeliveries`` so a snooze loop can't nag + forever ā€” at the cap the row expires (with reporting) instead. + +Existing rows are untouched (redeliver_at = NULL, count = 0), so +nothing is re-delivered retroactively. +""" + +from __future__ import annotations + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + + +async def up(db: aiosqlite.Connection) -> None: + await db.execute( + "ALTER TABLE notifications ADD COLUMN redeliver_at TEXT" + ) + await db.execute( + "ALTER TABLE notifications ADD COLUMN redelivery_count " + "INTEGER NOT NULL DEFAULT 0" + ) + # The maintenance tick polls "pending rows whose redeliver_at is due" + # every 15 minutes; keep that scan off the table. + await db.execute( + "CREATE INDEX IF NOT EXISTS idx_notifications_redeliver " + "ON notifications(status, redeliver_at)" + ) + logger.info( + "v037: added redeliver_at/redelivery_count to notifications + index" + ) diff --git a/nerve/db/notifications.py b/nerve/db/notifications.py index 23459e3..3415ddc 100644 --- a/nerve/db/notifications.py +++ b/nerve/db/notifications.py @@ -137,26 +137,71 @@ async def dismiss_all_notifications(self) -> int: await self.db.commit() return cursor.rowcount - async def expire_notifications(self) -> int: + async def expire_due_notifications(self) -> list[dict]: + """Flip pending rows past their expiry to ``expired``. + + Returns the affected rows (as they were *before* the flip, with + ``status`` already rewritten to ``'expired'`` in the returned + dicts) so the service layer can report each expiry ā€” inject a + note into the asking session, audit-log approvals, gray the web + card, edit the Telegram message. Select-then-update runs inside + one transaction so a concurrent answer can't slip between the + two statements. + """ now = datetime.now(timezone.utc).isoformat() - cursor = await self.db.execute( - """UPDATE notifications SET status = 'expired' - WHERE status = 'pending' AND expires_at IS NOT NULL AND expires_at < ?""", - (now,), - ) - await self.db.commit() - return cursor.rowcount + async with self._atomic(): + async with self.db.execute( + """SELECT * FROM notifications + WHERE status = 'pending' + AND expires_at IS NOT NULL AND expires_at < ?""", + (now,), + ) as cursor: + rows = [dict(row) async for row in cursor] + if rows: + placeholders = ",".join("?" for _ in rows) + await self.db.execute( + f"""UPDATE notifications SET status = 'expired' + WHERE id IN ({placeholders})""", + tuple(r["id"] for r in rows), + ) + for r in rows: + r["status"] = "expired" + return rows + + async def expire_notification(self, notification_id: str) -> bool: + """Flip a single pending row to ``expired``. + + Used by the re-delivery tick when a row hits the + ``max_redeliveries`` cap: instead of another fanout, it expires + (with reporting) even though ``expires_at`` may still be in the + future. Returns False if the row is not pending. + """ + async with self._atomic(): + async with self.db.execute( + "SELECT id FROM notifications WHERE id = ? AND status = 'pending'", + (notification_id,), + ) as cursor: + if not await cursor.fetchone(): + return False + await self.db.execute( + "UPDATE notifications SET status = 'expired' WHERE id = ?", + (notification_id,), + ) + return True async def snooze_notification( - self, notification_id: str, new_expires_at: str, + self, notification_id: str, redeliver_at: str, new_expires_at: str, ) -> bool: - """Push a pending notification's expiry forward. + """Queue a pending notification for re-delivery. - Used by the ``approval`` dispatcher when the user picks - ``snooze_24h``: the row stays at status=pending so a later - re-delivery tick (wired in PR 2) can surface it again, but the - expiry advances so it does not get caught by ``expire_stale`` - in the meantime. + Used when the user picks ``snooze_24h`` on an approval: the row + stays at status=pending, ``redeliver_at`` marks when the + periodic maintenance tick should fan it out again (fresh + Telegram card + web broadcast), and ``expires_at`` advances past + the re-delivery time so the row cannot expire before it + resurfaces. Re-snoozing after a re-delivery simply sets + ``redeliver_at`` again ā€” each snooze buys another cycle, up to + ``config.notifications.max_redeliveries``. Returns True on success, False if the row is not pending. """ @@ -168,11 +213,64 @@ async def snooze_notification( if not await cursor.fetchone(): return False await self.db.execute( - "UPDATE notifications SET expires_at = ? WHERE id = ?", - (new_expires_at, notification_id), + """UPDATE notifications SET redeliver_at = ?, expires_at = ? + WHERE id = ?""", + (redeliver_at, new_expires_at, notification_id), ) return True + async def get_due_redeliveries(self) -> list[dict]: + """Return pending rows whose ``redeliver_at`` has passed. + + Oldest-first so long-waiting rows resurface before fresher ones + when several come due in the same sweep. + """ + now = datetime.now(timezone.utc).isoformat() + async with self.db.execute( + """SELECT * FROM notifications + WHERE status = 'pending' + AND redeliver_at IS NOT NULL AND redeliver_at <= ? + ORDER BY created_at ASC""", + (now,), + ) as cursor: + return [dict(row) async for row in cursor] + + async def mark_notification_redelivered( + self, notification_id: str, new_expires_at: str | None = None, + ) -> bool: + """Record one re-delivery: bump the count, clear ``redeliver_at``. + + ``new_expires_at`` (when given) restarts the expiry window so + the fresh card gets a full answering window ā€” without it, a row + whose original expiry already passed would be expired by the + very next pass of the same sweep that just re-delivered it. + Returns False if the row is not pending. + """ + async with self._atomic(): + async with self.db.execute( + "SELECT id FROM notifications WHERE id = ? AND status = 'pending'", + (notification_id,), + ) as cursor: + if not await cursor.fetchone(): + return False + if new_expires_at is not None: + await self.db.execute( + """UPDATE notifications + SET redelivery_count = redelivery_count + 1, + redeliver_at = NULL, expires_at = ? + WHERE id = ?""", + (new_expires_at, notification_id), + ) + else: + await self.db.execute( + """UPDATE notifications + SET redelivery_count = redelivery_count + 1, + redeliver_at = NULL + WHERE id = ?""", + (notification_id,), + ) + return True + async def count_pending_notifications(self, channel: str | None = None) -> int: sql = "SELECT COUNT(*) FROM notifications WHERE status = 'pending'" params: tuple = () diff --git a/nerve/gateway/server.py b/nerve/gateway/server.py index b9e27f1..f191050 100644 --- a/nerve/gateway/server.py +++ b/nerve/gateway/server.py @@ -256,10 +256,21 @@ async def _periodic_idle_sweep(): idle_sweep_task = asyncio.create_task(_periodic_idle_sweep()) - # Periodic notification expiry (every 15 minutes) - async def _periodic_notify_expiry(): + # Periodic notification maintenance (every 15 minutes): re-deliver + # snoozed rows, then expire stale ones. Ordered so a row whose + # redeliver_at AND expires_at both passed gets its last chance + # (re-delivery restarts the expiry window) instead of dying. + async def _periodic_notify_maintenance(): while True: await asyncio.sleep(15 * 60) + try: + redelivered = await notification_service.redeliver_due() + if redelivered: + logger.info( + "Re-delivered %d snoozed notifications", redelivered, + ) + except Exception as e: + logger.error("Notification re-delivery failed: %s", e) try: expired = await notification_service.expire_stale() if expired: @@ -267,7 +278,7 @@ async def _periodic_notify_expiry(): except Exception as e: logger.error("Notification expiry failed: %s", e) - notify_expiry_task = asyncio.create_task(_periodic_notify_expiry()) + notify_maintenance_task = asyncio.create_task(_periodic_notify_maintenance()) # Periodic DB retention (opt-in). Compacts old memorized messages' # blocks/thinking JSON and prunes append-only telemetry + file snapshots, @@ -475,7 +486,7 @@ async def _periodic_backup(): await cron_task.stop() db_retention_task.cancel() - notify_expiry_task.cancel() + notify_maintenance_task.cancel() backup_task.cancel() idle_sweep_task.cancel() memorize_task.cancel() diff --git a/nerve/notifications/handlers.py b/nerve/notifications/handlers.py index 485ebd8..99112c5 100644 --- a/nerve/notifications/handlers.py +++ b/nerve/notifications/handlers.py @@ -7,8 +7,10 @@ return a structured ``DispatchResult`` so the service can audit-log the outcome uniformly. -PR 1 ships one dispatcher: ``mechanical-action``. PR 2 will add the -``plan`` dispatcher and the per-target re-delivery wiring. +One dispatcher ships today: ``mechanical-action``. Snoozed approvals +are re-surfaced by the notification service's periodic maintenance +tick (``NotificationService.redeliver_due``); a future ``plan`` +dispatcher can register here without service changes. Design notes: @@ -62,8 +64,11 @@ class DispatchResult: mechanical-actions audit log. Already includes the dispatcher's own event name (``approval-acted``); the service adds the notification id and timestamp. - - ``snooze_until``: ISO-8601 UTC timestamp for the new expiry when - the decision is a snooze. None for approve / decline. + - ``snooze_until``: ISO-8601 UTC timestamp for when a snoozed + notification should be re-delivered. The service stamps it into + the row's ``redeliver_at`` (and pushes ``expires_at`` past it) so + the periodic maintenance tick re-surfaces the card. None for + approve / decline. """ ok: bool @@ -151,9 +156,9 @@ def _dispatch_mechanical_action( Snooze: rather than touch the queue file directly, this dispatcher calls ``mechanical-action.sh snooze --hours 24`` and lets the decide-side script record the audit event and update the queue - entry's ``not_before`` field. PR 2 wires the re-delivery scheduler; - until then, the snooze just keeps the proposal in queue/ with a - future ``not_before`` timestamp. + entry's ``not_before`` field. The returned ``snooze_until`` makes + the service stamp the notification row's ``redeliver_at``, and the + periodic maintenance tick re-delivers the card at that time. """ notif_id = notification.get("id", "") base_event: dict[str, Any] = { diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index 0ab6d94..1527fb9 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -411,9 +411,9 @@ async def handle_answer( - For ``type=approval`` rows: look up the dispatcher in the handler registry, run it, audit-log the outcome, then flip - the row's status. Snooze answers advance ``expires_at`` and - keep the row pending so a later re-delivery tick can surface - it again. + the row's status. Snooze answers keep the row pending and + stamp ``redeliver_at`` so the periodic maintenance tick + (:meth:`redeliver_due`) fans it out again with a fresh card. - For ``type=question`` rows (legacy): persist the answer, inject it back into the originating session, broadcast. - Fire-and-forget ``type=notify`` rows do not flow through this @@ -564,11 +564,35 @@ async def _handle_approval_answer( await self._append_approval_audit(result.audit_event) - # Snooze keeps the row pending with a future expiry so a later - # re-delivery tick (wired in PR 2) can surface it again. - if result.snooze_until is not None and result.ok: + # Snooze keeps the row pending and stamps ``redeliver_at`` so + # the periodic maintenance tick (:meth:`redeliver_due`) fans it + # out again at the snooze time. ``expires_at`` moves past the + # re-delivery point so the expiry sweep cannot kill the row + # before it resurfaces. + snoozed = result.snooze_until is not None and result.ok + snooze_until = result.snooze_until + if snoozed: + try: + snooze_dt = datetime.fromisoformat(snooze_until) + except ValueError: + # A dispatcher returned a malformed timestamp. Fall back + # to a 24h snooze rather than crash the answer route + # (which would leave the row pending with its Telegram + # buttons already consumed ā€” the exact silent-loss shape + # this path exists to prevent). + logger.warning( + "dispatcher returned malformed snooze_until %r for %s; " + "defaulting to +24h", + snooze_until, notification_id, + ) + snooze_dt = datetime.now(timezone.utc) + timedelta(hours=24) + snooze_until = snooze_dt.isoformat() + expiry_hours = self.config.notifications.default_expiry_hours + new_expires_at = ( + snooze_dt + timedelta(hours=expiry_hours) + ).isoformat() await self.db.snooze_notification( - notification_id, result.snooze_until, + notification_id, snooze_until, new_expires_at, ) else: await self.db.answer_notification( @@ -576,29 +600,30 @@ async def _handle_approval_answer( ) from nerve.agent.streaming import broadcaster - broadcast_status = ( - "snoozed" if (result.snooze_until and result.ok) else "answered" - ) - await broadcaster.broadcast("__global__", { + payload = { "type": "notification_answered", "notification_id": notification_id, "session_id": session_id, "answer": answer, "answered_by": answered_by, - "approval_status": broadcast_status, + "approval_status": "snoozed" if snoozed else "answered", "dispatch_ok": result.ok, - }) + } + if snoozed: + payload["snooze_until"] = snooze_until + await broadcaster.broadcast("__global__", payload) return True async def _append_approval_audit(self, event: dict[str, Any]) -> None: - """Append an ``approval-acted`` record to the mechanical-actions log. + """Append an approval-lifecycle record to the mechanical-actions log. Uses the same audit log that the propose-mechanical-action primitive writes to (``~/.nerve/mechanical-actions/audit.jsonl``) so the proposal lifecycle (``proposed`` -> ``approval-acted`` - -> ``approved``/``declined``/``executed``) is visible in one - place. The shared helper module + -> ``approved``/``declined``/``executed``, plus + ``approval-expired`` when a card dies unanswered) is visible in + one place. The shared helper module ``scripts/_mechanical_action.py`` lives under the workspace, so we import it dynamically by path rather than as a real Python package. @@ -625,8 +650,9 @@ async def _append_approval_audit(self, event: dict[str, Any]) -> None: # so the helper validation stays strict for everything else. record = {"ts": ts, **event} valid = getattr(helper, "VALID_EVENTS", None) - if isinstance(valid, set) and "approval-acted" not in valid: - valid.add("approval-acted") + event_name = event.get("event") + if isinstance(valid, set) and event_name and event_name not in valid: + valid.add(event_name) try: await asyncio.to_thread(helper.append_audit, record, None) @@ -673,6 +699,7 @@ async def _fanout( channels: list[str] | None = None, silent: bool = False, option_labels: dict[str, str] | None = None, + extra_web: dict[str, Any] | None = None, ) -> None: """Deliver notification to all configured channels in parallel. @@ -681,6 +708,10 @@ async def _fanout( as the answer string) to the human-facing label rendered on the button. ``None`` for the legacy ``question`` path, where the label and the value are identical. + + ``extra_web`` merges additional fields into the web broadcast + payload only ā€” the re-delivery tick uses it to flag + ``redelivered: true`` so the UI can badge a resurfaced card. """ target_channels = channels or self.config.notifications.channels @@ -692,6 +723,7 @@ async def _deliver(channel_name: str) -> str | None: notification_id, session_id, notif_type, title, body, priority, options, option_labels=option_labels, + extra=extra_web, ) return "web" elif channel_name == "telegram": @@ -734,13 +766,15 @@ async def _deliver_web( priority: str, options: list[str] | None, option_labels: dict[str, str] | None = None, + extra: dict[str, Any] | None = None, ) -> None: """Broadcast notification to web UI via the global broadcaster. For approval-kind rows we also include ``option_labels`` so the web NotificationCard can render readable button text while the button click still sends the canonical ``value`` back through - the answer endpoint. + the answer endpoint. ``extra`` fields (e.g. ``redelivered``) + are merged into the payload verbatim. """ from nerve.agent.streaming import broadcaster message = { @@ -755,6 +789,8 @@ async def _deliver_web( } if option_labels: message["option_labels"] = option_labels + if extra: + message.update(extra) await broadcaster.broadcast("__global__", message) async def _broadcast_silenced_web( @@ -823,6 +859,26 @@ def _get_telegram_bot(self): return None return channel._app.bot + def _build_telegram_text( + self, session_id: str, title: str, body: str, priority: str, + ) -> str: + """Compose the Telegram message text for a notification. + + Shared by the initial delivery, the re-delivery tick, and the + expiry edit (which rebuilds the original text to append a + status line). + """ + priority_prefix = self.config.notifications.priority_prefixes.get(priority, "") + if title: + text = f"{priority_prefix}{title}" + if body: + text += f"\n\n{body}" + else: + text = body or "" + if self._should_show_session_label(session_id): + text += f"\n\nSession: {session_id}" + return text + async def _deliver_telegram( self, notification_id: str, @@ -845,16 +901,7 @@ async def _deliver_telegram( if not chat_id: return None - # Build message text - priority_prefix = self.config.notifications.priority_prefixes.get(priority, "") - if title: - text = f"{priority_prefix}{title}" - if body: - text += f"\n\n{body}" - else: - text = body or "" - if self._should_show_session_label(session_id): - text += f"\n\nSession: {session_id}" + text = self._build_telegram_text(session_id, title, body, priority) if notif_type in ("question", "approval") and options: button_labels: list[tuple[str, str]] = [] @@ -960,9 +1007,276 @@ async def _send_telegram_inline( return str(msg.message_id) # ------------------------------------------------------------------ # - # Expiry (called by periodic background task) # + # Maintenance (called by the periodic background tick) # # ------------------------------------------------------------------ # + async def redeliver_due(self) -> int: + """Re-fan-out pending rows whose ``redeliver_at`` has passed. + + This is the re-delivery tick that makes "Snooze 24h" a real + round trip: the snoozed row resurfaces as a fresh Telegram card + (new inline keyboard, new message id stored) and a fresh web + broadcast flagged ``redelivered: true``. Each cycle bumps + ``redelivery_count``; at ``config.notifications.max_redeliveries`` + the row expires (with reporting) instead of re-sending, so an + auto-snoozing loop can't nag forever. + + Returns the number of notifications re-delivered. + """ + rows = await self.db.get_due_redeliveries() + if not rows: + return 0 + + max_redeliveries = self.config.notifications.max_redeliveries + redelivered = 0 + capped: list[dict[str, Any]] = [] + + for notif in rows: + if (notif.get("redelivery_count") or 0) >= max_redeliveries: + capped.append(notif) + continue + try: + await self._redeliver_one(notif) + redelivered += 1 + except Exception as exc: # defensive: one bad row ā‰  dead tick + logger.error( + "re-delivery failed for %s: %s", notif.get("id"), exc, + ) + + if capped: + expired: list[dict[str, Any]] = [] + for notif in capped: + if await self.db.expire_notification(notif["id"]): + row = dict(notif) + row["status"] = "expired" + expired.append(row) + logger.info( + "%d snoozed notification(s) hit the re-delivery cap (%d) " + "and expired: %s", + len(expired), max_redeliveries, + ", ".join(r["id"] for r in expired), + ) + await self._report_expired(expired) + + return redelivered + + async def _redeliver_one(self, notif: dict[str, Any]) -> None: + """Fan a single snoozed row back out to its channels.""" + notification_id = notif["id"] + options = json.loads(notif["options"]) if notif.get("options") else None + try: + metadata = json.loads(notif["metadata"]) if notif.get("metadata") else {} + except (TypeError, ValueError): + metadata = {} + option_labels = metadata.get("option_labels") or None + new_count = (notif.get("redelivery_count") or 0) + 1 + + # Restart the expiry window: a resurfaced card needs a full + # answering window, and (edge case) a row whose original expiry + # already passed must not be reaped by the expire pass of the + # very sweep that just re-delivered it. + new_expires_at = ( + datetime.now(timezone.utc) + + timedelta(hours=self.config.notifications.default_expiry_hours) + ).isoformat() + + # Bump the counter *before* the fanout so a mid-fanout crash + # can't replay the send every 15 minutes. + marked = await self.db.mark_notification_redelivered( + notification_id, new_expires_at, + ) + if not marked: + return # answered/expired between select and mark + + logger.info( + "Re-delivering snoozed notification %s (cycle %d)", + notification_id, new_count, + ) + await self._fanout( + notification_id, + notif["session_id"], + notif["type"], + notif.get("title") or "", + notif.get("body") or "", + notif.get("priority") or "normal", + options=options, + option_labels=option_labels, + extra_web={"redelivered": True, "redelivery_count": new_count}, + ) + async def expire_stale(self) -> int: - """Expire pending notifications past their expiry time.""" - return await self.db.expire_notifications() + """Expire pending notifications past their expiry time. + + Unlike the original blind status flip, every expired + ``question``/``approval`` is reported: the asking session gets + an internal note (questions), the mechanical-actions audit log + gets an ``approval-expired`` event (approvals), the web UI gets + a ``notification_expired`` broadcast, and the Telegram card is + edited to show it expired. ``notify``-kind expiry stays silent. + """ + rows = await self.db.expire_due_notifications() + if rows: + await self._report_expired(rows) + return len(rows) + + async def _report_expired(self, rows: list[dict[str, Any]]) -> None: + """Report expired questions/approvals to every interested party. + + Silent-by-construction expiry was the bug: the asking session + believed the user simply never replied, and the user's card just + vanished. Each reporting leg is individually best-effort so one + failure (e.g. a Telegram edit on an old message) never blocks + the others. + """ + reportable = [ + r for r in rows if r.get("type") in ("question", "approval") + ] + if not reportable: + return + + from nerve.agent.streaming import broadcaster + + for notif in reportable: + # Web: gray the card live. + try: + await broadcaster.broadcast("__global__", { + "type": "notification_expired", + "notification_id": notif["id"], + "session_id": notif["session_id"], + "notification_type": notif["type"], + "title": notif.get("title") or "", + }) + except Exception as exc: # pragma: no cover - defensive + logger.warning( + "expiry broadcast failed for %s: %s", notif["id"], exc, + ) + + # Telegram: mark the card expired, drop dead buttons. + await self._edit_telegram_expired(notif) + + # Approvals: the proposer is the mechanical pipeline, not a + # conversation ā€” record the expiry in its audit log. + if notif.get("type") == "approval": + await self._append_approval_audit({ + "event": "approval-expired", + "notification_id": notif["id"], + "target_kind": notif.get("target_kind") or "", + "target_id": notif.get("target_id") or "", + "redelivery_count": notif.get("redelivery_count") or 0, + }) + + # Questions: tell the asking session its question died, so the + # agent can adapt (re-ask, escalate, or record "no decision"). + # Batched per session ā€” one injection per sweep, not per row. + questions = [r for r in reportable if r.get("type") == "question"] + by_session: dict[str, list[dict[str, Any]]] = {} + for notif in questions: + by_session.setdefault(notif["session_id"], []).append(notif) + for session_id, session_rows in by_session.items(): + await self._inject_expiry_note(session_id, session_rows) + + async def _inject_expiry_note( + self, session_id: str, rows: list[dict[str, Any]], + ) -> None: + """Inject an expired-unanswered note into the asking session. + + Dispatched unconditionally ā€” ``engine.run`` serializes per + session, so a mid-turn session just processes the note when its + current turn finishes (same pattern as the wakeup dispatcher; + deliberately NOT the stale ``is_running`` skip that drops + answers). Skipped for external (satellite) sessions, whose + conversation loop Nerve doesn't own, and archived/missing + sessions ā€” those got the web broadcast and nothing more. + """ + try: + session = await self.db.get_session(session_id) + except Exception as exc: # pragma: no cover - defensive + logger.warning( + "expiry injection: session lookup failed for %s: %s", + session_id, exc, + ) + return + if not session: + logger.info( + "expiry injection: session %s missing ā€” broadcast only", + session_id, + ) + return + if session.get("status") == "archived": + logger.info( + "expiry injection: session %s archived ā€” broadcast only", + session_id, + ) + return + if session.get("source") == "external": + logger.info( + "expiry injection: session %s is external ā€” broadcast only", + session_id, + ) + return + + if len(rows) == 1: + message = ( + f"[Question expired unanswered: {rows[0].get('title') or ''}]" + ) + else: + titles = "\n".join( + f"- {r.get('title') or ''}" for r in rows + ) + message = f"[Questions expired unanswered]\n{titles}" + + task = asyncio.create_task( + self.engine.run( + session_id=session_id, + user_message=message, + source="notification:expiry", + internal=True, + ) + ) + task.add_done_callback(self._on_answer_task_done) + + async def _edit_telegram_expired(self, notif: dict[str, Any]) -> None: + """Best-effort edit of the Telegram card to show it expired. + + Rebuilds the original message text from the row (same builder + the delivery used) and appends the status line, dropping the + now-dead inline keyboard. Telegram refuses edits on old + messages (>48h) ā€” all failures are swallowed by design. + """ + message_id = notif.get("telegram_message_id") + if not message_id: + return + bot = self._get_telegram_bot() + if not bot: + return + chat_id = notif.get("telegram_chat_id") or self._resolve_telegram_chat_id() + if not chat_id: + return + + text = self._build_telegram_text( + notif["session_id"], + notif.get("title") or "", + notif.get("body") or "", + notif.get("priority") or "normal", + ) + text += "\n\nā° Expired unanswered" + + from nerve.channels.telegram import _md_to_tg_html + from telegram.constants import ParseMode + + try: + await bot.edit_message_text( + chat_id=int(chat_id), message_id=int(message_id), + text=_md_to_tg_html(text), parse_mode=ParseMode.HTML, + ) + except Exception: + try: + await bot.edit_message_text( + chat_id=int(chat_id), message_id=int(message_id), + text=text, + ) + except Exception as exc: + logger.debug( + "telegram expiry edit failed for %s: %s", + notif["id"], exc, + ) diff --git a/tests/test_notification_lifecycle.py b/tests/test_notification_lifecycle.py new file mode 100644 index 0000000..e20ec8f --- /dev/null +++ b/tests/test_notification_lifecycle.py @@ -0,0 +1,562 @@ +"""Tests for the notification snooze/re-delivery + expiry-reporting lifecycle. + +Closes the loop that shipped half-built with the ``approval`` kind: + +- v037 migration adds ``redeliver_at`` / ``redelivery_count``. +- Snoozing stamps ``redeliver_at`` (and pushes ``expires_at`` past it) + instead of just delaying a silent expiry. +- ``NotificationService.redeliver_due`` (the periodic maintenance tick) + fans snoozed rows back out with a fresh card, up to + ``config.notifications.max_redeliveries`` cycles. +- ``NotificationService.expire_stale`` now *reports* every expired + question/approval: session injection for questions, audit event for + approvals, a ``notification_expired`` web broadcast for both. + ``notify``-kind expiry stays silent. + +Fixture style mirrors ``test_notifications_actionable``: fresh SQLite +per test, stubbed broadcaster + engine. +""" + +from __future__ import annotations + +import asyncio +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nerve.config import NerveConfig, NotificationsConfig +from nerve.db import Database +from nerve.notifications import handlers as _handlers +from nerve.notifications.service import NotificationService + +from tests.test_notifications_actionable import ( + _MINIMAL_HELPER_SRC, + read_audit_jsonl, +) + + +# ---------------------------------------------------------------------- +# Fixtures +# ---------------------------------------------------------------------- + + +@pytest.fixture +def fake_config(tmp_path: Path) -> NerveConfig: + """Minimal NerveConfig with workspace + notifications config wired.""" + cfg = NerveConfig() + cfg.workspace = tmp_path + cfg.notifications = NotificationsConfig( + channels=["web"], # skip telegram in unit tests + telegram_chat_id=None, + default_expiry_hours=48, + max_redeliveries=3, + priority_prefixes={"high": "", "urgent": ""}, + ) + return cfg + + +@pytest.fixture +def fake_engine() -> MagicMock: + """An engine stub with the minimum surface the service touches.""" + engine = MagicMock() + engine.sessions = MagicMock() + engine.sessions.is_running.return_value = False + engine.router = MagicMock() + engine.router.get_channel.return_value = None + engine.run = AsyncMock() + return engine + + +@pytest.fixture +def patch_broadcaster(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, dict]]: + """Capture broadcaster.broadcast() calls instead of hitting any WS.""" + captured: list[tuple[str, dict]] = [] + + class _FakeBroadcaster: + async def broadcast(self, channel: str, message: dict) -> None: + captured.append((channel, message)) + + from nerve.agent import streaming + monkeypatch.setattr(streaming, "broadcaster", _FakeBroadcaster()) + return captured + + +@pytest.fixture +def audit_workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Workspace with just the ``_mechanical_action.py`` audit helper. + + Enough for ``_append_approval_audit`` (expiry reporting) ā€” the + dispatcher's ``mechanical-action.sh`` is not needed because these + tests never route a live approve/decline through the shell script. + """ + scripts_dir = tmp_path / "scripts" + scripts_dir.mkdir() + (scripts_dir / "_mechanical_action.py").write_text(_MINIMAL_HELPER_SRC) + monkeypatch.setenv("NERVE_WORKSPACE_PATH", str(tmp_path)) + monkeypatch.setenv( + "NERVE_MECHANICAL_STATE_DIR", + str(tmp_path / ".nerve" / "mechanical-actions"), + ) + return tmp_path + + +def _iso(delta_hours: float) -> str: + """ISO timestamp ``delta_hours`` from now (negative = past).""" + return ( + datetime.now(timezone.utc) + timedelta(hours=delta_hours) + ).isoformat() + + +def _snooze_dispatcher(snooze_until: str): + """A dispatcher that always answers "snooze until ".""" + def dispatch(notification, target_id, decision, config): + return _handlers.DispatchResult( + ok=True, + audit_event={ + "event": "approval-acted", + "notification_id": notification.get("id", ""), + "target_kind": "lifecycle-test", + "target_id": target_id, + "decision": decision, + "ok": True, + }, + snooze_until=snooze_until, + ) + return dispatch + + +async def _make_approval( + svc: NotificationService, db: Database, session_id: str = "s1", + target_kind: str = "lifecycle-test", **kwargs, +) -> str: + result = await svc.propose_action( + session_id=session_id, + target_kind=target_kind, + target_id="prop-1", + title=kwargs.pop("title", "approve the thing"), + **kwargs, + ) + return result["notification_id"] + + +# ---------------------------------------------------------------------- +# Migration / schema +# ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestSchema: + async def test_v037_columns_exist_with_defaults(self, db: Database): + async with db.db.execute("PRAGMA table_info(notifications)") as cur: + cols = {row[1] async for row in cur} + assert "redeliver_at" in cols + assert "redelivery_count" in cols + + # Old-style insert (no new columns touched) ā†’ sane defaults. + await db.create_session("s1") + await db.create_notification( + notification_id="n1", session_id="s1", + type="question", title="t", + ) + notif = await db.get_notification("n1") + assert notif["redeliver_at"] is None + assert notif["redelivery_count"] == 0 + + +# ---------------------------------------------------------------------- +# Snooze semantics +# ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestSnooze: + async def test_snooze_sets_redeliver_at_and_pushes_expiry( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + """Snooze = redeliver_at stamped, expiry pushed PAST the + re-delivery time, row stays pending.""" + snooze_until = _iso(24) + _handlers.register("lifecycle-test", _snooze_dispatcher(snooze_until)) + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + nid = await _make_approval(svc, db, expiry_hours=2) + + ok = await svc.handle_answer(nid, "snooze_24h", "web") + assert ok is True + + notif = await db.get_notification(nid) + assert notif["status"] == "pending" + assert notif["redeliver_at"] == snooze_until + assert notif["answer"] is None + # expires_at = snooze_until + default_expiry_hours: cannot die + # before it resurfaces. + assert notif["expires_at"] > notif["redeliver_at"] + + # Broadcast carries the snoozed state + timestamp for the card. + answered = [ + m for _, m in patch_broadcaster + if m.get("type") == "notification_answered" + and m.get("notification_id") == nid + ] + assert answered and answered[0]["approval_status"] == "snoozed" + assert answered[0]["snooze_until"] == snooze_until + + async def test_db_snooze_rejects_non_pending(self, db: Database): + await db.create_session("s1") + await db.create_notification( + notification_id="n1", session_id="s1", type="approval", title="t", + ) + await db.answer_notification("n1", "approve", "web") + assert await db.snooze_notification("n1", _iso(24), _iso(72)) is False + + +# ---------------------------------------------------------------------- +# Re-delivery tick +# ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestRedelivery: + async def test_tick_redelivers_due_snoozed_row( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + _handlers.register("lifecycle-test", _snooze_dispatcher(_iso(-0.5))) + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + nid = await _make_approval(svc, db) + await svc.handle_answer(nid, "snooze_24h", "web") + patch_broadcaster.clear() + + redelivered = await svc.redeliver_due() + assert redelivered == 1 + + notif = await db.get_notification(nid) + assert notif["status"] == "pending" + assert notif["redeliver_at"] is None # consumed + assert notif["redelivery_count"] == 1 + # Fresh expiry window for the fresh card. + assert notif["expires_at"] > _iso(47) + + # Web fanout fired again, flagged as a re-delivery, with the + # original options + labels intact. + fanouts = [ + m for _, m in patch_broadcaster + if m.get("type") == "notification" + and m.get("notification_id") == nid + ] + assert len(fanouts) == 1 + assert fanouts[0]["redelivered"] is True + assert fanouts[0]["redelivery_count"] == 1 + assert fanouts[0]["options"] == ["approve", "decline", "snooze_24h"] + assert fanouts[0]["option_labels"]["snooze_24h"] == "Snooze 24h" + + async def test_not_due_rows_are_left_alone( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + _handlers.register("lifecycle-test", _snooze_dispatcher(_iso(24))) + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + nid = await _make_approval(svc, db) + await svc.handle_answer(nid, "snooze_24h", "web") + + assert await svc.redeliver_due() == 0 + notif = await db.get_notification(nid) + assert notif["redeliver_at"] is not None + assert notif["redelivery_count"] == 0 + + async def test_resnooze_after_redelivery_second_cycle( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + """Snooze is repeatable: each click buys another cycle.""" + _handlers.register("lifecycle-test", _snooze_dispatcher(_iso(-1))) + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + nid = await _make_approval(svc, db) + + await svc.handle_answer(nid, "snooze_24h", "web") + assert await svc.redeliver_due() == 1 + # User snoozes the re-delivered card again. + ok = await svc.handle_answer(nid, "snooze_24h", "web") + assert ok is True + notif = await db.get_notification(nid) + assert notif["status"] == "pending" + assert notif["redeliver_at"] is not None + + assert await svc.redeliver_due() == 1 + notif = await db.get_notification(nid) + assert notif["redelivery_count"] == 2 + assert notif["redeliver_at"] is None + + async def test_cap_reached_expires_with_report_no_resend( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + audit_workspace: Path, + ): + _handlers.register("lifecycle-test", _snooze_dispatcher(_iso(-1))) + fake_config.notifications.max_redeliveries = 2 + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + nid = await _make_approval(svc, db) + + for cycle in range(2): + await svc.handle_answer(nid, "snooze_24h", "web") + assert await svc.redeliver_due() == 1, f"cycle {cycle}" + + # Third snooze hits the cap: tick expires instead of re-sending. + await svc.handle_answer(nid, "snooze_24h", "web") + patch_broadcaster.clear() + assert await svc.redeliver_due() == 0 + + notif = await db.get_notification(nid) + assert notif["status"] == "expired" + assert notif["redelivery_count"] == 2 + + # No new fanout, but the expiry was reported to the web... + fanouts = [ + m for _, m in patch_broadcaster if m.get("type") == "notification" + ] + assert not fanouts + expired_events = [ + m for _, m in patch_broadcaster + if m.get("type") == "notification_expired" + and m.get("notification_id") == nid + ] + assert expired_events + + # ...and to the mechanical-actions audit log. + events = read_audit_jsonl( + audit_workspace / ".nerve" / "mechanical-actions", + ) + expired = [e for e in events if e.get("event") == "approval-expired"] + assert any( + e.get("notification_id") == nid + and e.get("redelivery_count") == 2 + for e in expired + ) + + async def test_row_due_for_both_redelivery_and_expiry_survives( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + """Ordering guarantee: redeliver-before-expire means a row whose + redeliver_at AND expires_at both passed gets its last chance.""" + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + await db.create_notification( + notification_id="n1", session_id="s1", type="approval", + title="both due", options=["approve", "decline", "snooze_24h"], + expires_at=_iso(-1), + target_kind="lifecycle-test", target_id="x", + ) + await db.update_notification("n1", redeliver_at=_iso(-2)) + + # Maintenance-tick order: redeliver first, then expire. + assert await svc.redeliver_due() == 1 + assert await svc.expire_stale() == 0 + + notif = await db.get_notification("n1") + assert notif["status"] == "pending" # survived, re-delivered + assert notif["redelivery_count"] == 1 + assert notif["expires_at"] > _iso(0) # expiry window restarted + + +# ---------------------------------------------------------------------- +# Expiry reporting +# ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestExpiryReporting: + async def _expired_question( + self, db: Database, svc: NotificationService, + session_id: str = "s1", title: str = "pick one", + ) -> str: + result = await svc.ask_question( + session_id=session_id, title=title, options=["yes", "no"], + ) + nid = result["notification_id"] + await db.update_notification(nid, expires_at=_iso(-1)) + return nid + + async def test_expired_question_injected_into_origin_session( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + nid = await self._expired_question(db, svc) + # The origin session is mid-turn: injection must STILL be + # dispatched (the per-session lock serializes it) ā€” this is + # deliberately not handle_answer's is_running skip. + fake_engine.sessions.is_running.return_value = True + + assert await svc.expire_stale() == 1 + await asyncio.sleep(0) # let the fire-and-forget task start + + notif = await db.get_notification(nid) + assert notif["status"] == "expired" + + fake_engine.run.assert_called_once() + kwargs = fake_engine.run.call_args.kwargs + assert kwargs["session_id"] == "s1" + assert kwargs["source"] == "notification:expiry" + assert kwargs["internal"] is True + assert "pick one" in kwargs["user_message"] + assert "expired unanswered" in kwargs["user_message"].lower() + + expired_events = [ + m for ch, m in patch_broadcaster + if m.get("type") == "notification_expired" and ch == "__global__" + ] + assert [e["notification_id"] for e in expired_events] == [nid] + + async def test_multiple_questions_same_session_single_injection( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + await self._expired_question(db, svc, title="first question") + await self._expired_question(db, svc, title="second question") + + assert await svc.expire_stale() == 2 + await asyncio.sleep(0) + + fake_engine.run.assert_called_once() + msg = fake_engine.run.call_args.kwargs["user_message"] + assert "first question" in msg + assert "second question" in msg + + async def test_external_session_broadcast_only( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("sat-1", source="external") + svc = NotificationService(fake_config, db, fake_engine) + nid = await self._expired_question(db, svc, session_id="sat-1") + + assert await svc.expire_stale() == 1 + await asyncio.sleep(0) + + fake_engine.run.assert_not_called() + assert any( + m.get("type") == "notification_expired" + and m.get("notification_id") == nid + for _, m in patch_broadcaster + ) + + async def test_archived_and_missing_sessions_broadcast_only( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("s1", status="archived") + svc = NotificationService(fake_config, db, fake_engine) + await self._expired_question(db, svc, session_id="s1") + # A question whose session row vanished entirely. + await db.create_notification( + notification_id="ghost-q", session_id="no-such-session", + type="question", title="orphan", expires_at=_iso(-1), + ) + + assert await svc.expire_stale() == 2 + await asyncio.sleep(0) + + fake_engine.run.assert_not_called() + expired_ids = { + m.get("notification_id") for _, m in patch_broadcaster + if m.get("type") == "notification_expired" + } + assert "ghost-q" in expired_ids + + async def test_expired_approval_audits_no_injection( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + audit_workspace: Path, + ): + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + nid = await _make_approval(svc, db) + await db.update_notification(nid, expires_at=_iso(-1)) + + assert await svc.expire_stale() == 1 + await asyncio.sleep(0) + + fake_engine.run.assert_not_called() # approvals never inject + events = read_audit_jsonl( + audit_workspace / ".nerve" / "mechanical-actions", + ) + expired = [e for e in events if e.get("event") == "approval-expired"] + assert any( + e.get("notification_id") == nid + and e.get("target_kind") == "lifecycle-test" + and e.get("target_id") == "prop-1" + for e in expired + ) + + async def test_notify_kind_expiry_stays_silent( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + await db.create_notification( + notification_id="fyi-1", session_id="s1", + type="notify", title="fyi", expires_at=_iso(-1), + ) + + assert await svc.expire_stale() == 1 + await asyncio.sleep(0) + + notif = await db.get_notification("fyi-1") + assert notif["status"] == "expired" + fake_engine.run.assert_not_called() + assert not [ + m for _, m in patch_broadcaster + if m.get("type") == "notification_expired" + ] + + async def test_expiry_edits_telegram_card( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + """Best-effort Telegram edit: expired card gets the status line + and loses its (dead) inline keyboard.""" + bot = MagicMock() + bot.edit_message_text = AsyncMock() + channel = MagicMock() + channel._app.bot = bot + fake_engine.router.get_channel.return_value = channel + + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + nid = await self._expired_question(db, svc, title="tg question") + await db.update_notification( + nid, telegram_message_id="4242", telegram_chat_id="1001", + ) + + assert await svc.expire_stale() == 1 + + bot.edit_message_text.assert_awaited_once() + kwargs = bot.edit_message_text.await_args.kwargs + assert kwargs["chat_id"] == 1001 + assert kwargs["message_id"] == 4242 + assert "Expired unanswered" in kwargs["text"] + + async def test_telegram_edit_failure_is_swallowed( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + """Telegram refuses edits on >48h-old messages ā€” expiry must + still complete.""" + bot = MagicMock() + bot.edit_message_text = AsyncMock(side_effect=RuntimeError("too old")) + channel = MagicMock() + channel._app.bot = bot + fake_engine.router.get_channel.return_value = channel + + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + nid = await self._expired_question(db, svc) + await db.update_notification( + nid, telegram_message_id="4242", telegram_chat_id="1001", + ) + + assert await svc.expire_stale() == 1 + notif = await db.get_notification(nid) + assert notif["status"] == "expired" + # HTML attempt + plain-text fallback, both swallowed. + assert bot.edit_message_text.await_count == 2 diff --git a/tests/test_notifications_actionable.py b/tests/test_notifications_actionable.py index 94bf925..a84564b 100644 --- a/tests/test_notifications_actionable.py +++ b/tests/test_notifications_actionable.py @@ -234,7 +234,7 @@ async def test_create_notification_with_target(self, db: Database): assert notif["target_id"] == "20260519T143906Z-d2e62e" assert notif["type"] == "approval" - async def test_snooze_notification_advances_expiry(self, db: Database): + async def test_snooze_notification_queues_redelivery(self, db: Database): await db.create_session("s1") future = ( datetime.now(timezone.utc) + timedelta(hours=1) @@ -244,12 +244,16 @@ async def test_snooze_notification_advances_expiry(self, db: Database): type="approval", title="t", expires_at=future, ) - new_expiry = ( + redeliver_at = ( datetime.now(timezone.utc) + timedelta(hours=24) ).isoformat() - ok = await db.snooze_notification("n1", new_expiry) + new_expiry = ( + datetime.now(timezone.utc) + timedelta(hours=72) + ).isoformat() + ok = await db.snooze_notification("n1", redeliver_at, new_expiry) assert ok is True notif = await db.get_notification("n1") + assert notif["redeliver_at"] == redeliver_at assert notif["expires_at"] == new_expiry assert notif["status"] == "pending" @@ -259,10 +263,15 @@ async def test_snooze_notification_rejects_non_pending(self, db: Database): notification_id="n1", session_id="s1", type="approval", title="t", ) await db.answer_notification("n1", "approve", "web") - new_expiry = ( + redeliver_at = ( datetime.now(timezone.utc) + timedelta(hours=24) ).isoformat() - assert await db.snooze_notification("n1", new_expiry) is False + new_expiry = ( + datetime.now(timezone.utc) + timedelta(hours=72) + ).isoformat() + assert await db.snooze_notification( + "n1", redeliver_at, new_expiry, + ) is False # ---------------------------------------------------------------------- @@ -443,6 +452,10 @@ async def test_snooze_keeps_pending_and_advances_expiry( assert after["expires_at"] is not None # Expiry advanced forward; sanity check it is not the original. assert after["expires_at"] != prior_expiry + # Queued for re-delivery by the maintenance tick, with the + # expiry pushed past the re-delivery time. + assert after["redeliver_at"] is not None + assert after["expires_at"] > after["redeliver_at"] # And no answer recorded (snooze is not a final answer). assert after["answer"] is None diff --git a/web/src/api/websocket.ts b/web/src/api/websocket.ts index fcb6dae..3413028 100644 --- a/web/src/api/websocket.ts +++ b/web/src/api/websocket.ts @@ -20,8 +20,9 @@ export type WSMessage = | { type: 'subagent_start'; session_id: string; tool_use_id: string; subagent_type: string; description: string; model?: string } | { type: 'subagent_complete'; session_id: string; tool_use_id: string; duration_ms: number; is_error?: boolean } | { type: 'file_changed'; session_id: string; path: string; operation: string; tool_use_id: string } - | { type: 'notification'; notification_id: string; notification_type: 'notify' | 'question' | 'approval'; session_id: string; title: string; body: string; priority: string; options: string[] | null; option_labels?: Record; target_kind?: string; target_id?: string; silenced?: boolean; silence_reason?: string; silence_pattern?: string; silenced_by?: string } - | { type: 'notification_answered'; notification_id: string; session_id: string; answer: string; answered_by: string; approval_status?: 'answered' | 'snoozed'; dispatch_ok?: boolean } + | { type: 'notification'; notification_id: string; notification_type: 'notify' | 'question' | 'approval'; session_id: string; title: string; body: string; priority: string; options: string[] | null; option_labels?: Record; target_kind?: string; target_id?: string; silenced?: boolean; silence_reason?: string; silence_pattern?: string; silenced_by?: string; redelivered?: boolean; redelivery_count?: number } + | { type: 'notification_answered'; notification_id: string; session_id: string; answer: string; answered_by: string; approval_status?: 'answered' | 'snoozed'; dispatch_ok?: boolean; snooze_until?: string } + | { type: 'notification_expired'; notification_id: string; session_id: string; notification_type: string; title: string } | { type: 'answer_injected'; session_id: string; notification_id: string; title: string; answer: string; answered_by: string; content: string } | { type: 'session_running'; session_id: string; is_running: boolean } | { type: 'session_awaiting_input'; session_id: string; awaiting: boolean } diff --git a/web/src/pages/NotificationsPage.tsx b/web/src/pages/NotificationsPage.tsx index 4f66e6e..e9acee9 100644 --- a/web/src/pages/NotificationsPage.tsx +++ b/web/src/pages/NotificationsPage.tsx @@ -1,6 +1,6 @@ import { useEffect, useState } from 'react'; import { useNavigate } from 'react-router-dom'; -import { Bell, X, CheckCheck, EyeOff, Check, XCircle, Moon, BellOff, Trash2, Plus } from 'lucide-react'; +import { Bell, X, CheckCheck, EyeOff, Check, XCircle, Moon, BellOff, Trash2, Plus, RotateCw, Clock } from 'lucide-react'; import { useNotificationStore, type Notification, type Silence } from '../stores/notificationStore'; const STATUS_STYLES: Record = { @@ -118,6 +118,14 @@ function formatExpiry(expiresAt: string | null): string { return `expires ${expiresAt.slice(0, 16).replace('T', ' ')}`; } +function formatLocalTime(iso: string): string { + const d = new Date(iso); + if (isNaN(d.getTime())) return iso.slice(0, 16).replace('T', ' '); + return d.toLocaleString(undefined, { + month: 'short', day: 'numeric', hour: '2-digit', minute: '2-digit', + }); +} + function FreeTextInput({ onSubmit }: { onSubmit: (text: string) => void }) { const [text, setText] = useState(''); const [open, setOpen] = useState(false); @@ -201,6 +209,15 @@ function NotificationCard({ notif }: { notif: Notification }) { )}
+ {(notif.redelivery_count ?? 0) > 0 && ( + + + {(notif.redelivery_count ?? 0) > 1 ? `Ɨ${notif.redelivery_count}` : 're-delivered'} + + )} {notif.status} @@ -267,6 +284,16 @@ function NotificationCard({ notif }: { notif: Notification }) {
)} + {/* Snoozed: still pending server-side, will be re-delivered */} + {notif.status === 'pending' && notif.redeliver_at && ( +
+ + + Snoozed ā€” returns {formatLocalTime(notif.redeliver_at)}. You can still decide now. + +
+ )} + {/* Show answer if answered */} {notif.status === 'answered' && (
diff --git a/web/src/stores/chatStore.ts b/web/src/stores/chatStore.ts index f4f0b16..15bfbf1 100644 --- a/web/src/stores/chatStore.ts +++ b/web/src/stores/chatStore.ts @@ -12,7 +12,7 @@ import { extractTodosFromMessages, extractCCTasksFromMessages } from './helpers/ import { handleThinking, handleToken, handleToolUse, handleToolResult, handleDone, handleStopped, handleError, handleWakeup, handleAutoTurn, handleModelChanged } from './handlers/streamingHandlers'; import { handleSessionUpdated, handleSessionStatus, handleSessionSwitched, handleSessionForked, handleSessionResumed, handleSessionArchived, handleSessionRunning, handleSessionAwaitingInput, handleAnswerInjected } from './handlers/sessionHandlers'; import { handlePlanUpdate, handleSubagentStart, handleSubagentComplete, handleHoaProgress, handleWorkflowProgress } from './handlers/panelHandlers'; -import { handleInteraction, handleFileChanged, handleNotification, handleNotificationAnswered, handleBackgroundTasksUpdate } from './handlers/auxiliaryHandlers'; +import { handleInteraction, handleFileChanged, handleNotification, handleNotificationAnswered, handleNotificationExpired, handleBackgroundTasksUpdate } from './handlers/auxiliaryHandlers'; export interface TodoItem { content: string; @@ -744,6 +744,7 @@ export const useChatStore = create((set, get) => ({ case 'file_changed': return handleFileChanged(msg, get, set); case 'notification': return handleNotification(msg, get, set); case 'notification_answered': return handleNotificationAnswered(msg, get, set); + case 'notification_expired': return handleNotificationExpired(msg, get, set); case 'background_tasks_update': return handleBackgroundTasksUpdate(msg, get, set); } }, diff --git a/web/src/stores/handlers/auxiliaryHandlers.ts b/web/src/stores/handlers/auxiliaryHandlers.ts index 8755664..a435cf9 100644 --- a/web/src/stores/handlers/auxiliaryHandlers.ts +++ b/web/src/stores/handlers/auxiliaryHandlers.ts @@ -75,6 +75,16 @@ export function handleNotificationAnswered( ); } +export function handleNotificationExpired( + msg: Extract, + _get: Get, + _set: Set, +): void { + import('../notificationStore').then(({ useNotificationStore }) => + useNotificationStore.getState().handleWSNotificationExpired(msg) + ); +} + export function handleBackgroundTasksUpdate( msg: Extract, get: Get, diff --git a/web/src/stores/notificationStore.ts b/web/src/stores/notificationStore.ts index ed60b5c..11ac442 100644 --- a/web/src/stores/notificationStore.ts +++ b/web/src/stores/notificationStore.ts @@ -16,6 +16,11 @@ export interface Notification { answered_at: string | null; created_at: string; expires_at: string | null; + // Set while a snoozed row waits for the maintenance tick to fan it + // back out. Present on pending rows only; cleared on re-delivery. + redeliver_at?: string | null; + // How many times the row has been re-delivered (snooze round trips). + redelivery_count?: number; target_kind?: string | null; target_id?: string | null; // Optional label map for approval-kind rows: value -> human label. @@ -58,6 +63,7 @@ interface NotificationState { dismissAll: () => Promise; handleWSNotification: (data: any) => void; handleWSNotificationAnswered: (data: any) => void; + handleWSNotificationExpired: (data: any) => void; dismissToast: (id: string) => void; loadSilences: () => Promise; addSilence: (pattern: string, reason: string, ttlHours: number) => Promise; @@ -127,6 +133,7 @@ export const useNotificationStore = create((set, get) => ({ handleWSNotification: (data: any) => { const silenced = data.silenced === true; + const redelivered = data.redelivered === true; const notif: Notification = { id: data.notification_id, session_id: data.session_id, @@ -142,6 +149,8 @@ export const useNotificationStore = create((set, get) => ({ answered_at: null, created_at: new Date().toISOString(), expires_at: null, + redeliver_at: null, + redelivery_count: data.redelivery_count ?? 0, target_kind: data.target_kind ?? null, target_id: data.target_id ?? null, option_labels: data.option_labels ?? null, @@ -156,16 +165,43 @@ export const useNotificationStore = create((set, get) => ({ : null, }; - set(s => ({ - notifications: [notif, ...s.notifications], - // A silenced notification was NOT delivered/escalated ā€” it does not - // count as pending and never raises a toast/sound. - pendingCount: silenced ? s.pendingCount : s.pendingCount + 1, - toastQueue: silenced ? s.toastQueue : [...s.toastQueue, notif], - })); + set(s => { + // A re-delivered (previously snoozed) row already exists in the + // list ā€” refresh it in place and move it to the top instead of + // duplicating. Its pending count was never decremented on snooze, + // so it doesn't count again. + const existing = redelivered + ? s.notifications.find(n => n.id === notif.id) + : undefined; + const rest = existing + ? s.notifications.filter(n => n.id !== notif.id) + : s.notifications; + return { + notifications: [existing ? { ...existing, ...notif } : notif, ...rest], + // A silenced notification was NOT delivered/escalated ā€” it does not + // count as pending and never raises a toast/sound. A re-delivered + // row was already pending (snooze never decremented it), so it + // must not count twice. + pendingCount: silenced || redelivered ? s.pendingCount : s.pendingCount + 1, + toastQueue: silenced ? s.toastQueue : [...s.toastQueue, notif], + }; + }); }, handleWSNotificationAnswered: (data: any) => { + // A snoozed approval stays pending server-side ā€” mirror that: keep + // the row actionable, stamp when it will resurface, leave the + // pending count alone (matches the server's pending_count). + if (data.approval_status === 'snoozed') { + set(s => ({ + notifications: s.notifications.map(n => + n.id === data.notification_id + ? { ...n, status: 'pending', redeliver_at: data.snooze_until ?? null } + : n + ), + })); + return; + } set(s => ({ notifications: s.notifications.map(n => n.id === data.notification_id @@ -176,6 +212,22 @@ export const useNotificationStore = create((set, get) => ({ })); }, + handleWSNotificationExpired: (data: any) => { + set(s => { + const wasPending = s.notifications.some( + n => n.id === data.notification_id && n.status === 'pending' + ); + return { + notifications: s.notifications.map(n => + n.id === data.notification_id ? { ...n, status: 'expired' } : n + ), + pendingCount: wasPending + ? Math.max(0, s.pendingCount - 1) + : s.pendingCount, + }; + }); + }, + dismissToast: (id: string) => { set(s => ({ toastQueue: s.toastQueue.filter(n => n.id !== id) })); },