feat(notifications): add dismiss/stats RPC, DomainEvents, and triage bus integration (#875)

app/src/lib/webviewNotifications/service.test.ts

@@ -1,9 +1,15 @@
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
+import { ingestNotification } from '../../services/notificationService';
 import { store } from '../../store';
 import { addAccount } from '../../store/accountsSlice';
+import { setNotifications } from '../../store/notificationsSlice';
 import { __handleFiredForTests, __resetForTests, handleNotificationClick } from './service';
 
+vi.mock('../../services/notificationService', () => ({
+  ingestNotification: vi.fn().mockResolvedValue({ skipped: true, reason: 'test-default' }),
+}));
+
 const sampleAccount = {
   id: 'acct1',
   provider: 'slack' as const,
@@ -12,9 +18,33 @@ const sampleAccount = {
   status: 'open' as const,
 };
 
+function makeFiredPayload(
+  overrides: Partial<{
+    account_id: string;
+    provider: 'slack';
+    title: string;
+    body: string;
+    tag: string | null;
+  }> = {}
+) {
+  return {
+    account_id: 'acct1',
+    provider: 'slack' as const,
+    title: 'OpenHuman: Slack - Ping',
+    body: 'hi',
+    tag: null,
+    ...overrides,
+  };
+}
+
 describe('webviewNotifications service', () => {
+  const ingestNotificationMock = vi.mocked(ingestNotification);
+
   beforeEach(() => {
     __resetForTests();
+    ingestNotificationMock.mockReset();
+    ingestNotificationMock.mockResolvedValue({ skipped: true, reason: 'test-default' });
+    store.dispatch(setNotifications({ items: [], unread_count: 0 }));
     store.dispatch(addAccount(sampleAccount));
   });
 
@@ -24,24 +54,13 @@ describe('webviewNotifications service', () => {
 
   it('fired events increment unread via Redux', () => {
     const before = store.getState().accounts.unread.acct1 ?? 0;
-    __handleFiredForTests({
-      account_id: 'acct1',
-      provider: 'slack',
-      title: 'OpenHuman: Slack — Ping',
-      body: 'hi',
-      tag: null,
-    });
+    __handleFiredForTests(makeFiredPayload());
     const after = store.getState().accounts.unread.acct1 ?? 0;
     expect(after).toBe(before + 1);
   });
 
   it('handleNotificationClick focuses account and clears unread', () => {
-    __handleFiredForTests({
-      account_id: 'acct1',
-      provider: 'slack',
-      title: 'OpenHuman: Slack — Ping',
-      body: '',
-    });
+    __handleFiredForTests(makeFiredPayload({ body: '' }));
     expect(store.getState().accounts.unread.acct1).toBeGreaterThan(0);
 
     handleNotificationClick('acct1');
@@ -50,7 +69,37 @@ describe('webviewNotifications service', () => {
   });
 
   it('fired events for unknown accounts are no-ops', () => {
-    __handleFiredForTests({ account_id: 'ghost', provider: 'slack', title: 't', body: 'b' });
+    __handleFiredForTests(makeFiredPayload({ account_id: 'ghost', title: 't', body: 'b' }));
     expect(store.getState().accounts.unread.ghost).toBeUndefined();
   });
+
+  it('ingest success adds integration notification', async () => {
+    ingestNotificationMock.mockResolvedValue({ id: 'notif-1', skipped: false });
+    __handleFiredForTests(makeFiredPayload({ title: 'Hello', body: 'World', tag: 'message' }));
+
+    await Promise.resolve();
+
+    const items = store.getState().integrationNotifications.items;
+    expect(items.some(item => item.id === 'notif-1')).toBe(true);
+  });
+
+  it('ingest skipped does not add integration notification', async () => {
+    ingestNotificationMock.mockResolvedValue({ skipped: true, reason: 'duplicate' });
+    __handleFiredForTests(makeFiredPayload({ title: 'Hello', body: 'World', tag: 'message' }));
+
+    await Promise.resolve();
+
+    const items = store.getState().integrationNotifications.items;
+    expect(items).toHaveLength(0);
+  });
+
+  it('ingest error does not add integration notification', async () => {
+    ingestNotificationMock.mockRejectedValue(new Error('network down'));
+    __handleFiredForTests(makeFiredPayload({ title: 'Hello', body: 'World', tag: 'message' }));
+
+    await Promise.resolve();
+
+    const items = store.getState().integrationNotifications.items;
+    expect(items).toHaveLength(0);
+  });
 });

app/src/lib/webviewNotifications/service.ts

@@ -18,6 +18,12 @@ const errLog = debug('webview-notifications:error');
 let started = false;
 let unlisten: UnlistenFn | null = null;
 
+function redactAccountId(accountId: string): string {
+  if (!accountId) return 'redacted';
+  if (accountId.length <= 4) return '***';
+  return `***${accountId.slice(-4)}`;
+}
+
 /**
  * Subscribe to `webview-notification:fired` events from the Tauri shell and
  * mirror each fire into Redux so the sidebar can bump an unread badge on
@@ -63,9 +69,10 @@ export function handleNotificationClick(accountId: string): void {
 
 function handleFired(payload: WebviewNotificationFired): void {
   const { account_id: accountId, provider, title, body, tag } = payload;
+  const redactedAccountId = redactAccountId(accountId);
   log(
     'fired account=%s provider=%s title_chars=%d body_chars=%d',
-    accountId,
+    redactedAccountId,
     provider,
     title.length,
     body.length
@@ -88,16 +95,16 @@ function handleFired(payload: WebviewNotificationFired): void {
 
   // Mirror into the core triage pipeline — fire-and-forget.
   log(
-    '[notification_intel] forwarding to core ingest provider=%s account_present=%s',
+    '[notification_intel] forwarding to core ingest provider=%s account=%s',
     provider,
-    accountId ? 'yes' : 'no'
+    redactedAccountId
   );
   void ingestNotification({
     provider,
     account_id: accountId,
     title,
     body,
-    raw_payload: { tag, provider, account_id: accountId, title, body },
+    raw_payload: payload as unknown as Record<string, unknown>,
   })
     .then(result => {
       if (!result.skipped) {
@@ -109,7 +116,7 @@ function handleFired(payload: WebviewNotificationFired): void {
             account_id: accountId,
             title,
             body,
-            raw_payload: { tag, provider, account_id: accountId, title, body },
+            raw_payload: payload as unknown as Record<string, unknown>,
             status: 'unread',
             received_at: new Date().toISOString(),
             importance_score: undefined,

app/src/services/notificationService.ts

@@ -48,6 +48,13 @@ export async function markNotificationRead(id: string): Promise<void> {
 }
 
 type NotificationIngestResult = { id: string; skipped?: false } | { skipped: true; reason: string };
+type NotificationStats = {
+  total: number;
+  unread: number;
+  unscored: number;
+  by_provider: Record<string, number>;
+  by_action: Record<string, number>;
+};
 
 /**
  * Ingest a new notification via the core RPC pipeline.
@@ -106,3 +113,51 @@ export async function setNotificationSettings(payload: {
     params: payload,
   });
 }
+
+export async function dismissNotification(id: string): Promise<void> {
+  log('dismissNotification id=%s', id);
+  try {
+    await callCoreRpc<{ ok: boolean }>({
+      method: 'openhuman.notification_dismiss',
+      params: { id },
+    });
+    log('dismissNotification ok id=%s', id);
+  } catch (err) {
+    errLog('dismissNotification failed id=%s: %o', id, err);
+    throw err;
+  }
+}
+
+export async function markNotificationActed(id: string): Promise<void> {
+  log('markNotificationActed id=%s', id);
+  try {
+    await callCoreRpc<{ ok: boolean }>({
+      method: 'openhuman.notification_mark_acted',
+      params: { id },
+    });
+    log('markNotificationActed ok id=%s', id);
+  } catch (err) {
+    errLog('markNotificationActed failed id=%s: %o', id, err);
+    throw err;
+  }
+}
+
+export async function fetchNotificationStats(): Promise<NotificationStats> {
+  log('fetchNotificationStats');
+  try {
+    const result = await callCoreRpc<NotificationStats>({
+      method: 'openhuman.notification_stats',
+      params: {},
+    });
+    log(
+      'fetchNotificationStats ok total=%d unread=%d unscored=%d',
+      result.total,
+      result.unread,
+      result.unscored
+    );
+    return result;
+  } catch (err) {
+    errLog('fetchNotificationStats failed: %o', err);
+    throw err;
+  }
+}

src/core/event_bus/events.rs

@@ -310,6 +310,25 @@ pub enum DomainEvent {
     /// A full tree rebuild completed.
     TreeSummarizerRebuildCompleted { namespace: String, total_nodes: u64 },
 
+    // ── Notification ────────────────────────────────────────────────────
+    /// An integration notification was ingested from an embedded webview.
+    NotificationIngested {
+        id: String,
+        provider: String,
+        account_id: Option<String>,
+    },
+    /// An integration notification's triage scoring completed.
+    NotificationTriaged {
+        id: String,
+        provider: String,
+        /// One of: "drop", "acknowledge", "react", "escalate"
+        action: String,
+        importance_score: f32,
+        latency_ms: u64,
+        /// True when the triage result was actually routed to the orchestrator path.
+        routed: bool,
+    },
+
     // ── System lifecycle ────────────────────────────────────────────────
     /// A system component started up.
     SystemStartup { component: String },
@@ -379,6 +398,8 @@ impl DomainEvent {
             | Self::TreeSummarizerPropagated { .. }
             | Self::TreeSummarizerRebuildCompleted { .. } => "tree_summarizer",
 
+            Self::NotificationIngested { .. } | Self::NotificationTriaged { .. } => "notification",
+
             Self::SystemStartup { .. }
             | Self::SystemShutdown { .. }
             | Self::SystemRestartRequested { .. }
@@ -753,6 +774,26 @@ mod tests {
                 },
                 "tree_summarizer",
             ),
+            // Notification
+            (
+                DomainEvent::NotificationIngested {
+                    id: "n1".into(),
+                    provider: "slack".into(),
+                    account_id: None,
+                },
+                "notification",
+            ),
+            (
+                DomainEvent::NotificationTriaged {
+                    id: "n1".into(),
+                    provider: "slack".into(),
+                    action: "escalate".into(),
+                    importance_score: 0.9,
+                    latency_ms: 150,
+                    routed: true,
+                },
+                "notification",
+            ),
             // System
             (
                 DomainEvent::SystemStartup {

src/openhuman/about_app/catalog.rs

@@ -186,6 +186,36 @@ const CAPABILITIES: &[Capability] = &[
         status: CapabilityStatus::Beta,
         privacy: None,
     },
+    Capability {
+        id: "intelligence.notifications_dismiss",
+        name: "Dismiss Notifications",
+        domain: "intelligence",
+        category: CapabilityCategory::Intelligence,
+        description: "Dismiss low-value notifications from the intelligence inbox.",
+        how_to: "Notifications > Item actions",
+        status: CapabilityStatus::Beta,
+        privacy: None,
+    },
+    Capability {
+        id: "intelligence.notifications_mark_acted",
+        name: "Mark Notifications Acted",
+        domain: "intelligence",
+        category: CapabilityCategory::Intelligence,
+        description: "Mark a notification as acted upon after taking follow-up action.",
+        how_to: "Notifications > Item actions",
+        status: CapabilityStatus::Beta,
+        privacy: None,
+    },
+    Capability {
+        id: "intelligence.notifications_stats",
+        name: "View Notification Stats",
+        domain: "intelligence",
+        category: CapabilityCategory::Intelligence,
+        description: "View aggregate unread, unscored, and provider/action notification stats.",
+        how_to: "Notifications > Summary cards",
+        status: CapabilityStatus::Beta,
+        privacy: None,
+    },
     Capability {
         id: "skills.discover",
         name: "Discover Skills",

src/openhuman/notifications/bus.rs

@@ -137,6 +137,27 @@ pub fn event_to_notification(event: &DomainEvent) -> Option<CoreNotificationEven
             deep_link: Some("/chat".into()),
             timestamp_ms: ts,
         }),
+        DomainEvent::NotificationTriaged {
+            id,
+            provider,
+            action,
+            importance_score,
+            latency_ms,
+            routed,
+        } if *routed && (action == "escalate" || action == "react") => {
+            Some(CoreNotificationEvent {
+                id: format!("notification-triaged:{}:{}:{}", id, action, latency_ms),
+                category: CoreNotificationCategory::Agents,
+                title: format!("High-priority {} notification", provider),
+                body: format!(
+                    "Action: {} (score: {:.0}%). Routed to orchestrator.",
+                    action,
+                    importance_score * 100.0
+                ),
+                deep_link: Some("/notifications".into()),
+                timestamp_ms: ts,
+            })
+        }
         _ => None,
     }
 }
@@ -278,4 +299,46 @@ mod tests {
         };
         assert!(event_to_notification(&ev).is_none());
     }
+
+    #[test]
+    fn notification_triaged_escalate_produces_agents_notification() {
+        let ev = DomainEvent::NotificationTriaged {
+            id: "n1".into(),
+            provider: "slack".into(),
+            action: "escalate".into(),
+            importance_score: 0.9,
+            latency_ms: 100,
+            routed: true,
+        };
+        let n = event_to_notification(&ev).expect("should produce notification");
+        assert_eq!(n.category, CoreNotificationCategory::Agents);
+        assert!(n.body.contains("escalate"));
+        assert!(n.deep_link.as_deref() == Some("/notifications"));
+    }
+
+    #[test]
+    fn notification_triaged_drop_is_silent() {
+        let ev = DomainEvent::NotificationTriaged {
+            id: "n1".into(),
+            provider: "gmail".into(),
+            action: "drop".into(),
+            importance_score: 0.1,
+            latency_ms: 50,
+            routed: false,
+        };
+        assert!(event_to_notification(&ev).is_none());
+    }
+
+    #[test]
+    fn notification_triaged_unrouted_escalate_is_silent() {
+        let ev = DomainEvent::NotificationTriaged {
+            id: "n1".into(),
+            provider: "slack".into(),
+            action: "escalate".into(),
+            importance_score: 0.9,
+            latency_ms: 100,
+            routed: false,
+        };
+        assert!(event_to_notification(&ev).is_none());
+    }
 }

src/openhuman/notifications/rpc.rs

@@ -1,14 +1,18 @@
 //! JSON-RPC handler functions for the notifications domain.
 //!
-//! Three endpoints:
+//! Notification endpoints:
 //!  - `notification_ingest`   — write a new notification, kick off background triage
 //!  - `notifications_list`    — paginated query with optional provider / min-score filters
 //!  - `notification_mark_read`— mark a single notification as read
+//!  - `notification_dismiss`  — mark a single notification as dismissed
+//!  - `notification_mark_acted` — mark a single notification as acted upon
+//!  - `notification_stats`    — return aggregate pipeline statistics
 
 use chrono::Utc;
 use serde_json::{json, Map, Value};
 use uuid::Uuid;
 
+use crate::core::event_bus::{publish_global, DomainEvent};
 use crate::openhuman::agent::triage::{apply_decision, run_triage, TriggerEnvelope, TriggerSource};
 use crate::openhuman::config::rpc as config_rpc;
 use crate::rpc::RpcOutcome;
@@ -28,6 +32,7 @@ use super::types::{
 /// Writes the record immediately, returns the new `id`, then spawns a
 /// background task to run the triage pipeline and back-fill the score.
 pub async fn handle_ingest(params: Map<String, Value>) -> Result<Value, String> {
+    let ingest_started_at = Utc::now();
     let config = config_rpc::load_config_with_timeout().await?;
 
     let req: NotificationIngestRequest = serde_json::from_value(Value::Object(params.clone()))
@@ -54,7 +59,7 @@ pub async fn handle_ingest(params: Map<String, Value>) -> Result<Value, String>
         triage_action: None,
         triage_reason: None,
         status: NotificationStatus::Unread,
-        received_at: Utc::now(),
+        received_at: ingest_started_at,
         scored_at: None,
     };
 
@@ -96,7 +101,7 @@ pub async fn handle_ingest(params: Map<String, Value>) -> Result<Value, String>
                 "body": req.body,
                 "raw": req.raw_payload,
             }),
-            received_at: Utc::now(),
+            received_at: ingest_started_at,
         };
 
         match run_triage(&envelope).await {
@@ -125,21 +130,76 @@ pub async fn handle_ingest(params: Map<String, Value>) -> Result<Value, String>
                         error = %e,
                         "[notification_intel] failed to persist triage result"
                     );
+                    return;
                 }
 
-                // Auto-escalate high-importance notifications to the orchestrator.
-                if (action == "escalate" || action == "react")
-                    && score >= provider_settings.importance_threshold
-                    && provider_settings.route_to_orchestrator
-                {
-                    if let Err(e) = apply_decision(triage_run, &envelope).await {
-                        tracing::warn!(
-                            id = %id_for_triage,
-                            error = %e,
-                            "[notification_intel] apply_decision failed"
-                        );
+                // Compute triage latency from ingest time.
+                let latency_ms = Utc::now()
+                    .signed_duration_since(ingest_started_at)
+                    .num_milliseconds()
+                    .max(0) as u64;
+
+                let mut routed = false;
+                let route_candidate = action == "escalate" || action == "react";
+                if route_candidate {
+                    // Re-read provider settings right before potential escalation so
+                    // runtime toggles apply even while triage is in-flight.
+                    let latest_settings = match store::get_settings(
+                        &config_for_triage,
+                        &req.provider,
+                    ) {
+                        Ok(s) => s,
+                        Err(e) => {
+                            tracing::warn!(
+                                id = %id_for_triage,
+                                provider = %req.provider,
+                                error = %e,
+                                "[notification_intel] failed to refresh provider settings for routing gate"
+                            );
+                            publish_global(DomainEvent::NotificationTriaged {
+                                id: id_for_triage.clone(),
+                                provider: req.provider.clone(),
+                                action: action.clone(),
+                                importance_score: score,
+                                latency_ms,
+                                routed: false,
+                            });
+                            return;
+                        }
+                    };
+
+                    // Auto-escalate high-importance notifications to the orchestrator.
+                    if score >= latest_settings.importance_threshold
+                        && latest_settings.route_to_orchestrator
+                    {
+                        if let Err(e) = apply_decision(triage_run, &envelope).await {
+                            tracing::warn!(
+                                id = %id_for_triage,
+                                error = %e,
+                                "[notification_intel] apply_decision failed"
+                            );
+                        } else {
+                            routed = true;
+                        }
                     }
                 }
+
+                publish_global(DomainEvent::NotificationTriaged {
+                    id: id_for_triage.clone(),
+                    provider: req.provider.clone(),
+                    action: action.clone(),
+                    importance_score: score,
+                    latency_ms,
+                    routed,
+                });
+                tracing::debug!(
+                    id = %id_for_triage,
+                    action = %action,
+                    score = score,
+                    latency_ms = latency_ms,
+                    routed = routed,
+                    "[notification_intel] published NotificationTriaged event"
+                );
             }
             Err(e) => {
                 tracing::warn!(
@@ -248,6 +308,65 @@ pub async fn handle_settings_set(params: Map<String, Value>) -> Result<Value, St
     outcome.into_cli_compatible_json()
 }
 
+// ─────────────────────────────────────────────────────────────────────────────
+// notification_dismiss
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Mark a single notification as dismissed.
+pub async fn handle_dismiss(params: Map<String, Value>) -> Result<Value, String> {
+    let config = config_rpc::load_config_with_timeout().await?;
+    let id = params
+        .get("id")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| "[notification_intel] missing required param 'id'".to_string())?
+        .to_string();
+
+    let updated = store::mark_dismissed(&config, &id)
+        .map_err(|e| format!("[notification_intel] mark_dismissed failed: {e}"))?;
+    tracing::debug!(id = %id, updated = updated, "[notification_intel] notification dismissed");
+    let outcome = RpcOutcome::new(json!({ "ok": updated }), vec![]);
+    outcome.into_cli_compatible_json()
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// notification_mark_acted
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Mark a single notification as acted upon.
+pub async fn handle_mark_acted(params: Map<String, Value>) -> Result<Value, String> {
+    let config = config_rpc::load_config_with_timeout().await?;
+    let id = params
+        .get("id")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| "[notification_intel] missing required param 'id'".to_string())?
+        .to_string();
+
+    let updated = store::mark_acted(&config, &id)
+        .map_err(|e| format!("[notification_intel] mark_acted failed: {e}"))?;
+    tracing::debug!(
+        id = %id,
+        updated = updated,
+        "[notification_intel] notification marked acted"
+    );
+    let outcome = RpcOutcome::new(json!({ "ok": updated }), vec![]);
+    outcome.into_cli_compatible_json()
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// notification_stats
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Return aggregate pipeline statistics.
+pub async fn handle_stats(_params: Map<String, Value>) -> Result<Value, String> {
+    let config = config_rpc::load_config_with_timeout().await?;
+    tracing::debug!("[notification_intel] stats requested");
+
+    let s = store::stats(&config).map_err(|e| format!("[notification_intel] stats failed: {e}"))?;
+
+    let outcome = RpcOutcome::new(json!(s), vec![]);
+    outcome.into_cli_compatible_json()
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Helpers
 // ─────────────────────────────────────────────────────────────────────────────

src/openhuman/notifications/schemas.rs

@@ -19,6 +19,9 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
         schemas("mark_read"),
         schemas("settings_get"),
         schemas("settings_set"),
+        schemas("dismiss"),
+        schemas("mark_acted"),
+        schemas("stats"),
     ]
 }
 
@@ -44,6 +47,18 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
             schema: schemas("settings_set"),
             handler: handle_settings_set_wrap,
         },
+        RegisteredController {
+            schema: schemas("dismiss"),
+            handler: handle_dismiss_wrap,
+        },
+        RegisteredController {
+            schema: schemas("mark_acted"),
+            handler: handle_mark_acted_wrap,
+        },
+        RegisteredController {
+            schema: schemas("stats"),
+            handler: handle_stats_wrap,
+        },
     ]
 }
 
@@ -237,6 +252,79 @@ pub fn schemas(function: &str) -> ControllerSchema {
             ],
         },
 
+        "dismiss" => ControllerSchema {
+            namespace: "notification",
+            function: "dismiss",
+            description: "Mark a notification as dismissed (user explicitly hid it).",
+            inputs: vec![FieldSchema {
+                name: "id",
+                ty: TypeSchema::String,
+                comment: "UUID of the notification to dismiss.",
+                required: true,
+            }],
+            outputs: vec![FieldSchema {
+                name: "ok",
+                ty: TypeSchema::Bool,
+                comment: "True when the update succeeded.",
+                required: true,
+            }],
+        },
+        "mark_acted" => ControllerSchema {
+            namespace: "notification",
+            function: "mark_acted",
+            description: "Mark a notification as acted upon (user took an action from it).",
+            inputs: vec![FieldSchema {
+                name: "id",
+                ty: TypeSchema::String,
+                comment: "UUID of the notification to mark as acted.",
+                required: true,
+            }],
+            outputs: vec![FieldSchema {
+                name: "ok",
+                ty: TypeSchema::Bool,
+                comment: "True when the update succeeded.",
+                required: true,
+            }],
+        },
+        "stats" => ControllerSchema {
+            namespace: "notification",
+            function: "stats",
+            description: "Return aggregate statistics for the notification intelligence pipeline.",
+            inputs: vec![],
+            outputs: vec![
+                FieldSchema {
+                    name: "total",
+                    ty: TypeSchema::I64,
+                    comment: "Total notification count.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "unread",
+                    ty: TypeSchema::I64,
+                    comment: "Count of unread notifications.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "unscored",
+                    ty: TypeSchema::I64,
+                    comment: "Count of notifications pending triage scoring.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "by_provider",
+                    ty: TypeSchema::Map(Box::new(TypeSchema::I64)),
+                    comment: "Notification counts grouped by provider slug.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "by_action",
+                    ty: TypeSchema::Map(Box::new(TypeSchema::I64)),
+                    comment: "Notification counts grouped by triage action.",
+                    required: true,
+                },
+            ],
+        },
+
         _other => ControllerSchema {
             namespace: "notification",
             function: "unknown",
@@ -281,6 +369,18 @@ fn handle_settings_set_wrap(params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move { super::rpc::handle_settings_set(params).await })
 }
 
+fn handle_dismiss_wrap(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { super::rpc::handle_dismiss(params).await })
+}
+
+fn handle_mark_acted_wrap(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { super::rpc::handle_mark_acted(params).await })
+}
+
+fn handle_stats_wrap(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { super::rpc::handle_stats(params).await })
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Tests
 // ─────────────────────────────────────────────────────────────────────────────
@@ -290,7 +390,7 @@ mod tests {
     use super::*;
 
     #[test]
-    fn all_controller_schemas_covers_three_functions() {
+    fn all_controller_schemas_covers_registered_functions() {
         let names: Vec<_> = all_controller_schemas()
             .into_iter()
             .map(|s| s.function)
@@ -302,7 +402,10 @@ mod tests {
                 "list",
                 "mark_read",
                 "settings_get",
-                "settings_set"
+                "settings_set",
+                "dismiss",
+                "mark_acted",
+                "stats",
             ]
         );
     }
@@ -310,7 +413,7 @@ mod tests {
     #[test]
     fn all_registered_controllers_has_handler_per_schema() {
         let controllers = all_registered_controllers();
-        assert_eq!(controllers.len(), 5);
+        assert_eq!(controllers.len(), 8);
         let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect();
         assert_eq!(
             names,
@@ -319,11 +422,62 @@ mod tests {
                 "list",
                 "mark_read",
                 "settings_get",
-                "settings_set"
+                "settings_set",
+                "dismiss",
+                "mark_acted",
+                "stats",
             ]
         );
     }
 
+    #[test]
+    fn schemas_dismiss_and_mark_acted_require_id_and_return_ok() {
+        let dismiss = schemas("dismiss");
+        assert_eq!(dismiss.inputs.len(), 1);
+        assert_eq!(dismiss.inputs[0].name, "id");
+        assert_eq!(dismiss.inputs[0].ty, TypeSchema::String);
+        assert!(dismiss.inputs[0].required);
+        assert_eq!(dismiss.outputs.len(), 1);
+        assert_eq!(dismiss.outputs[0].name, "ok");
+        assert_eq!(dismiss.outputs[0].ty, TypeSchema::Bool);
+        assert!(dismiss.outputs[0].required);
+
+        let mark_acted = schemas("mark_acted");
+        assert_eq!(mark_acted.inputs.len(), 1);
+        assert_eq!(mark_acted.inputs[0].name, "id");
+        assert_eq!(mark_acted.inputs[0].ty, TypeSchema::String);
+        assert!(mark_acted.inputs[0].required);
+        assert_eq!(mark_acted.outputs.len(), 1);
+        assert_eq!(mark_acted.outputs[0].name, "ok");
+        assert_eq!(mark_acted.outputs[0].ty, TypeSchema::Bool);
+        assert!(mark_acted.outputs[0].required);
+    }
+
+    #[test]
+    fn schemas_stats_matches_notification_stats_shape() {
+        let stats = schemas("stats");
+        assert!(stats.inputs.is_empty());
+        assert_eq!(stats.outputs.len(), 5);
+
+        let expected = [
+            ("total", TypeSchema::I64),
+            ("unread", TypeSchema::I64),
+            ("unscored", TypeSchema::I64),
+            ("by_provider", TypeSchema::Map(Box::new(TypeSchema::I64))),
+            ("by_action", TypeSchema::Map(Box::new(TypeSchema::I64))),
+        ];
+
+        for (name, ty) in expected {
+            let field = stats
+                .outputs
+                .iter()
+                .find(|f| f.name == name)
+                .unwrap_or_else(|| panic!("missing stats output field `{name}`"));
+            assert_eq!(field.ty, ty, "unexpected type for stats.{name}");
+            assert!(field.required, "stats.{name} should be required");
+        }
+    }
+
     #[test]
     fn schemas_ingest_requires_provider_title_body_raw_payload() {
         let s = schemas("ingest");

src/openhuman/notifications/store.rs

@@ -298,6 +298,165 @@ pub fn unread_count(config: &Config) -> Result<i64> {
     })
 }
 
+/// Check whether a notification with identical content was received in the
+/// last 60 seconds.
+pub fn exists_recent(
+    config: &Config,
+    provider: &str,
+    account_id: Option<&str>,
+    title: &str,
+    body: &str,
+) -> Result<bool> {
+    with_connection(config, |conn| {
+        let count: i64 = match account_id {
+            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![provider, aid, title, 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![provider, title, body],
+                |row| row.get(0),
+            ),
+        }
+        .context("[notifications::store] exists_recent query failed")?;
+        Ok(count > 0)
+    })
+}
+
+/// Transition a notification status to 'dismissed'.
+///
+/// Returns `true` when at least one row matched and was updated.
+pub fn mark_dismissed(config: &Config, id: &str) -> Result<bool> {
+    with_connection(config, |conn| {
+        let updated = conn
+            .execute(
+                "UPDATE integration_notifications SET status = 'dismissed' WHERE id = ?1",
+                params![id],
+            )
+            .context("[notification_intel] mark_dismissed failed")?;
+        let matched = updated > 0;
+        if !matched {
+            tracing::warn!(id = %id, "[notification_intel] mark_dismissed matched no rows");
+        } else {
+            tracing::debug!(id = %id, "[notification_intel] mark_dismissed applied");
+        }
+        Ok(matched)
+    })
+}
+
+/// Transition a notification status to 'acted'.
+///
+/// Returns `true` when at least one row matched and was updated.
+pub fn mark_acted(config: &Config, id: &str) -> Result<bool> {
+    with_connection(config, |conn| {
+        let updated = conn
+            .execute(
+                "UPDATE integration_notifications SET status = 'acted' WHERE id = ?1",
+                params![id],
+            )
+            .context("[notification_intel] mark_acted failed")?;
+        let matched = updated > 0;
+        if !matched {
+            tracing::warn!(id = %id, "[notification_intel] mark_acted matched no rows");
+        } else {
+            tracing::debug!(id = %id, "[notification_intel] mark_acted applied");
+        }
+        Ok(matched)
+    })
+}
+
+/// Return aggregate statistics for the notification intelligence pipeline.
+pub fn stats(config: &Config) -> Result<super::types::NotificationStats> {
+    use std::collections::HashMap;
+    with_connection(config, |conn| {
+        let total: i64 = conn
+            .query_row("SELECT COUNT(*) FROM integration_notifications", [], |r| {
+                r.get(0)
+            })
+            .context("[notification_intel] stats total query failed")?;
+
+        let unread: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM integration_notifications WHERE status = 'unread'",
+                [],
+                |r| r.get(0),
+            )
+            .context("[notification_intel] stats unread query failed")?;
+
+        let unscored: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM integration_notifications WHERE importance_score IS NULL",
+                [],
+                |r| r.get(0),
+            )
+            .context("[notification_intel] stats unscored query failed")?;
+
+        // Per-provider counts
+        let mut by_provider = HashMap::new();
+        {
+            let mut stmt = conn
+                .prepare(
+                    "SELECT provider, COUNT(*) FROM integration_notifications GROUP BY provider",
+                )
+                .context("[notification_intel] stats by_provider prepare failed")?;
+            let rows = stmt
+                .query_map([], |row| {
+                    Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
+                })
+                .context("[notification_intel] stats by_provider query failed")?;
+            for row in rows {
+                let (provider, count) =
+                    row.context("[notification_intel] stats by_provider row failed")?;
+                by_provider.insert(provider, count);
+            }
+        }
+
+        // Per-action counts (only where triage_action is set)
+        let mut by_action = HashMap::new();
+        {
+            let mut stmt = conn
+                .prepare(
+                    "SELECT triage_action, COUNT(*) FROM integration_notifications \
+                     WHERE triage_action IS NOT NULL GROUP BY triage_action",
+                )
+                .context("[notification_intel] stats by_action prepare failed")?;
+            let rows = stmt
+                .query_map([], |row| {
+                    Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
+                })
+                .context("[notification_intel] stats by_action query failed")?;
+            for row in rows {
+                let (action, count) =
+                    row.context("[notification_intel] stats by_action row failed")?;
+                by_action.insert(action, count);
+            }
+        }
+
+        tracing::debug!(
+            total = total,
+            unread = unread,
+            unscored = unscored,
+            "[notification_intel] stats query completed"
+        );
+
+        Ok(super::types::NotificationStats {
+            total,
+            unread,
+            unscored,
+            by_provider,
+            by_action,
+        })
+    })
+}
+
 /// Upsert provider-level notification settings.
 pub fn upsert_settings(config: &Config, settings: &NotificationSettings) -> Result<()> {
     with_connection(config, |conn| {
@@ -551,4 +710,76 @@ mod tests {
         assert_eq!(updated.importance_threshold, 0.75);
         assert!(!updated.route_to_orchestrator);
     }
+
+    #[test]
+    fn exists_recent_detects_with_and_without_account_id() {
+        let dir = TempDir::new().unwrap();
+        let config = test_config(&dir);
+
+        let mut n = sample_notification("acct-1", "slack");
+        n.account_id = Some("acct-main".to_string());
+        insert(&config, &n).unwrap();
+
+        assert!(exists_recent(
+            &config,
+            "slack",
+            Some("acct-main"),
+            "Test notification",
+            "Test body"
+        )
+        .unwrap());
+        assert!(!exists_recent(
+            &config,
+            "slack",
+            Some("acct-other"),
+            "Test notification",
+            "Test body"
+        )
+        .unwrap());
+
+        let n_null = sample_notification("acct-null", "slack");
+        insert(&config, &n_null).unwrap();
+        assert!(exists_recent(&config, "slack", None, "Test notification", "Test body").unwrap());
+    }
+
+    #[test]
+    fn mark_dismissed_and_mark_acted_report_match_and_update_status() {
+        let dir = TempDir::new().unwrap();
+        let config = test_config(&dir);
+        insert(&config, &sample_notification("m1", "gmail")).unwrap();
+        insert(&config, &sample_notification("m2", "gmail")).unwrap();
+
+        assert!(mark_dismissed(&config, "m1").unwrap());
+        assert!(mark_acted(&config, "m2").unwrap());
+        assert!(!mark_dismissed(&config, "missing").unwrap());
+        assert!(!mark_acted(&config, "missing").unwrap());
+
+        let items = list(&config, 10, 0, Some("gmail"), None).unwrap();
+        let m1 = items.iter().find(|n| n.id == "m1").unwrap();
+        let m2 = items.iter().find(|n| n.id == "m2").unwrap();
+        assert_eq!(m1.status, NotificationStatus::Dismissed);
+        assert_eq!(m2.status, NotificationStatus::Acted);
+    }
+
+    #[test]
+    fn stats_returns_correct_aggregates() {
+        let dir = TempDir::new().unwrap();
+        let config = test_config(&dir);
+
+        insert(&config, &sample_notification("s1", "gmail")).unwrap();
+        insert(&config, &sample_notification("s2", "gmail")).unwrap();
+        insert(&config, &sample_notification("s3", "slack")).unwrap();
+        update_triage(&config, "s2", 0.9, "escalate", "urgent").unwrap();
+        update_triage(&config, "s3", 0.2, "drop", "noise").unwrap();
+        mark_read(&config, "s2").unwrap();
+
+        let out = stats(&config).unwrap();
+        assert_eq!(out.total, 3);
+        assert_eq!(out.unread, 2);
+        assert_eq!(out.unscored, 1);
+        assert_eq!(out.by_provider.get("gmail"), Some(&2));
+        assert_eq!(out.by_provider.get("slack"), Some(&1));
+        assert_eq!(out.by_action.get("escalate"), Some(&1));
+        assert_eq!(out.by_action.get("drop"), Some(&1));
+    }
 }

src/openhuman/notifications/types.rs

@@ -126,6 +126,16 @@ impl Default for NotificationSettings {
     }
 }
 
+/// Aggregate statistics for the notification intelligence pipeline.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct NotificationStats {
+    pub total: i64,
+    pub unread: i64,
+    pub unscored: i64,
+    pub by_provider: std::collections::HashMap<String, i64>,
+    pub by_action: std::collections::HashMap<String, i64>,
+}
+
 /// Payload for the `notification_ingest` RPC endpoint.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct NotificationIngestRequest {

tests/webview_apis_bridge.rs

@@ -32,6 +32,8 @@ use openhuman_core::openhuman::webview_apis::{client, types::GmailLabel};
 /// server and funnel every test through it.
 static TEST_SERVER: once_cell::sync::Lazy<Mutex<Option<u16>>> =
     once_cell::sync::Lazy::new(|| Mutex::new(None));
+static REQUEST_LOCK: once_cell::sync::Lazy<Mutex<()>> =
+    once_cell::sync::Lazy::new(|| Mutex::new(()));
 
 async fn ensure_mock_server() -> u16 {
     let mut guard = TEST_SERVER.lock().await;
@@ -46,6 +48,7 @@ async fn ensure_mock_server() -> u16 {
     *guard = Some(port);
     tokio::spawn(async move {
         loop {
+            tracing::debug!("[webview_apis_bridge:test] waiting for mock ws client");
             let (stream, _peer) = match listener.accept().await {
                 Ok(v) => v,
                 Err(_) => continue,
@@ -56,12 +59,23 @@ async fn ensure_mock_server() -> u16 {
                     Err(_) => return,
                 };
                 let (mut sink, mut stream) = ws.split();
+                tracing::debug!("[webview_apis_bridge:test] mock ws client connected");
                 while let Some(msg) = stream.next().await {
                     match msg {
                         Ok(Message::Text(text)) => {
                             let req: Value = serde_json::from_str(&text).unwrap();
                             let id = req["id"].as_str().unwrap().to_string();
                             let method = req["method"].as_str().unwrap().to_string();
+                            let redacted_id = if id.len() <= 4 {
+                                "***".to_string()
+                            } else {
+                                format!("***{}", &id[id.len() - 4..])
+                            };
+                            tracing::debug!(
+                                id = %redacted_id,
+                                method = %method,
+                                "[webview_apis_bridge:test] handling text rpc message"
+                            );
                             let resp = match method.as_str() {
                                 "gmail.list_labels" => json!({
                                     "kind": "response",
@@ -86,12 +100,31 @@ async fn ensure_mock_server() -> u16 {
                                 }),
                             };
                             if sink.send(Message::Text(resp.to_string())).await.is_err() {
+                                tracing::warn!(
+                                    id = %redacted_id,
+                                    method = %method,
+                                    "[webview_apis_bridge:test] failed to send mock response"
+                                );
                                 break;
                             }
                         }
-                        Ok(Message::Close(_)) => break,
-                        Ok(_) => continue,
-                        Err(_) => break,
+                        Ok(Message::Close(_)) => {
+                            tracing::debug!("[webview_apis_bridge:test] client closed connection");
+                            break;
+                        }
+                        Ok(_) => {
+                            tracing::debug!(
+                                "[webview_apis_bridge:test] ignoring non-text ws message"
+                            );
+                            continue;
+                        }
+                        Err(err) => {
+                            tracing::warn!(
+                                error = %err,
+                                "[webview_apis_bridge:test] websocket receive error"
+                            );
+                            break;
+                        }
                     }
                 }
             });
@@ -101,7 +134,8 @@ async fn ensure_mock_server() -> u16 {
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn request_round_trips_list_labels_through_mock_server() {
+async fn request_round_trips_and_surfaces_errors_through_mock_server() {
+    let _request_guard = REQUEST_LOCK.lock().await;
     let _port = ensure_mock_server().await;
     let labels: Vec<GmailLabel> = client::request(
         "gmail.list_labels",
@@ -113,11 +147,6 @@ async fn request_round_trips_list_labels_through_mock_server() {
     assert_eq!(labels[0].id, "INBOX");
     assert_eq!(labels[0].unread, Some(3));
     assert_eq!(labels[1].kind, "user");
-}
-
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn request_surfaces_bridge_error_verbatim() {
-    let _port = ensure_mock_server().await;
     let err: Result<Vec<GmailLabel>, String> = client::request(
         "gmail.trash",
         serde_json::from_value(json!({"account_id": "gmail", "message_id": "m1"})).unwrap(),