test(notifications): add Phase 4 RPC and dismiss reducer coverage (#878)

app/src/components/notifications/NotificationCenter.tsx

@@ -30,6 +30,9 @@ const NotificationCenter = () => {
   // All providers seen across unfiltered loads — kept separate so the filter
   // pill row doesn't collapse when a provider filter is active.
   const [allProviders, setAllProviders] = useState<string[]>([]);
+  const visibleItems = items.filter(
+    n => n.status !== 'dismissed' && (!selectedProvider || n.provider === selectedProvider)
+  );
 
   // Fetch on mount and when provider filter changes.
   useEffect(() => {
@@ -80,10 +83,10 @@ const NotificationCenter = () => {
   };
 
   // Unread count scoped to the currently displayed (filtered) items.
-  const filteredUnreadCount = items.filter(n => n.status === 'unread').length;
+  const filteredUnreadCount = visibleItems.filter(n => n.status === 'unread').length;
 
   const handleMarkAllRead = async () => {
-    const unreadIds = items.filter(n => n.status === 'unread').map(n => n.id);
+    const unreadIds = visibleItems.filter(n => n.status === 'unread').map(n => n.id);
     for (const id of unreadIds) {
       dispatch(markIntegrationRead(id));
       try {
@@ -158,7 +161,7 @@ const NotificationCenter = () => {
           </div>
         )}
 
-        {!loading && !error && items.length === 0 && (
+        {!loading && !error && visibleItems.length === 0 && (
           <div className="flex flex-col items-center justify-center py-16 text-stone-400">
             <svg
               className="w-10 h-10 mb-3 opacity-40"
@@ -179,9 +182,9 @@ const NotificationCenter = () => {
           </div>
         )}
 
-        {!loading && !error && items.length > 0 && (
+        {!loading && !error && visibleItems.length > 0 && (
           <div className="divide-y-0">
-            {items.map(n => (
+            {visibleItems.map(n => (
               <NotificationCard
                 key={n.id}
                 notification={n}

app/src/store/__tests__/notificationsSlice.dismissActions.test.ts

@@ -0,0 +1,83 @@
+import { describe, expect, it } from 'vitest';
+
+import type { IntegrationNotification } from '../../types/notifications';
+import notificationReducer, {
+  dismissIntegrationNotification,
+  setIntegrationNotifications,
+} from '../notificationSlice';
+
+const makeNotification = (
+  overrides: Partial<IntegrationNotification> = {}
+): IntegrationNotification => ({
+  id: 'n-1',
+  provider: 'slack',
+  title: 'Test notification',
+  body: 'Hello',
+  raw_payload: {},
+  status: 'unread',
+  received_at: '2026-04-23T00:00:00Z',
+  ...overrides,
+});
+
+const baseState = notificationReducer(undefined, { type: '@@init' });
+const initialState = {
+  ...baseState,
+  integrationItems: [],
+  integrationUnreadCount: 0,
+  integrationLoading: false,
+  integrationError: null,
+};
+
+describe('notificationSlice — dismissIntegrationNotification', () => {
+  it('marks unread notifications as dismissed and decrements unread count', () => {
+    const unread = makeNotification({ id: 'n-unread', status: 'unread' });
+    const loaded = notificationReducer(
+      initialState,
+      setIntegrationNotifications({ items: [unread], unread_count: 1 })
+    );
+
+    const state = notificationReducer(loaded, dismissIntegrationNotification('n-unread'));
+
+    expect(state.integrationItems[0].status).toBe('dismissed');
+    expect(state.integrationUnreadCount).toBe(0);
+  });
+
+  it('marks read notifications as dismissed without changing unread count', () => {
+    const read = makeNotification({ id: 'n-read', status: 'read' });
+    const loaded = notificationReducer(
+      initialState,
+      setIntegrationNotifications({ items: [read], unread_count: 0 })
+    );
+
+    const state = notificationReducer(loaded, dismissIntegrationNotification('n-read'));
+
+    expect(state.integrationItems[0].status).toBe('dismissed');
+    expect(state.integrationUnreadCount).toBe(0);
+  });
+
+  it('is a no-op when the id does not exist', () => {
+    const item = makeNotification({ id: 'n-1', status: 'unread' });
+    const loaded = notificationReducer(
+      initialState,
+      setIntegrationNotifications({ items: [item], unread_count: 1 })
+    );
+
+    const state = notificationReducer(loaded, dismissIntegrationNotification('does-not-exist'));
+
+    expect(state.integrationItems[0].status).toBe('unread');
+    expect(state.integrationUnreadCount).toBe(1);
+  });
+
+  it('does not decrement unread count when dismissing an already dismissed item', () => {
+    const dismissed = makeNotification({ id: 'n-dismissed', status: 'dismissed' });
+    const loaded = notificationReducer(
+      initialState,
+      setIntegrationNotifications({ items: [dismissed], unread_count: 0 })
+    );
+
+    const state = notificationReducer(loaded, dismissIntegrationNotification('n-dismissed'));
+
+    expect(state.integrationItems[0].status).toBe('dismissed');
+    expect(state.integrationUnreadCount).toBe(0);
+  });
+});

src/openhuman/notifications/store.rs

@@ -118,62 +118,71 @@ pub fn insert(config: &Config, n: &IntegrationNotification) -> Result<()> {
 /// Returns `true` when inserted, `false` when skipped as duplicate.
 pub fn insert_if_not_recent(config: &Config, n: &IntegrationNotification) -> Result<bool> {
     with_connection(config, |conn| {
-        let tx = conn
-            .unchecked_transaction()
+        conn.execute_batch("BEGIN IMMEDIATE")
             .context("[notifications::store] begin insert_if_not_recent tx failed")?;
 
-        let count: i64 = match n.account_id.as_deref() {
-            Some(aid) => tx.query_row(
-                "SELECT COUNT(*) FROM integration_notifications
-                 WHERE provider = ?1 AND account_id = ?2
-                   AND title = ?3 AND body = ?4
-                   AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')",
-                params![&n.provider, aid, &n.title, &n.body],
-                |row| row.get(0),
-            ),
-            None => tx.query_row(
-                "SELECT COUNT(*) FROM integration_notifications
-                 WHERE provider = ?1 AND account_id IS NULL
-                   AND title = ?2 AND body = ?3
-                   AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')",
-                params![&n.provider, &n.title, &n.body],
-                |row| row.get(0),
-            ),
-        }
-        .context("[notifications::store] insert_if_not_recent dedup query failed")?;
+        let result: Result<bool> = (|| {
+            let count: i64 = match n.account_id.as_deref() {
+                Some(aid) => conn.query_row(
+                    "SELECT COUNT(*) FROM integration_notifications
+                     WHERE provider = ?1 AND account_id = ?2
+                       AND title = ?3 AND body = ?4
+                       AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')",
+                    params![&n.provider, aid, &n.title, &n.body],
+                    |row| row.get(0),
+                ),
+                None => conn.query_row(
+                    "SELECT COUNT(*) FROM integration_notifications
+                     WHERE provider = ?1 AND account_id IS NULL
+                       AND title = ?2 AND body = ?3
+                       AND unixepoch(received_at) >= unixepoch('now', '-60 seconds')",
+                    params![&n.provider, &n.title, &n.body],
+                    |row| row.get(0),
+                ),
+            }
+            .context("[notifications::store] insert_if_not_recent dedup query failed")?;
 
-        if count > 0 {
-            tx.commit()
-                .context("[notifications::store] commit duplicate tx failed")?;
-            return Ok(false);
+            if count > 0 {
+                return Ok(false);
+            }
+
+            conn.execute(
+                "INSERT INTO integration_notifications
+                 (id, provider, account_id, title, body, raw_payload,
+                  importance_score, triage_action, triage_reason, status,
+                  received_at, scored_at)
+                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
+                params![
+                    n.id,
+                    n.provider,
+                    n.account_id,
+                    n.title,
+                    n.body,
+                    n.raw_payload.to_string(),
+                    n.importance_score,
+                    n.triage_action,
+                    n.triage_reason,
+                    n.status.as_str(),
+                    n.received_at.to_rfc3339(),
+                    n.scored_at.map(|t| t.to_rfc3339()),
+                ],
+            )
+            .context("[notifications::store] insert_if_not_recent insert failed")?;
+
+            Ok(true)
+        })();
+
+        if result.is_ok() {
+            conn.execute_batch("COMMIT")
+                .context("[notifications::store] commit insert_if_not_recent tx failed")?;
+        } else if let Err(rollback_err) = conn.execute_batch("ROLLBACK") {
+            tracing::warn!(
+                error = %rollback_err,
+                "[notifications::store] rollback insert_if_not_recent tx failed"
+            );
         }
 
-        tx.execute(
-            "INSERT INTO integration_notifications
-             (id, provider, account_id, title, body, raw_payload,
-              importance_score, triage_action, triage_reason, status,
-              received_at, scored_at)
-             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
-            params![
-                n.id,
-                n.provider,
-                n.account_id,
-                n.title,
-                n.body,
-                n.raw_payload.to_string(),
-                n.importance_score,
-                n.triage_action,
-                n.triage_reason,
-                n.status.as_str(),
-                n.received_at.to_rfc3339(),
-                n.scored_at.map(|t| t.to_rfc3339()),
-            ],
-        )
-        .context("[notifications::store] insert_if_not_recent insert failed")?;
-
-        tx.commit()
-            .context("[notifications::store] commit insert_if_not_recent tx failed")?;
-        Ok(true)
+        result
     })
 }
 
@@ -581,6 +590,7 @@ fn row_to_notification(row: &rusqlite::Row<'_>) -> Result<IntegrationNotificatio
 mod tests {
     use super::*;
     use crate::openhuman::config::Config;
+    use std::sync::{Arc, Barrier};
     use tempfile::TempDir;
 
     fn test_config(dir: &TempDir) -> Config {
@@ -683,6 +693,34 @@ mod tests {
         assert!(insert_if_not_recent(&config, &fresh_same_content).unwrap());
     }
 
+    #[test]
+    fn insert_if_not_recent_is_atomic_under_concurrent_calls() {
+        let dir = TempDir::new().unwrap();
+        let config = Arc::new(test_config(&dir));
+        let gate = Arc::new(Barrier::new(3));
+
+        let run = |id: &'static str, gate: Arc<Barrier>, config: Arc<Config>| {
+            std::thread::spawn(move || {
+                let n = sample_notification(id, "slack");
+                gate.wait();
+                insert_if_not_recent(&config, &n)
+            })
+        };
+
+        let t1 = run("race-a", Arc::clone(&gate), Arc::clone(&config));
+        let t2 = run("race-b", Arc::clone(&gate), Arc::clone(&config));
+
+        gate.wait();
+        let inserted_1 = t1.join().unwrap().unwrap();
+        let inserted_2 = t2.join().unwrap().unwrap();
+
+        let inserted_total = usize::from(inserted_1) + usize::from(inserted_2);
+        assert_eq!(inserted_total, 1);
+
+        let items = list(&config, 10, 0, Some("slack"), None).unwrap();
+        assert_eq!(items.len(), 1);
+    }
+
     #[test]
     fn exists_recent_rejects_expired_notification() {
         let dir = TempDir::new().unwrap();

tests/json_rpc_e2e.rs

@@ -175,11 +175,31 @@ fn mock_upstream_router() -> Router {
         if let Some(model) = body.get("model").and_then(Value::as_str) {
             with_chat_completion_models(|models| models.push(model.to_string()));
         }
+        let is_triage_turn = body
+            .get("messages")
+            .and_then(Value::as_array)
+            .map(|messages| {
+                messages.iter().any(|m| {
+                    m.get("content")
+                        .and_then(Value::as_str)
+                        .is_some_and(|content| {
+                            content.contains("SOURCE: ")
+                                && content.contains("DISPLAY_LABEL: ")
+                                && content.contains("PAYLOAD:")
+                        })
+                })
+            })
+            .unwrap_or(false);
+        let content = if is_triage_turn {
+            "{\"action\":\"react\",\"reason\":\"e2e triage mock\"}"
+        } else {
+            "Hello from e2e mock agent"
+        };
         Json(json!({
             "choices": [{
                 "message": {
                     "role": "assistant",
-                    "content": "Hello from e2e mock agent"
+                    "content": content
                 }
             }]
         }))