From 11c67a70b98b5b3cdc0fee2b6ec963a9120ded26 Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Fri, 24 Apr 2026 22:33:51 +0530 Subject: [PATCH] feat(notifications): add dismiss/stats RPC, DomainEvents, and triage bus integration (#875) --- .../lib/webviewNotifications/service.test.ts | 77 ++++-- app/src/lib/webviewNotifications/service.ts | 17 +- app/src/services/notificationService.ts | 55 +++++ src/core/event_bus/events.rs | 41 ++++ src/openhuman/about_app/catalog.rs | 30 +++ src/openhuman/notifications/bus.rs | 63 +++++ src/openhuman/notifications/rpc.rs | 147 +++++++++-- src/openhuman/notifications/schemas.rs | 162 +++++++++++- src/openhuman/notifications/store.rs | 231 ++++++++++++++++++ src/openhuman/notifications/types.rs | 10 + tests/webview_apis_bridge.rs | 47 +++- 11 files changed, 834 insertions(+), 46 deletions(-) diff --git a/app/src/lib/webviewNotifications/service.test.ts b/app/src/lib/webviewNotifications/service.test.ts index 3d1d76a8c..217df11b5 100644 --- a/app/src/lib/webviewNotifications/service.test.ts +++ b/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); + }); }); diff --git a/app/src/lib/webviewNotifications/service.ts b/app/src/lib/webviewNotifications/service.ts index 6acc52f8c..291e0137f 100644 --- a/app/src/lib/webviewNotifications/service.ts +++ b/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, }) .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, status: 'unread', received_at: new Date().toISOString(), importance_score: undefined, diff --git a/app/src/services/notificationService.ts b/app/src/services/notificationService.ts index 3c3a01244..b646925f8 100644 --- a/app/src/services/notificationService.ts +++ b/app/src/services/notificationService.ts @@ -48,6 +48,13 @@ export async function markNotificationRead(id: string): Promise { } type NotificationIngestResult = { id: string; skipped?: false } | { skipped: true; reason: string }; +type NotificationStats = { + total: number; + unread: number; + unscored: number; + by_provider: Record; + by_action: Record; +}; /** * 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 { + 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 { + 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 { + log('fetchNotificationStats'); + try { + const result = await callCoreRpc({ + 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; + } +} diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index fc6aa6fa6..a3687669f 100644 --- a/src/core/event_bus/events.rs +++ b/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, + }, + /// 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 { diff --git a/src/openhuman/about_app/catalog.rs b/src/openhuman/about_app/catalog.rs index 9c5afa203..16c65de09 100644 --- a/src/openhuman/about_app/catalog.rs +++ b/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", diff --git a/src/openhuman/notifications/bus.rs b/src/openhuman/notifications/bus.rs index 9a031fe55..df255ad59 100644 --- a/src/openhuman/notifications/bus.rs +++ b/src/openhuman/notifications/bus.rs @@ -137,6 +137,27 @@ pub fn event_to_notification(event: &DomainEvent) -> Option { + 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()); + } } diff --git a/src/openhuman/notifications/rpc.rs b/src/openhuman/notifications/rpc.rs index db43ad15b..a504574bb 100644 --- a/src/openhuman/notifications/rpc.rs +++ b/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) -> Result { + 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) -> Result 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) -> Result "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) -> Result 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) -> Result) -> Result { + 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) -> Result { + 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) -> Result { + 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 // ───────────────────────────────────────────────────────────────────────────── diff --git a/src/openhuman/notifications/schemas.rs b/src/openhuman/notifications/schemas.rs index 9cdcc783a..e711247c9 100644 --- a/src/openhuman/notifications/schemas.rs +++ b/src/openhuman/notifications/schemas.rs @@ -19,6 +19,9 @@ pub fn all_controller_schemas() -> Vec { 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 { 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) -> ControllerFuture { Box::pin(async move { super::rpc::handle_settings_set(params).await }) } +fn handle_dismiss_wrap(params: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_dismiss(params).await }) +} + +fn handle_mark_acted_wrap(params: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_mark_acted(params).await }) +} + +fn handle_stats_wrap(params: Map) -> 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"); diff --git a/src/openhuman/notifications/store.rs b/src/openhuman/notifications/store.rs index c7cec5952..5abe3b3b2 100644 --- a/src/openhuman/notifications/store.rs +++ b/src/openhuman/notifications/store.rs @@ -298,6 +298,165 @@ pub fn unread_count(config: &Config) -> Result { }) } +/// 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 { + 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 { + 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 { + 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 { + 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)); + } } diff --git a/src/openhuman/notifications/types.rs b/src/openhuman/notifications/types.rs index 1706bd734..4bcadf46a 100644 --- a/src/openhuman/notifications/types.rs +++ b/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, + pub by_action: std::collections::HashMap, +} + /// Payload for the `notification_ingest` RPC endpoint. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NotificationIngestRequest { diff --git a/tests/webview_apis_bridge.rs b/tests/webview_apis_bridge.rs index f06f6bfe5..404cfda54 100644 --- a/tests/webview_apis_bridge.rs +++ b/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>> = once_cell::sync::Lazy::new(|| Mutex::new(None)); +static REQUEST_LOCK: once_cell::sync::Lazy> = + 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 = 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, String> = client::request( "gmail.trash", serde_json::from_value(json!({"account_id": "gmail", "message_id": "m1"})).unwrap(),