From bc14103b13d8001855ed9b72486265a94d43a78f Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Sun, 23 Aug 2026 19:48:04 +0200 Subject: [PATCH 1/8] Deliver notifications through Slack Fan out questions and approvals as Block Kit cards, persist Slack delivery identifiers, attribute answers to the workspace member who acted, and keep expiry and re-delivery behavior consistent with the existing channels. --- nerve/config.py | 2 + .../db/migrations/v045_slack_notifications.py | 26 +++ nerve/notifications/service.py | 160 +++++++++++++++++- tests/test_db_migrations.py | 110 ++++++++++++ tests/test_notification_lifecycle.py | 120 +++++++++++++ tests/test_slack_channel.py | 7 + 6 files changed, 416 insertions(+), 9 deletions(-) create mode 100644 nerve/db/migrations/v045_slack_notifications.py create mode 100644 tests/test_db_migrations.py diff --git a/nerve/config.py b/nerve/config.py index 526156a6d..da292a21a 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -2054,6 +2054,7 @@ class NotificationsConfig: """Async notification delivery settings.""" channels: list[str] = field(default_factory=lambda: ["web", "telegram"]) telegram_chat_id: int | None = None # Target chat; falls back to first allowed_user + slack_channel_id: str = "" # Target conversation; falls back to a literal id in slack.allow_channels 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: { @@ -2072,6 +2073,7 @@ def from_dict(cls, d: dict) -> NotificationsConfig: return cls( channels=d.get("channels", ["web", "telegram"]), telegram_chat_id=d.get("telegram_chat_id"), + slack_channel_id=str(d.get("slack_channel_id") or ""), default_expiry_hours=d.get("default_expiry_hours", 48), max_redeliveries=d.get("max_redeliveries", 3), priority_prefixes=d.get("priority_prefixes", { diff --git a/nerve/db/migrations/v045_slack_notifications.py b/nerve/db/migrations/v045_slack_notifications.py new file mode 100644 index 000000000..070685f43 --- /dev/null +++ b/nerve/db/migrations/v045_slack_notifications.py @@ -0,0 +1,26 @@ +"""V45: Slack delivery ids on notifications.""" + +from __future__ import annotations + +import logging + +import aiosqlite + +logger = logging.getLogger(__name__) + +COLUMNS = ( + ("slack_message_id", "TEXT"), + ("slack_channel_id", "TEXT"), +) + + +async def up(db: aiosqlite.Connection) -> None: + cursor = await db.execute("PRAGMA table_info(notifications)") + existing = {row[1] for row in await cursor.fetchall()} + for name, decl in COLUMNS: + if name in existing: + continue + await db.execute( + f"ALTER TABLE notifications ADD COLUMN {name} {decl}", + ) + logger.info("V45 migration: notifications carries Slack delivery ids") diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index fc7e62726..d2e656c1f 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -2,7 +2,7 @@ Coordinates between MCP tools (agent-side), channels (delivery), and the answer routing mechanism (user-side). Supports fire-and-forget notifications, -async questions with multi-channel delivery (web UI + Telegram), and +async questions with multi-channel delivery (web UI + Telegram + Slack), and ``approval``-kind notifications that route to a server-side dispatcher when the user picks an inline option (see ``nerve.notifications.handlers``). """ @@ -850,6 +850,19 @@ async def _deliver(channel_name: str) -> str | None: telegram_message_id=str(msg_id), ) return "telegram" if msg_id else None + elif channel_name == "slack": + msg_id = await self._deliver_slack( + notification_id, session_id, notif_type, + title, body, priority, options, + option_labels=option_labels, + ) + if not msg_id: + return None + await self.db.update_notification( + notification_id, + slack_message_id=str(msg_id), + ) + return "slack" except Exception as e: logger.error( "Failed to deliver %s to %s: %s", @@ -972,14 +985,15 @@ def _get_telegram_bot(self): return None return channel._app.bot - def _build_telegram_text( + def _build_notification_text( self, session_id: str, title: str, body: str, priority: str, ) -> str: - """Compose the Telegram message text for a notification. + """Compose the chat 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). + Shared by Telegram and Slack, and within each by the initial + delivery, the re-delivery tick, and the expiry edit (which rebuilds + the original text to append a status line). Markdown is converted + per channel at send time. """ priority_prefix = self.config.notifications.priority_prefixes.get(priority, "") if title: @@ -1014,7 +1028,7 @@ async def _deliver_telegram( if not chat_id: return None - text = self._build_telegram_text(session_id, title, body, priority) + text = self._build_notification_text(session_id, title, body, priority) if notif_type in ("question", "approval") and options: button_labels: list[tuple[str, str]] = [] @@ -1125,6 +1139,133 @@ async def _send_telegram_inline( return str(msg.message_id) + # ------------------------------------------------------------------ # + # Slack delivery # + # ------------------------------------------------------------------ # + + def _get_slack_channel(self): + """Get the connected SlackChannel, or None if unavailable.""" + channel = self.engine.router.get_channel("slack") + if not channel or getattr(channel, "_web", None) is None: + return None + return channel + + def _resolve_slack_channel_id(self) -> str | None: + """Resolve the Slack conversation for notification delivery. + + Falls back to the first entry of ``slack.allow_channels`` that is a + literal Slack id. A glob cannot be posted to, and neither can a + channel *name* or the synthetic ``dm`` marker the guardrails match + on — so a list without a real id resolves to nothing and the operator + must set ``notifications.slack_channel_id``. + """ + from nerve.channels.slack import is_slack_id + + configured = self.config.notifications.slack_channel_id + if configured: + return configured + for entry in self.config.slack.allow_channels: + if entry and is_slack_id(entry): + return entry + logger.warning( + "No notifications.slack_channel_id set and slack.allow_channels " + "holds no literal channel id — Slack notifications cannot be " + "delivered", + ) + return None + + async def _deliver_slack( + self, + notification_id: str, + session_id: str, + notif_type: str, + title: str, + body: str, + priority: str, + options: list[str] | None, + option_labels: dict[str, str] | None = None, + ) -> str | None: + """Send a notification to Slack, with Block Kit buttons for answers.""" + channel = self._get_slack_channel() + if not channel: + logger.warning( + "Slack channel not available for notification %s", notification_id, + ) + return None + + target = self._resolve_slack_channel_id() + if not target: + return None + + from nerve.channels.slack import build_notification_blocks + + text = self._build_notification_text(session_id, title, body, priority) + + button_options: list[tuple[str, str]] = [] + if notif_type in ("question", "approval") and options: + for value in options: + if notif_type == "approval": + label = ( + (option_labels or {}).get(value) + or value.replace("_", " ").title() + ) + emoji = _APPROVAL_EMOJIS.get(value, "") + rendered = f"{emoji} {label}".strip() if emoji else label + else: + rendered = value + button_options.append((rendered, value)) + + blocks = build_notification_blocks( + text, notification_id, button_options or None, + ) + msg_id = await channel._post(target, text, blocks) + + if msg_id: + channel._cache_message(msg_id, target, text) + await self.db.update_notification( + notification_id, slack_channel_id=target, + ) + return msg_id + + async def _edit_slack_expired(self, notif: dict[str, Any]) -> None: + """Best-effort edit of the Slack card to show it expired. + + Rebuilds the original text from the row and appends the status line, + dropping the now-dead buttons. All failures are swallowed by design. + """ + message_id = notif.get("slack_message_id") + if not message_id: + return + channel = self._get_slack_channel() + if not channel: + return + target = notif.get("slack_channel_id") or self._resolve_slack_channel_id() + if not target: + return + + from nerve.channels.slack import _md_to_slack, parse_target + + text = self._build_notification_text( + notif["session_id"], + notif.get("title") or "", + notif.get("body") or "", + notif.get("priority") or "normal", + ) + text += "\n\n⏰ Expired unanswered" + + channel_id, _ = parse_target(target) + try: + await channel._web.chat_update( + channel=channel_id, + ts=str(message_id), + text=_md_to_slack(text), + blocks=[], + ) + except Exception as exc: + logger.debug( + "slack expiry edit failed for %s: %s", notif["id"], exc, + ) + # ------------------------------------------------------------------ # # Maintenance (called by the periodic background tick) # # ------------------------------------------------------------------ # @@ -1270,8 +1411,9 @@ async def _report_expired(self, rows: list[dict[str, Any]]) -> None: "expiry broadcast failed for %s: %s", notif["id"], exc, ) - # Telegram: mark the card expired, drop dead buttons. + # Chat channels: mark the card expired, drop dead buttons. await self._edit_telegram_expired(notif) + await self._edit_slack_expired(notif) # Approvals: the proposer is the mechanical pipeline, not a # conversation — record the expiry in its audit log. @@ -1378,7 +1520,7 @@ async def _edit_telegram_expired(self, notif: dict[str, Any]) -> None: if not chat_id: return - text = self._build_telegram_text( + text = self._build_notification_text( notif["session_id"], notif.get("title") or "", notif.get("body") or "", diff --git a/tests/test_db_migrations.py b/tests/test_db_migrations.py new file mode 100644 index 000000000..62789529d --- /dev/null +++ b/tests/test_db_migrations.py @@ -0,0 +1,110 @@ +"""The upgrade path an existing installation actually takes. + +A fresh database applies every migration file, whatever version each one +claims, so a duplicated version number is invisible there. The runner skips +any migration at or below the version already recorded, which is why only an +upgrade from an older database shows a collision. +""" + +from __future__ import annotations + +import importlib + +import aiosqlite +import pytest + +from nerve.db import Database +from nerve.db.migrations.runner import discover_migrations + +# The migration under test, found by suffix so that renumbering the file +# still selects it — and still fails these tests if the number is too low. +_SLACK_MIGRATION_SUFFIX = "_slack_notifications" + + +def _slack_migration() -> tuple[int, str]: + found = [ + (v, name) for v, name in discover_migrations() + if name.endswith(_SLACK_MIGRATION_SUFFIX) + ] + assert len(found) == 1, f"expected one Slack migration, found {found}" + return found[0] + + +async def _build_database_without(path, skipped: str) -> int: + """Create the schema an install had before *skipped* was written. + + Applies every other migration in order and stamps the version at the + highest of them — the state a running installation upgrades from. + Returns that version. + """ + others = [(v, name) for v, name in discover_migrations() if name != skipped] + assert others, "no migrations discovered" + stamp = max(v for v, _ in others) + async with aiosqlite.connect(str(path)) as db: + for _version, module_name in others: + module = importlib.import_module(f"nerve.db.migrations.{module_name}") + await module.up(db) + await db.execute( + "INSERT OR REPLACE INTO schema_version (version) VALUES (?)", + (stamp,), + ) + await db.commit() + return stamp + + +async def _columns(db: Database, table: str) -> set[str]: + async with db.db.execute(f"PRAGMA table_info({table})") as cursor: + return {row[1] for row in await cursor.fetchall()} + + +class TestMigrationVersions: + def test_no_two_migrations_claim_the_same_version(self): + # Two files at the same version both run on a fresh database and the + # second is skipped on every upgrade, so the collision only shows up + # on installs that already exist. + versions = [v for v, _ in discover_migrations()] + assert len(versions) == len(set(versions)) + + def test_the_slack_migration_is_above_every_earlier_one(self): + version, name = _slack_migration() + earlier = [v for v, n in discover_migrations() if n != name] + assert version > max(earlier) + + +@pytest.mark.asyncio +class TestSlackNotificationUpgrade: + async def test_an_existing_database_gains_the_slack_delivery_columns( + self, tmp_path, + ): + path = tmp_path / "upgrade.db" + await _build_database_without(path, _slack_migration()[1]) + + db = Database(path) + await db.connect() + try: + columns = await _columns(db, "notifications") + finally: + await db.close() + assert {"slack_message_id", "slack_channel_id"} <= columns + + async def test_delivery_ids_can_be_written_after_the_upgrade(self, tmp_path): + # Without the columns the Slack post still succeeds and only the + # follow-up write fails, so a card the workspace can see is recorded + # as undelivered and expiry edits lose their target. + path = tmp_path / "upgrade.db" + await _build_database_without(path, _slack_migration()[1]) + + db = Database(path) + await db.connect() + try: + await db.create_notification("n1", "s1", "question", "Ship it?") + await db.update_notification( + "n1", slack_message_id="1699887766.123456", slack_channel_id="C1", + ) + row = await db.get_notification("n1") + finally: + await db.close() + + assert row is not None + assert row["slack_message_id"] == "1699887766.123456" + assert row["slack_channel_id"] == "C1" diff --git a/tests/test_notification_lifecycle.py b/tests/test_notification_lifecycle.py index 38d3800dc..4e48e073d 100644 --- a/tests/test_notification_lifecycle.py +++ b/tests/test_notification_lifecycle.py @@ -613,3 +613,123 @@ async def test_telegram_edit_failure_is_swallowed( assert notif["status"] == "expired" # HTML attempt + plain-text fallback, both swallowed. assert bot.edit_message_text.await_count == 2 + + +# ---------------------------------------------------------------------- +# Answer attribution +# ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestAnswerAttribution: + """A shared workspace needs to know which member approved an action. + + ``answered_by`` names the transport and the injection path routes on + it, so the person travels beside it in the row's metadata rather than + inside the same string. + """ + + async def test_a_question_answer_records_the_actor( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + result = await svc.ask_question(session_id="s1", title="pick one") + + assert await svc.handle_answer( + result["notification_id"], "yes", "slack", actor="U0123ABC", + ) + notif = await db.get_notification(result["notification_id"]) + assert notif["answered_by"] == "slack" + assert json.loads(notif["metadata"])["answered_by_actor"] == "U0123ABC" + + async def test_the_transport_still_names_the_channel_on_its_own( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + # engine.run is handed answered_by as the channel name, so folding + # the member id into that string would route the reply nowhere. + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + result = await svc.ask_question(session_id="s1", title="pick one") + + await svc.handle_answer( + result["notification_id"], "yes", "slack", actor="U0123ABC", + ) + await asyncio.sleep(0) + kwargs = fake_engine.run.call_args.kwargs + assert kwargs["channel"] == "slack" + assert kwargs["source"] == "notification:slack" + + async def test_an_answer_with_no_actor_leaves_the_metadata_alone( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + result = await svc.ask_question(session_id="s1", title="pick one") + + await svc.handle_answer(result["notification_id"], "yes", "web") + notif = await db.get_notification(result["notification_id"]) + assert "answered_by_actor" not in json.loads(notif["metadata"]) + + async def test_existing_metadata_survives_the_answer( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + # The column also carries option_labels, which the Telegram and web + # renderers read after the row is answered. + await db.create_session("s1") + await db.create_notification( + notification_id="n1", session_id="s1", type="question", + title="t", metadata={"option_labels": {"yes": "Ship it"}}, + ) + svc = NotificationService(fake_config, db, fake_engine) + + await svc.handle_answer("n1", "yes", "slack", actor="U0123ABC") + metadata = json.loads((await db.get_notification("n1"))["metadata"]) + assert metadata["option_labels"] == {"yes": "Ship it"} + assert metadata["answered_by_actor"] == "U0123ABC" + + async def test_the_actor_reaches_the_web_broadcast( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + await db.create_session("s1") + svc = NotificationService(fake_config, db, fake_engine) + result = await svc.ask_question(session_id="s1", title="pick one") + + await svc.handle_answer( + result["notification_id"], "yes", "slack", actor="U0123ABC", + ) + answered = [ + m for _, m in patch_broadcaster + if m.get("type") == "notification_answered" + ] + assert answered + assert answered[-1]["answered_by"] == "slack" + assert answered[-1]["answered_by_actor"] == "U0123ABC" + + async def test_an_approval_audit_record_names_the_actor( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + audit_workspace, + ): + await db.create_session("s1") + _handlers.register("attribution-test", lambda *a: _handlers.DispatchResult( + ok=True, + audit_event={ + "event": "approval-acted", + "target_kind": "attribution-test", + "decision": "approve", + "ok": True, + }, + )) + svc = NotificationService(fake_config, db, fake_engine) + nid = await _make_approval( + svc, db, target_kind="attribution-test", + ) + + await svc.handle_answer(nid, "approve", "slack", actor="U0123ABC") + records = read_audit_jsonl( + audit_workspace / ".nerve" / "mechanical-actions", + ) + acted = [r for r in records if r.get("event") == "approval-acted"] + assert acted + assert acted[-1]["answered_by"] == "slack" + assert acted[-1]["answered_by_actor"] == "U0123ABC" diff --git a/tests/test_slack_channel.py b/tests/test_slack_channel.py index 993832895..1ef25068d 100644 --- a/tests/test_slack_channel.py +++ b/tests/test_slack_channel.py @@ -2215,6 +2215,13 @@ async def _press(self, action_id="notif:n1:approve", value="approve"): ) return channel, service + @pytest.mark.asyncio + async def test_the_button_press_carries_the_slack_member_id(self): + _, service = await self._press() + kwargs = service.handle_answer.await_args.kwargs + assert kwargs["answered_by"] == "slack" + assert kwargs["actor"] == "U0123ABC" + @pytest.mark.asyncio async def test_the_settled_card_names_who_answered(self): channel, service = await self._press() From 83baa14e42536de00f32ee21d35e45afe23ec7f8 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Sun, 23 Aug 2026 19:48:14 +0200 Subject: [PATCH 2/8] Document Slack notification delivery --- README.md | 2 +- docs/architecture.md | 4 ++-- docs/config.md | 13 +++++++++++++ 3 files changed, 16 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index c44ef2133..8b9287fa9 100644 --- a/README.md +++ b/README.md @@ -218,7 +218,7 @@ Revisions happen in the same persistent planner session — full context preserv ### 🔔 Notifications -Async communication between agent and human, delivered to both web UI and Telegram. +Async communication between agent and human, delivered to the web UI, Telegram, and Slack. - **`notify`** — Fire-and-forget alerts (status updates, completions, reminders) - **`ask_user`** — Questions with predefined options, rendered as buttons diff --git a/docs/architecture.md b/docs/architecture.md index e19c30d24..f7e036714 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -93,8 +93,8 @@ Async notification system for agent→user communication: - **`notify` tool** — fire-and-forget notifications (status updates, alerts, reminders) - **`ask_user` tool** — questions with predefined options (rendered as buttons) + free-text input. Supports blocking mode (`wait=true`) and async mode (answer injected as session message) - **NotificationService** — centralized fanout to configurable channels (web + Telegram by default), answer routing, periodic expiry -- **Multi-channel delivery** — web UI via `__global__` WebSocket broadcast channel, Telegram via direct bot API with inline keyboard buttons for questions -- **Answer routing** — answers from any channel (web UI, Telegram inline button, `/reply` command) are persisted and either unblock a waiting tool or injected as a user message into the originating session +- **Multi-channel delivery** — web UI via `__global__` WebSocket broadcast channel, Telegram via direct bot API with inline keyboard buttons, Slack via Block Kit action buttons +- **Answer routing** — answers from any channel (web UI, Telegram inline button, Slack button, `/reply` command) are persisted and either unblock a waiting tool or injected as a user message into the originating session - **Web UI** — `/notifications` page with status/type filters, inline answer buttons, dismiss, dismiss-all; real-time toast overlay for new notifications; NavRail badge for pending count ### Cron Service (`nerve/cron/`) diff --git a/docs/config.md b/docs/config.md index 8e1dde448..805b9e29c 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1259,6 +1259,19 @@ Mode delivers an event to only one connection per app, so each instance needs its **own app** — a second instance sharing one app's tokens would take events away from the first rather than run beside it. +### Notifications + +Add `slack` to `notifications.channels` to deliver questions and approvals +as Block Kit cards with buttons. + +| Key | Type | Default | Description | +|-----|------|---------|-------------| +| `notifications.slack_channel_id` | string | `""` | Conversation for notifications; falls back to the first literal channel id in `slack.allow_channels` | + +The fallback only accepts a real channel id. A list of names and globs — or +one holding just `dm` — resolves to nothing, and Slack notifications are +skipped with a warning rather than posted to a made-up target. + ## Quiet Hours | Key | Type | Default | Description | From eee8f3cf329aea54c6c844272f4dee9bd7b8749c Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Mon, 24 Aug 2026 09:11:51 +0200 Subject: [PATCH 3/8] Remove redundant Slack migration ordering test The existing-database upgrade test stamps the schema at the highest non-Slack migration, so it already fails if the Slack migration is not strictly newer. Keep the behavioral upgrade assertion and drop the weaker structural duplicate. --- tests/test_db_migrations.py | 6 ------ 1 file changed, 6 deletions(-) diff --git a/tests/test_db_migrations.py b/tests/test_db_migrations.py index 62789529d..224a676dd 100644 --- a/tests/test_db_migrations.py +++ b/tests/test_db_migrations.py @@ -65,12 +65,6 @@ def test_no_two_migrations_claim_the_same_version(self): versions = [v for v, _ in discover_migrations()] assert len(versions) == len(set(versions)) - def test_the_slack_migration_is_above_every_earlier_one(self): - version, name = _slack_migration() - earlier = [v for v, n in discover_migrations() if n != name] - assert version > max(earlier) - - @pytest.mark.asyncio class TestSlackNotificationUpgrade: async def test_an_existing_database_gains_the_slack_delivery_columns( From 8687fa900896d51923ac46d1c473c8b6044fe664 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Mon, 24 Aug 2026 13:43:09 +0200 Subject: [PATCH 4/8] Align Slack notifications with explicit DMs --- docs/config.md | 6 +++--- nerve/notifications/service.py | 7 +++---- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/docs/config.md b/docs/config.md index 805b9e29c..d14823fdc 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1268,9 +1268,9 @@ as Block Kit cards with buttons. |-----|------|---------|-------------| | `notifications.slack_channel_id` | string | `""` | Conversation for notifications; falls back to the first literal channel id in `slack.allow_channels` | -The fallback only accepts a real channel id. A list of names and globs — or -one holding just `dm` — resolves to nothing, and Slack notifications are -skipped with a warning rather than posted to a made-up target. +The fallback only accepts a real channel id. A list of names and globs resolves +to nothing, and Slack notifications are skipped with a warning rather than +posted to a made-up target. ## Quiet Hours diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index d2e656c1f..4bcfbcb5b 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -1154,10 +1154,9 @@ def _resolve_slack_channel_id(self) -> str | None: """Resolve the Slack conversation for notification delivery. Falls back to the first entry of ``slack.allow_channels`` that is a - literal Slack id. A glob cannot be posted to, and neither can a - channel *name* or the synthetic ``dm`` marker the guardrails match - on — so a list without a real id resolves to nothing and the operator - must set ``notifications.slack_channel_id``. + literal Slack id. A glob or channel *name* cannot be posted to, so a + list without a real id resolves to nothing and the operator must set + ``notifications.slack_channel_id``. """ from nerve.channels.slack import is_slack_id From 32bdac840abf66afb5c9e059828224eb22e15467 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Mon, 24 Aug 2026 14:37:17 +0200 Subject: [PATCH 5/8] Condense Slack notification documentation --- docs/config.md | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/docs/config.md b/docs/config.md index d14823fdc..5e917476e 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1261,16 +1261,14 @@ away from the first rather than run beside it. ### Notifications -Add `slack` to `notifications.channels` to deliver questions and approvals -as Block Kit cards with buttons. +Add `slack` to `notifications.channels` to send question and approval cards. | Key | Type | Default | Description | |-----|------|---------|-------------| -| `notifications.slack_channel_id` | string | `""` | Conversation for notifications; falls back to the first literal channel id in `slack.allow_channels` | +| `notifications.slack_channel_id` | string | `""` | Target channel ID; defaults to the first literal ID in `slack.allow_channels` | -The fallback only accepts a real channel id. A list of names and globs resolves -to nothing, and Slack notifications are skipped with a warning rather than -posted to a made-up target. +Names and globs are not resolved for this fallback. Without a literal channel +ID, delivery is skipped with a warning. ## Quiet Hours From a203721d7c75bfba7f38ffc75b78a688c970f770 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 12:55:25 +0200 Subject: [PATCH 6/8] Keep Slack notification delivery behind channel boundaries --- nerve/channels/slack.py | 62 +++++++++++ .../db/migrations/v045_slack_notifications.py | 26 ----- nerve/db/notifications.py | 16 +++ nerve/notifications/service.py | 102 ++++++----------- tests/test_db_migrations.py | 104 ------------------ tests/test_notification_lifecycle.py | 84 +++++++++++++- tests/test_slack_channel.py | 50 ++++++++- 7 files changed, 241 insertions(+), 203 deletions(-) delete mode 100644 nerve/db/migrations/v045_slack_notifications.py delete mode 100644 tests/test_db_migrations.py diff --git a/nerve/channels/slack.py b/nerve/channels/slack.py index f2bed7a45..149fa95f5 100644 --- a/nerve/channels/slack.py +++ b/nerve/channels/slack.py @@ -36,6 +36,7 @@ _MAX_ACTION_ELEMENTS, _SESSIONS_BUTTON_LIMIT, _md_to_slack, + build_notification_blocks, build_sessions_blocks, slack_emoji_name, slack_to_plain, @@ -1225,6 +1226,67 @@ async def _post( ) return resp.get("ts") + def _notification_target(self) -> str | None: + """Resolve a concrete conversation from the active config generation.""" + configured = self.config.notifications.slack_channel_id.strip() + if configured: + if is_slack_id(configured) and configured[0] in "CGD": + return configured + logger.warning( + "notifications.slack_channel_id is not a Slack conversation id", + ) + return None + + for entry in self.config.slack.allow_channels: + if is_slack_id(entry) and entry[0] in "CG": + return entry + logger.warning( + "No notifications.slack_channel_id is set and slack.allow_channels " + "has no literal conversation id", + ) + return None + + async def post_notification( + self, + notification_id: str, + text: str, + options: list[tuple[str, str]] | None = None, + ) -> tuple[str, str] | None: + """Render and post one notification using the active Slack config.""" + if not self.is_available: + return None + target = self._notification_target() + if not target: + return None + blocks = build_notification_blocks(text, notification_id, options) + message_id = await self._post(target, text, blocks) + if not message_id: + return None + self._cache_message(message_id, target, text) + return target, message_id + + async def expire_notification( + self, + target: str, + message_id: str, + text: str, + ) -> None: + """Replace a notification card with its expired state.""" + if not self.is_available: + return + channel_id, _ = parse_target(target) + try: + await self._web.chat_update( + channel=channel_id, + ts=message_id, + text=_md_to_slack(text), + blocks=[], + ) + except Exception as exc: + logger.debug( + "Slack expiry edit failed for %s: %s", message_id, exc, + ) + async def send(self, message: OutboundMessage) -> None: """Send a complete message, split to fit Slack's render limit. diff --git a/nerve/db/migrations/v045_slack_notifications.py b/nerve/db/migrations/v045_slack_notifications.py deleted file mode 100644 index 070685f43..000000000 --- a/nerve/db/migrations/v045_slack_notifications.py +++ /dev/null @@ -1,26 +0,0 @@ -"""V45: Slack delivery ids on notifications.""" - -from __future__ import annotations - -import logging - -import aiosqlite - -logger = logging.getLogger(__name__) - -COLUMNS = ( - ("slack_message_id", "TEXT"), - ("slack_channel_id", "TEXT"), -) - - -async def up(db: aiosqlite.Connection) -> None: - cursor = await db.execute("PRAGMA table_info(notifications)") - existing = {row[1] for row in await cursor.fetchall()} - for name, decl in COLUMNS: - if name in existing: - continue - await db.execute( - f"ALTER TABLE notifications ADD COLUMN {name} {decl}", - ) - logger.info("V45 migration: notifications carries Slack delivery ids") diff --git a/nerve/db/notifications.py b/nerve/db/notifications.py index 2f01cbed3..068640636 100644 --- a/nerve/db/notifications.py +++ b/nerve/db/notifications.py @@ -176,6 +176,22 @@ async def get_notification_delivery( row = await cursor.fetchone() return dict(row) if row else None + async def get_latest_notification_delivery( + self, + notification_id: str, + channel: str, + ) -> dict | None: + """Return the most recent delivery through one transport.""" + async with self.db.execute( + """SELECT * FROM notification_deliveries + WHERE notification_id = ? AND channel = ? + ORDER BY delivered_at DESC, rowid DESC + LIMIT 1""", + (notification_id, channel), + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + async def find_pending_question_for_delivery( self, channel: str, diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index 4bcfbcb5b..21a2dfbcc 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -858,10 +858,6 @@ async def _deliver(channel_name: str) -> str | None: ) if not msg_id: return None - await self.db.update_notification( - notification_id, - slack_message_id=str(msg_id), - ) return "slack" except Exception as e: logger.error( @@ -1146,33 +1142,10 @@ async def _send_telegram_inline( def _get_slack_channel(self): """Get the connected SlackChannel, or None if unavailable.""" channel = self.engine.router.get_channel("slack") - if not channel or getattr(channel, "_web", None) is None: + if not channel or not getattr(channel, "is_available", False): return None return channel - def _resolve_slack_channel_id(self) -> str | None: - """Resolve the Slack conversation for notification delivery. - - Falls back to the first entry of ``slack.allow_channels`` that is a - literal Slack id. A glob or channel *name* cannot be posted to, so a - list without a real id resolves to nothing and the operator must set - ``notifications.slack_channel_id``. - """ - from nerve.channels.slack import is_slack_id - - configured = self.config.notifications.slack_channel_id - if configured: - return configured - for entry in self.config.slack.allow_channels: - if entry and is_slack_id(entry): - return entry - logger.warning( - "No notifications.slack_channel_id set and slack.allow_channels " - "holds no literal channel id — Slack notifications cannot be " - "delivered", - ) - return None - async def _deliver_slack( self, notification_id: str, @@ -1192,12 +1165,6 @@ async def _deliver_slack( ) return None - target = self._resolve_slack_channel_id() - if not target: - return None - - from nerve.channels.slack import build_notification_blocks - text = self._build_notification_text(session_id, title, body, priority) button_options: list[tuple[str, str]] = [] @@ -1214,17 +1181,21 @@ async def _deliver_slack( rendered = value button_options.append((rendered, value)) - blocks = build_notification_blocks( - text, notification_id, button_options or None, + delivery = await channel.post_notification( + notification_id, + text, + button_options or None, ) - msg_id = await channel._post(target, text, blocks) - - if msg_id: - channel._cache_message(msg_id, target, text) - await self.db.update_notification( - notification_id, slack_channel_id=target, - ) - return msg_id + if not delivery: + return None + target, message_id = delivery + await self.db.record_notification_delivery( + notification_id, + "slack", + target=target, + message_id=message_id, + ) + return message_id async def _edit_slack_expired(self, notif: dict[str, Any]) -> None: """Best-effort edit of the Slack card to show it expired. @@ -1232,18 +1203,15 @@ async def _edit_slack_expired(self, notif: dict[str, Any]) -> None: Rebuilds the original text from the row and appends the status line, dropping the now-dead buttons. All failures are swallowed by design. """ - message_id = notif.get("slack_message_id") - if not message_id: - return channel = self._get_slack_channel() if not channel: return - target = notif.get("slack_channel_id") or self._resolve_slack_channel_id() - if not target: + delivery = await self.db.get_latest_notification_delivery( + notif["id"], "slack", + ) + if not delivery or not delivery.get("message_id"): return - from nerve.channels.slack import _md_to_slack, parse_target - text = self._build_notification_text( notif["session_id"], notif.get("title") or "", @@ -1251,19 +1219,11 @@ async def _edit_slack_expired(self, notif: dict[str, Any]) -> None: notif.get("priority") or "normal", ) text += "\n\n⏰ Expired unanswered" - - channel_id, _ = parse_target(target) - try: - await channel._web.chat_update( - channel=channel_id, - ts=str(message_id), - text=_md_to_slack(text), - blocks=[], - ) - except Exception as exc: - logger.debug( - "slack expiry edit failed for %s: %s", notif["id"], exc, - ) + await channel.expire_notification( + delivery["target"], + str(delivery["message_id"]), + text, + ) # ------------------------------------------------------------------ # # Maintenance (called by the periodic background tick) # @@ -1500,12 +1460,14 @@ async def _edit_telegram_expired(self, notif: dict[str, Any]) -> None: now-dead inline keyboard. Telegram refuses edits on old messages (>48h) — all failures are swallowed by design. """ - target = str(notif.get("telegram_chat_id") or "") - delivery = None - if target: - delivery = await self.db.get_notification_delivery( - notif["id"], "telegram", target, - ) + delivery = await self.db.get_latest_notification_delivery( + notif["id"], "telegram", + ) + target = str( + (delivery or {}).get("target") + or notif.get("telegram_chat_id") + or "" + ) message_id = ( (delivery or {}).get("message_id") or notif.get("telegram_message_id") diff --git a/tests/test_db_migrations.py b/tests/test_db_migrations.py deleted file mode 100644 index 224a676dd..000000000 --- a/tests/test_db_migrations.py +++ /dev/null @@ -1,104 +0,0 @@ -"""The upgrade path an existing installation actually takes. - -A fresh database applies every migration file, whatever version each one -claims, so a duplicated version number is invisible there. The runner skips -any migration at or below the version already recorded, which is why only an -upgrade from an older database shows a collision. -""" - -from __future__ import annotations - -import importlib - -import aiosqlite -import pytest - -from nerve.db import Database -from nerve.db.migrations.runner import discover_migrations - -# The migration under test, found by suffix so that renumbering the file -# still selects it — and still fails these tests if the number is too low. -_SLACK_MIGRATION_SUFFIX = "_slack_notifications" - - -def _slack_migration() -> tuple[int, str]: - found = [ - (v, name) for v, name in discover_migrations() - if name.endswith(_SLACK_MIGRATION_SUFFIX) - ] - assert len(found) == 1, f"expected one Slack migration, found {found}" - return found[0] - - -async def _build_database_without(path, skipped: str) -> int: - """Create the schema an install had before *skipped* was written. - - Applies every other migration in order and stamps the version at the - highest of them — the state a running installation upgrades from. - Returns that version. - """ - others = [(v, name) for v, name in discover_migrations() if name != skipped] - assert others, "no migrations discovered" - stamp = max(v for v, _ in others) - async with aiosqlite.connect(str(path)) as db: - for _version, module_name in others: - module = importlib.import_module(f"nerve.db.migrations.{module_name}") - await module.up(db) - await db.execute( - "INSERT OR REPLACE INTO schema_version (version) VALUES (?)", - (stamp,), - ) - await db.commit() - return stamp - - -async def _columns(db: Database, table: str) -> set[str]: - async with db.db.execute(f"PRAGMA table_info({table})") as cursor: - return {row[1] for row in await cursor.fetchall()} - - -class TestMigrationVersions: - def test_no_two_migrations_claim_the_same_version(self): - # Two files at the same version both run on a fresh database and the - # second is skipped on every upgrade, so the collision only shows up - # on installs that already exist. - versions = [v for v, _ in discover_migrations()] - assert len(versions) == len(set(versions)) - -@pytest.mark.asyncio -class TestSlackNotificationUpgrade: - async def test_an_existing_database_gains_the_slack_delivery_columns( - self, tmp_path, - ): - path = tmp_path / "upgrade.db" - await _build_database_without(path, _slack_migration()[1]) - - db = Database(path) - await db.connect() - try: - columns = await _columns(db, "notifications") - finally: - await db.close() - assert {"slack_message_id", "slack_channel_id"} <= columns - - async def test_delivery_ids_can_be_written_after_the_upgrade(self, tmp_path): - # Without the columns the Slack post still succeeds and only the - # follow-up write fails, so a card the workspace can see is recorded - # as undelivered and expiry edits lose their target. - path = tmp_path / "upgrade.db" - await _build_database_without(path, _slack_migration()[1]) - - db = Database(path) - await db.connect() - try: - await db.create_notification("n1", "s1", "question", "Ship it?") - await db.update_notification( - "n1", slack_message_id="1699887766.123456", slack_channel_id="C1", - ) - row = await db.get_notification("n1") - finally: - await db.close() - - assert row is not None - assert row["slack_message_id"] == "1699887766.123456" - assert row["slack_channel_id"] == "C1" diff --git a/tests/test_notification_lifecycle.py b/tests/test_notification_lifecycle.py index 4e48e073d..cb5712fa4 100644 --- a/tests/test_notification_lifecycle.py +++ b/tests/test_notification_lifecycle.py @@ -23,7 +23,6 @@ import json from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any from unittest.mock import AsyncMock, MagicMock import pytest @@ -176,6 +175,24 @@ async def test_v045_delivery_scope_exists(self, db: Database): @pytest.mark.asyncio class TestScopedAnswers: + async def test_latest_delivery_can_move_between_targets( + self, db: Database, + ): + await db.create_session("s1", source="external") + await db.create_notification("n1", "s1", "question", "Question") + await db.record_notification_delivery( + "n1", "slack", target="C0123ABC", message_id="1.0", + ) + await db.record_notification_delivery( + "n1", "slack", target="C0456DEF", message_id="2.0", + ) + + delivery = await db.get_latest_notification_delivery("n1", "slack") + + assert delivery + assert delivery["target"] == "C0456DEF" + assert delivery["message_id"] == "2.0" + async def test_latest_question_is_scoped_to_delivery_target( self, db: Database, fake_config, fake_engine, patch_broadcaster, ): @@ -220,6 +237,71 @@ async def test_explicit_answer_rejects_a_different_delivery_target( assert (await db.get_notification("n1"))["status"] == "pending" +@pytest.mark.asyncio +class TestSlackDeliveryBoundary: + async def test_service_records_the_reference_returned_by_the_channel( + self, db: Database, fake_config, fake_engine, + ): + await db.create_session("s1", source="external") + await db.create_notification("n1", "s1", "question", "Question") + channel = MagicMock(is_available=True) + channel.post_notification = AsyncMock( + return_value=("C0456DEF", "1.0"), + ) + fake_engine.router.get_channel.return_value = channel + service = NotificationService(fake_config, db, fake_engine) + + message_id = await service._deliver_slack( + "n1", "s1", "question", "Question", "Body", "normal", ["yes"], + ) + + assert message_id == "1.0" + delivery = await db.get_notification_delivery( + "n1", "slack", "C0456DEF", + ) + assert delivery and delivery["message_id"] == "1.0" + options = channel.post_notification.await_args.args[2] + assert options == [("yes", "yes")] + + async def test_quiescing_channel_is_not_used( + self, db: Database, fake_config, fake_engine, + ): + channel = MagicMock(is_available=False) + channel.post_notification = AsyncMock() + fake_engine.router.get_channel.return_value = channel + service = NotificationService(fake_config, db, fake_engine) + + message_id = await service._deliver_slack( + "n1", "s1", "notify", "Notice", "Body", "normal", None, + ) + + assert message_id is None + channel.post_notification.assert_not_awaited() + + async def test_expiry_uses_the_latest_recorded_target( + self, db: Database, fake_config, fake_engine, + ): + await db.create_session("s1", source="external") + await db.create_notification("n1", "s1", "question", "Question") + await db.record_notification_delivery( + "n1", "slack", target="C0123ABC", message_id="1.0", + ) + await db.record_notification_delivery( + "n1", "slack", target="C0456DEF", message_id="2.0", + ) + channel = MagicMock(is_available=True) + channel.expire_notification = AsyncMock() + fake_engine.router.get_channel.return_value = channel + service = NotificationService(fake_config, db, fake_engine) + + await service._edit_slack_expired(await db.get_notification("n1")) + + channel.expire_notification.assert_awaited_once() + args = channel.expire_notification.await_args.args + assert args[:2] == ("C0456DEF", "2.0") + assert args[2].endswith("⏰ Expired unanswered") + + # ---------------------------------------------------------------------- # Snooze semantics # ---------------------------------------------------------------------- diff --git a/tests/test_slack_channel.py b/tests/test_slack_channel.py index 1ef25068d..237d2c259 100644 --- a/tests/test_slack_channel.py +++ b/tests/test_slack_channel.py @@ -266,6 +266,53 @@ def test_section_text_stays_inside_slacks_limit(self): assert len(blocks[0]["text"]["text"]) <= 3000 +class TestNotificationDelivery: + @pytest.mark.asyncio + async def test_the_channel_owns_target_resolution_and_block_rendering(self): + channel = _channel(allow_channels=["engineering-*", "C0456DEF"]) + + delivery = await channel.post_notification( + "n1", "Deploy?", [("Approve", "approve")], + ) + + assert delivery == ("C0456DEF", "1.1") + posted = channel._web.chat_postMessage.await_args.kwargs + assert posted["channel"] == "C0456DEF" + assert posted["blocks"][1]["elements"][0]["action_id"] == ( + "notif:n1:approve" + ) + + @pytest.mark.asyncio + async def test_an_explicit_dm_is_a_notification_target(self): + channel = _channel() + channel.config.notifications.slack_channel_id = "D0123ABC" + + delivery = await channel.post_notification("n1", "Hello") + + assert delivery == ("D0123ABC", "1.1") + + @pytest.mark.asyncio + async def test_a_quiescing_channel_refuses_external_delivery(self): + channel = _channel(allow_channels=["C0456DEF"]) + channel._state = "quiescing" + + assert await channel.post_notification("n1", "Hello") is None + channel._web.chat_postMessage.assert_not_awaited() + + @pytest.mark.asyncio + async def test_expiry_replaces_the_card_without_buttons(self): + channel = _channel() + + await channel.expire_notification("C0456DEF", "1.1", "Expired") + + channel._web.chat_update.assert_awaited_once_with( + channel="C0456DEF", + ts="1.1", + text="Expired", + blocks=[], + ) + + # ---------------------------------------------------------------------- # # Channel wiring # # ---------------------------------------------------------------------- # @@ -2218,8 +2265,7 @@ async def _press(self, action_id="notif:n1:approve", value="approve"): @pytest.mark.asyncio async def test_the_button_press_carries_the_slack_member_id(self): _, service = await self._press() - kwargs = service.handle_answer.await_args.kwargs - assert kwargs["answered_by"] == "slack" + kwargs = service.answer_delivered_notification.await_args.kwargs assert kwargs["actor"] == "U0123ABC" @pytest.mark.asyncio From 627761b0975616ed663d57a0a3e4f9164d3fa7d4 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 25 Aug 2026 15:33:33 +0200 Subject: [PATCH 7/8] Deliver Slack notifications by default and keep their cards answerable Three faults that between them meant a card either never appeared or stopped taking answers. notifications.channels defaulted to web and telegram, so nothing reached Slack until an operator found a key documented in one sentence with no default and no example. Add slack to the default, document the key, and put a notifications block in the example. README said delivery already covered Slack, which only held once that key was set by hand. A press carried the message's thread_ts into the delivery target. Slack fills thread_ts in on any message that has replies, so one reply under a card made every later press miss the record that post_notification wrote against the bare conversation, and each one answered "already answered or expired" while the row stayed pending until it expired. Look the record up by the conversation, which is where the card is posted. A reaction on a card routed to slack:. Shared channels have no conversation-wide session and the pickers do not list one, so an emoji opened a session that /nerve stop could never reach. Require a thread outside DMs, where one conversation is the session. Slack now being on by default, an absent channel is the ordinary case and logs at debug; a registered one that cannot take traffic still warns. A channel name nothing delivers to no longer passes in silence. --- README.md | 4 ++- config.example.yaml | 9 ++++++ docs/config.md | 11 +++++-- nerve/channels/slack.py | 22 ++++++++++--- nerve/config.py | 6 ++-- nerve/notifications/service.py | 23 +++++++++++-- tests/test_notification_lifecycle.py | 13 ++++++++ tests/test_slack_channel.py | 48 ++++++++++++++++++++++++++-- 8 files changed, 120 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index 8b9287fa9..9be562614 100644 --- a/README.md +++ b/README.md @@ -218,7 +218,9 @@ Revisions happen in the same persistent planner session — full context preserv ### 🔔 Notifications -Async communication between agent and human, delivered to the web UI, Telegram, and Slack. +Async communication between agent and human, delivered to the web UI, Telegram, +and Slack. `notifications.channels` chooses which of them; all three are on by +default and a transport that is off is skipped. - **`notify`** — Fire-and-forget alerts (status updates, completions, reminders) - **`ask_user`** — Questions with predefined options, rendered as buttons diff --git a/config.example.yaml b/config.example.yaml index 7170526d9..e1861dc95 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -144,6 +144,15 @@ slack: # doctor/restart affect the host; sessions lists other channels. Opt in. # commands: [sessions, new, stop, reply] +# Where notify, ask_user, and propose_action deliver. The list replaces the +# default rather than adding to it, so name every transport you want. A +# transport that is off costs nothing here. +notifications: + channels: [web, telegram, slack] + # Target conversation for Slack cards. Without this, the first literal ID + # in slack.allow_channels is used; names and globs are not resolved. + # slack_channel_id: "C0456DEF" + # Quiet hours (local timezone) quiet_start: "02:00" quiet_end: "12:00" diff --git a/docs/config.md b/docs/config.md index 5e917476e..7172221b8 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1261,14 +1261,19 @@ away from the first rather than run beside it. ### Notifications -Add `slack` to `notifications.channels` to send question and approval cards. +Question and approval cards go to Slack by default. | Key | Type | Default | Description | |-----|------|---------|-------------| +| `notifications.channels` | list | `[web, telegram, slack]` | Where `notify`, `ask_user`, and `propose_action` deliver | | `notifications.slack_channel_id` | string | `""` | Target channel ID; defaults to the first literal ID in `slack.allow_channels` | -Names and globs are not resolved for this fallback. Without a literal channel -ID, delivery is skipped with a warning. +`notifications.channels` replaces the default rather than adding to it, so +list every transport you want. A name nothing delivers to is skipped with a +warning. Slack in the list costs nothing while Slack is off. + +Names and globs are not resolved for the `slack_channel_id` fallback. Without +a literal channel ID, delivery is skipped with a warning. ## Quiet Hours diff --git a/nerve/channels/slack.py b/nerve/channels/slack.py index 149fa95f5..d66029f37 100644 --- a/nerve/channels/slack.py +++ b/nerve/channels/slack.py @@ -1049,6 +1049,16 @@ async def _handle_reaction_event(self, event: dict[str, Any]) -> None: return target, original_text = cached + _, thread_ts = parse_target(target) + if not channel_id.startswith("D") and thread_ts is None: + # A shared channel has no conversation-wide session: each thread + # owns one. A message cached at channel level, such as a + # notification card, has no thread for a reaction to join, and + # opening one would write a slack: mapping that the + # session pickers deliberately do not list. A DM is one + # conversation, so it has no thread to require. + return + channel_type = "im" if channel_id.startswith("D") else "channel" if not await self._authorize(user_id, channel_id, channel_type): return @@ -2059,15 +2069,17 @@ async def _handle_notification_button( return actor = (payload.get("user") or {}).get("id") or "" - thread_ts = (payload.get("message") or {}).get("thread_ts") or None + # A card is posted at conversation level, so the target recorded for + # it is the bare conversation. Slack fills in thread_ts on any + # message that has replies, so carrying it across from the press + # would stop matching that record the moment somebody replied under + # the card, and every later press would read as already answered. + target = format_target((payload.get("channel") or {}).get("id") or "") result = await self._notification_service.answer_delivered_notification( notification_id, answer, channel="slack", - target=format_target( - (payload.get("channel") or {}).get("id") or "", - thread_ts, - ), + target=target, actor=actor, ) if not result: diff --git a/nerve/config.py b/nerve/config.py index da292a21a..216108fde 100644 --- a/nerve/config.py +++ b/nerve/config.py @@ -2052,7 +2052,9 @@ def from_dict(cls, d: dict) -> AuthConfig: @dataclass class NotificationsConfig: """Async notification delivery settings.""" - channels: list[str] = field(default_factory=lambda: ["web", "telegram"]) + channels: list[str] = field( + default_factory=lambda: ["web", "telegram", "slack"], + ) telegram_chat_id: int | None = None # Target chat; falls back to first allowed_user slack_channel_id: str = "" # Target conversation; falls back to a literal id in slack.allow_channels default_expiry_hours: int = 48 # Auto-expire unanswered questions @@ -2071,7 +2073,7 @@ class NotificationsConfig: @_coerced def from_dict(cls, d: dict) -> NotificationsConfig: return cls( - channels=d.get("channels", ["web", "telegram"]), + channels=d.get("channels", ["web", "telegram", "slack"]), telegram_chat_id=d.get("telegram_chat_id"), slack_channel_id=str(d.get("slack_channel_id") or ""), default_expiry_hours=d.get("default_expiry_hours", 48), diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index 21a2dfbcc..ab278f808 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -859,6 +859,12 @@ async def _deliver(channel_name: str) -> str | None: if not msg_id: return None return "slack" + else: + logger.warning( + "notifications.channels names %r, which nothing " + "delivers to; notification %s skips it", + channel_name, notification_id, + ) except Exception as e: logger.error( "Failed to deliver %s to %s: %s", @@ -1160,9 +1166,20 @@ async def _deliver_slack( """Send a notification to Slack, with Block Kit buttons for answers.""" channel = self._get_slack_channel() if not channel: - logger.warning( - "Slack channel not available for notification %s", notification_id, - ) + # Slack is in the default channel list, so most installations + # reach here with it switched off. An absent channel is that + # case and stays quiet; a registered one that cannot take + # traffic is worth a line. + if self.engine.router.get_channel("slack") is None: + logger.debug( + "Slack is not running; notification %s skips it", + notification_id, + ) + else: + logger.warning( + "Slack channel not available for notification %s", + notification_id, + ) return None text = self._build_notification_text(session_id, title, body, priority) diff --git a/tests/test_notification_lifecycle.py b/tests/test_notification_lifecycle.py index cb5712fa4..2eb2d23f4 100644 --- a/tests/test_notification_lifecycle.py +++ b/tests/test_notification_lifecycle.py @@ -172,6 +172,19 @@ async def test_v045_delivery_scope_exists(self, db: Database): cols = {row[1] async for row in cur} assert {"notification_id", "channel", "target", "message_id"} <= cols + async def test_slack_is_a_default_notification_channel(self): + # Cards reach nobody unless the transport is in this list, and the + # list replaces the default rather than extending it. + from nerve.config import NotificationsConfig + + assert NotificationsConfig().channels == ["web", "telegram", "slack"] + assert NotificationsConfig.from_dict({}).channels == [ + "web", "telegram", "slack", + ] + assert NotificationsConfig.from_dict( + {"channels": ["web"]}, + ).channels == ["web"] + @pytest.mark.asyncio class TestScopedAnswers: diff --git a/tests/test_slack_channel.py b/tests/test_slack_channel.py index 237d2c259..b0e35de9c 100644 --- a/tests/test_slack_channel.py +++ b/tests/test_slack_channel.py @@ -762,6 +762,36 @@ async def test_a_reaction_does_not_cross_conversations_on_a_shared_ts(self): }) channel.router.handle_message.assert_not_called() + @pytest.mark.asyncio + async def test_a_reaction_on_a_notification_card_opens_no_session(self): + # A card is posted at channel level, so routing a reaction on it + # would write a slack: mapping. Shared channels have no + # conversation-wide session, and the pickers do not list one, so + # nothing could stop it afterwards. + channel = _channel(allow_channels=["C0123ABCD"]) + channel.config.notifications.slack_channel_id = "C0123ABCD" + posted = await channel.post_notification("n1", "Approve this?", None) + assert posted == ("C0123ABCD", "1.1") + + await channel._handle_reaction_event({ + "type": "reaction_added", "user": "U1", "reaction": "eyes", + "item": {"channel": "C0123ABCD", "ts": "1.1"}, + }) + channel.router.handle_message.assert_not_called() + + @pytest.mark.asyncio + async def test_a_reaction_on_a_direct_message_card_still_answers(self): + # A DM is one conversation, so a card there has a session to join. + channel = _channel(allow_users=["U1"], allow_direct_messages=True) + channel.config.notifications.slack_channel_id = "D0123ABCD" + assert await channel.post_notification("n1", "Approve this?", None) + await channel._handle_reaction_event({ + "type": "reaction_added", "user": "U1", "reaction": "eyes", + "item": {"channel": "D0123ABCD", "ts": "1.1"}, + }) + msg = channel.router.handle_message.await_args[0][0] + assert msg.channel_key == "slack:D0123ABCD" + @pytest.mark.asyncio async def test_a_reaction_still_reaches_its_own_thread_session(self): channel = _channel(allow_channels=["C1"]) @@ -2239,7 +2269,9 @@ async def test_a_split_card_keeps_every_section(self, monkeypatch): class TestApprovalAttribution: """`answered_by="slack"` alone loses which member pressed the button.""" - async def _press(self, action_id="notif:n1:approve", value="approve"): + async def _press( + self, action_id="notif:n1:approve", value="approve", **message, + ): channel = _channel(allow_users=["U0123ABC"]) channel._replace_via_url = AsyncMock() service = MagicMock() @@ -2257,11 +2289,23 @@ async def _press(self, action_id="notif:n1:approve", value="approve"): "channel": {"id": "C1"}, "response_url": "https://hooks.slack.test/x", "actions": [{"action_id": action_id, "value": value}], - "message": {"blocks": build_notification_blocks("Ship it?", "n1")}, + "message": { + "blocks": build_notification_blocks("Ship it?", "n1"), + **message, + }, } ) return channel, service + @pytest.mark.asyncio + async def test_a_thread_reply_under_the_card_keeps_the_buttons_working(self): + # Slack fills thread_ts in on a message once it has replies, so a + # press after one reply was looking up "C1:1.1" while the delivery + # record held "C1", and every press read as already answered. + _, service = await self._press(ts="1.1", thread_ts="1.1") + kwargs = service.answer_delivered_notification.await_args.kwargs + assert kwargs["target"] == "C1" + @pytest.mark.asyncio async def test_the_button_press_carries_the_slack_member_id(self): _, service = await self._press() From 5447a7ff41deb78dc9125a08d933bf85a7716c07 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Wed, 16 Sep 2026 15:14:39 +0200 Subject: [PATCH 8/8] Claim an approval before its dispatcher runs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Two presses ran the action twice.** `handle_answer` reads the row's status, and `_handle_approval_answer` runs the dispatcher before anything settles the row, so that read cannot decide who acts. Every Socket Mode envelope is dispatched as its own task, so two members pressing inside the same second both passed the read, both dispatched, and both wrote an `approval-acted` audit record. Both callers were told True, so both cards said "Answered". The question path never had this: `answer_notification` claims the row in one transaction and the loser gets False. The code is older than this branch, but this branch is what makes the race ordinary rather than theoretical — it puts approval cards in channels with as many pressers as the channel has members. `claim_pending_approval` stamps `answered_at` while the row is pending and unstamped, in one write under the store's write lock, and reports whether it changed a row. One caller goes on to the dispatcher; the other returns False and its card reads "Already answered or expired". The claim does not touch the status, because a dispatcher may answer with a snooze and the row has to stay pending for that. `snooze_notification` therefore clears the stamp, so the card that resurfaces can be pressed again. A claim left behind by a process that died mid-dispatch refuses later presses until the expiry sweep takes the row, which is what an approval nobody answers does in any case. The settling write no longer has its result dropped. The expiry sweep can take a claimed row while its dispatcher runs, which leaves the action done and the row expired; that now logs rather than passing in silence. Verification: 3728 passed, no failures, five of them the tests added here. The concurrency test runs two presses through `asyncio.gather` against a dispatcher that holds the dispatch open for 50ms. Against the code before this commit both answers come back True and the dispatcher records two decisions, which is the reported failure. Co-Authored-By: Claude Opus 5 (1M context) --- nerve/db/notifications.py | 32 +++++++++- nerve/notifications/service.py | 40 +++++++++--- tests/test_notification_lifecycle.py | 86 ++++++++++++++++++++++++++ tests/test_notifications_actionable.py | 35 +++++++++++ 4 files changed, 183 insertions(+), 10 deletions(-) diff --git a/nerve/db/notifications.py b/nerve/db/notifications.py index 068640636..10f4a52cf 100644 --- a/nerve/db/notifications.py +++ b/nerve/db/notifications.py @@ -142,6 +142,31 @@ async def answer_notification( ) return True + async def claim_pending_approval(self, notification_id: str) -> bool: + """Take a pending approval, so one presser acts on it. + + An approval runs its dispatcher before the row is settled, and that + can take seconds, so reading the status first cannot decide who + acts: two presses arriving together both see a pending row. This + write decides. It stamps ``answered_at`` only while the row is + pending and unstamped, so exactly one caller sees a row change and + the action behind the card runs once. + + The row stays pending, because the dispatcher may answer with a + snooze and keep it that way. ``answered_at`` alone is the claim, so + :meth:`snooze_notification` clears it and the card that resurfaces + can be pressed again. A claim left behind by a process that died + mid-dispatch refuses later presses until the expiry sweep takes the + row, which is what an approval nobody answers does in any case. + """ + now = datetime.now(timezone.utc).isoformat() + result = await self._write( + """UPDATE notifications SET answered_at = ? + WHERE id = ? AND status = 'pending' AND answered_at IS NULL""", + (now, notification_id), + ) + return result.rowcount == 1 + async def record_notification_delivery( self, notification_id: str, @@ -322,6 +347,10 @@ async def snooze_notification( ``redeliver_at`` again — each snooze buys another cycle, up to ``config.notifications.max_redeliveries``. + Clearing ``answered_at`` releases the claim + :meth:`claim_pending_approval` took, so the card that resurfaces + can be pressed again. + Returns True on success, False if the row is not pending. """ async with self._atomic(): @@ -332,7 +361,8 @@ async def snooze_notification( if not await cursor.fetchone(): return False await self.db.execute( - """UPDATE notifications SET redeliver_at = ?, expires_at = ? + """UPDATE notifications + SET redeliver_at = ?, expires_at = ?, answered_at = NULL WHERE id = ?""", (redeliver_at, new_expires_at, notification_id), ) diff --git a/nerve/notifications/service.py b/nerve/notifications/service.py index ab278f808..2c046578e 100644 --- a/nerve/notifications/service.py +++ b/nerve/notifications/service.py @@ -477,11 +477,12 @@ async def handle_answer( ``answered_by`` identifies the transport; ``actor`` identifies the person within that transport and is retained for audit. - - 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 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=approval`` rows: claim the row, look up the + dispatcher in the handler registry, run it, audit-log the + outcome, then flip 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 @@ -580,12 +581,27 @@ async def _handle_approval_answer( answered_by: str, actor: str | None = None, ) -> bool: - """Route an approval answer through the dispatcher registry.""" + """Route an approval answer through the dispatcher registry. + + Claims the row before the dispatcher runs. The status read in + :meth:`handle_answer` cannot decide who acts, because the dispatch + it precedes takes long enough for a second press to pass the same + read, and the action behind an approval card is not repeatable. A + card in a shared channel has as many pressers as the channel has + members, so the claim is what keeps one decision to one action. + """ notification_id = notif["id"] session_id = notif["session_id"] target_kind = notif.get("target_kind") or "" target_id = notif.get("target_id") or "" + if not await self.db.claim_pending_approval(notification_id): + logger.info( + "approval %s is already claimed; the %r press from %s is " + "ignored", notification_id, answer, actor or answered_by, + ) + return False + dispatcher = _handlers.get(target_kind) if target_kind else None if dispatcher is None: logger.warning( @@ -675,9 +691,15 @@ async def _handle_approval_answer( await self.db.snooze_notification( notification_id, snooze_until, new_expires_at, ) - else: - await self.db.answer_notification( - notification_id, answer, answered_by, actor=actor, + elif not await self.db.answer_notification( + notification_id, answer, answered_by, actor=actor, + ): + # The expiry sweep can take a claimed row while its dispatcher + # runs, which leaves the action done and the row expired. + logger.warning( + "approval %s stopped being pending during its dispatch; the " + "%r decision ran but the row does not record it", + notification_id, answer, ) from nerve.agent.streaming import broadcaster diff --git a/tests/test_notification_lifecycle.py b/tests/test_notification_lifecycle.py index 2eb2d23f4..458055ac9 100644 --- a/tests/test_notification_lifecycle.py +++ b/tests/test_notification_lifecycle.py @@ -710,6 +710,92 @@ async def test_telegram_edit_failure_is_swallowed( assert bot.edit_message_text.await_count == 2 +# ---------------------------------------------------------------------- +# Concurrent presses +# ---------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestConcurrentPresses: + """An approval card in a shared channel has many pressers. + + Every Socket Mode envelope is dispatched as its own task, so two + members pressing inside the same second run two answers at once. + """ + + async def test_two_presses_dispatch_the_action_once( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + audit_workspace, + ): + decisions: list[str] = [] + + async def dispatch(notification, target_id, decision, config): + decisions.append(decision) + # Hold the dispatch open, which is where the second press + # arrives: a real one runs a shell script. + await asyncio.sleep(0.05) + return _handlers.DispatchResult( + ok=True, + audit_event={ + "event": "approval-acted", + "notification_id": notification.get("id", ""), + "target_kind": "concurrency-test", + "target_id": target_id, + "decision": decision, + "ok": True, + }, + ) + + _handlers.register("concurrency-test", dispatch) + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + nid = await _make_approval(svc, db, target_kind="concurrency-test") + + alice, bob = await asyncio.gather( + svc.handle_answer(nid, "approve", "slack", actor="U-alice"), + svc.handle_answer(nid, "approve", "slack", actor="U-bob"), + ) + + assert sorted([alice, bob]) == [False, True] + assert decisions == ["approve"] + records = read_audit_jsonl( + audit_workspace / ".nerve" / "mechanical-actions", + ) + acted = [r for r in records if r.get("event") == "approval-acted"] + assert len(acted) == 1 + assert (await db.get_notification(nid))["status"] == "answered" + + async def test_a_snooze_leaves_the_card_answerable( + self, db: Database, fake_config, fake_engine, patch_broadcaster, + ): + # A snooze releases the claim, so the claim refuses a press for one + # round only, not for the life of the card. + def dispatch(notification, target_id, decision, config): + return _handlers.DispatchResult( + ok=True, + audit_event={ + "event": "approval-acted", + "target_kind": "claim-release-test", + "decision": decision, + "ok": True, + }, + snooze_until=_iso(24) if decision.startswith("snooze") else None, + ) + + _handlers.register("claim-release-test", dispatch) + svc = NotificationService(fake_config, db, fake_engine) + await db.create_session("s1") + nid = await _make_approval( + svc, db, target_kind="claim-release-test", expiry_hours=2, + ) + + assert await svc.handle_answer(nid, "snooze_24h", "web") is True + assert await svc.handle_answer(nid, "approve", "web") is True + notif = await db.get_notification(nid) + assert notif["status"] == "answered" + assert notif["answer"] == "approve" + + # ---------------------------------------------------------------------- # Answer attribution # ---------------------------------------------------------------------- diff --git a/tests/test_notifications_actionable.py b/tests/test_notifications_actionable.py index 3e456a19e..67b45acc5 100644 --- a/tests/test_notifications_actionable.py +++ b/tests/test_notifications_actionable.py @@ -273,6 +273,41 @@ async def test_snooze_notification_rejects_non_pending(self, db: Database): "n1", redeliver_at, new_expiry, ) is False + async def test_a_pending_approval_is_claimed_once(self, db: Database): + await db.create_session("s1") + await db.create_notification( + notification_id="n1", session_id="s1", type="approval", title="t", + ) + assert await db.claim_pending_approval("n1") is True + assert await db.claim_pending_approval("n1") is False + notif = await db.get_notification("n1") + assert notif["status"] == "pending" + assert notif["answered_at"] is not None + + async def test_a_settled_approval_cannot_be_claimed(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.claim_pending_approval("n1") is False + + async def test_a_snooze_releases_the_claim(self, db: Database): + await db.create_session("s1") + await db.create_notification( + notification_id="n1", session_id="s1", type="approval", title="t", + ) + assert await db.claim_pending_approval("n1") is True + assert await db.snooze_notification( + "n1", + (datetime.now(timezone.utc) + timedelta(hours=24)).isoformat(), + (datetime.now(timezone.utc) + timedelta(hours=72)).isoformat(), + ) is True + notif = await db.get_notification("n1") + assert notif["answered_at"] is None + # The card resurfaces answerable, which is the point of a snooze. + assert await db.claim_pending_approval("n1") is True + # ---------------------------------------------------------------------- # propose_action