From 0a749f5cec1d901a9be122f9346eb33c46f7b361 Mon Sep 17 00:00:00 2001 From: YellowSnnowmann <167776381+YellowSnnowmann@users.noreply.github.com> Date: Sat, 9 May 2026 07:33:31 +0530 Subject: [PATCH] feat(heartbeat): deliver durable proactive meeting/reminder notifications with dedup + category controls (#1369) --- .../settings/panels/NotificationsPanel.tsx | 15 + app/src/lib/notificationRouter.test.ts | 12 + app/src/lib/notificationRouter.ts | 9 + app/src/pages/Notifications.tsx | 3 + .../store/__tests__/notificationSlice.test.ts | 28 ++ app/src/store/__tests__/settingsSlice.test.ts | 10 +- app/src/store/notificationSlice.ts | 40 +- src/openhuman/config/schema/heartbeat_cron.rs | 37 ++ src/openhuman/heartbeat/engine.rs | 30 +- src/openhuman/heartbeat/mod.rs | 2 + src/openhuman/heartbeat/planner/collectors.rs | 338 +++++++++++++ src/openhuman/heartbeat/planner/mod.rs | 443 ++++++++++++++++++ .../heartbeat/planner/persistence.rs | 52 ++ src/openhuman/heartbeat/planner/plan.rs | 109 +++++ src/openhuman/heartbeat/planner/store.rs | 174 +++++++ src/openhuman/heartbeat/planner/types.rs | 75 +++ src/openhuman/heartbeat/planner/utils.rs | 56 +++ src/openhuman/heartbeat/rpc.rs | 130 +++++ src/openhuman/heartbeat/schemas.rs | 157 ++++++- src/openhuman/notifications/types.rs | 3 + 20 files changed, 1715 insertions(+), 8 deletions(-) create mode 100644 src/openhuman/heartbeat/planner/collectors.rs create mode 100644 src/openhuman/heartbeat/planner/mod.rs create mode 100644 src/openhuman/heartbeat/planner/persistence.rs create mode 100644 src/openhuman/heartbeat/planner/plan.rs create mode 100644 src/openhuman/heartbeat/planner/store.rs create mode 100644 src/openhuman/heartbeat/planner/types.rs create mode 100644 src/openhuman/heartbeat/planner/utils.rs create mode 100644 src/openhuman/heartbeat/rpc.rs diff --git a/app/src/components/settings/panels/NotificationsPanel.tsx b/app/src/components/settings/panels/NotificationsPanel.tsx index be663ed24..36bd13046 100644 --- a/app/src/components/settings/panels/NotificationsPanel.tsx +++ b/app/src/components/settings/panels/NotificationsPanel.tsx @@ -23,6 +23,21 @@ const CATEGORIES: { id: NotificationCategory; title: string; description: string title: 'System', description: 'Connection issues, background process errors, updates.', }, + { + id: 'meetings', + title: 'Meetings', + description: 'Upcoming meetings and calendar events detected by heartbeat.', + }, + { + id: 'reminders', + title: 'Reminders', + description: 'Upcoming reminders and scheduled tasks from cron jobs.', + }, + { + id: 'important', + title: 'Important events', + description: 'Urgent or time-sensitive events surfaced from connected sources.', + }, ]; const NotificationsPanel = () => { diff --git a/app/src/lib/notificationRouter.test.ts b/app/src/lib/notificationRouter.test.ts index 32f33fc4b..6092d95a4 100644 --- a/app/src/lib/notificationRouter.test.ts +++ b/app/src/lib/notificationRouter.test.ts @@ -86,6 +86,18 @@ describe('resolveSystemRoute', () => { expect(resolveSystemRoute(makeSystem({ category: 'system' }))).toBe('/home'); }); + it('routes meetings category to /notifications', () => { + expect(resolveSystemRoute(makeSystem({ category: 'meetings' }))).toBe('/notifications'); + }); + + it('routes reminders category to /notifications', () => { + expect(resolveSystemRoute(makeSystem({ category: 'reminders' }))).toBe('/notifications'); + }); + + it('routes important category to /notifications', () => { + expect(resolveSystemRoute(makeSystem({ category: 'important' }))).toBe('/notifications'); + }); + it('prefers deepLink over category default', () => { const item = makeSystem({ category: 'messages', deepLink: '/notifications' }); expect(resolveSystemRoute(item)).toBe('/notifications'); diff --git a/app/src/lib/notificationRouter.ts b/app/src/lib/notificationRouter.ts index 779cd2ed8..9ff8b6543 100644 --- a/app/src/lib/notificationRouter.ts +++ b/app/src/lib/notificationRouter.ts @@ -87,6 +87,15 @@ export function resolveSystemRoute(item: NotificationItem): string { case 'system': log('[notification-router] system id=%s category=system → /home', item.id); return ROUTES.home; + case 'meetings': + log('[notification-router] system id=%s category=meetings → /notifications', item.id); + return ROUTES.notifications; + case 'reminders': + log('[notification-router] system id=%s category=reminders → /notifications', item.id); + return ROUTES.notifications; + case 'important': + log('[notification-router] system id=%s category=important → /notifications', item.id); + return ROUTES.notifications; default: log( '[notification-router] system id=%s category=%s → /notifications (fallback)', diff --git a/app/src/pages/Notifications.tsx b/app/src/pages/Notifications.tsx index f705e0c54..3bf25f792 100644 --- a/app/src/pages/Notifications.tsx +++ b/app/src/pages/Notifications.tsx @@ -18,6 +18,9 @@ const CATEGORY_LABEL: Record = { agents: 'Agents', skills: 'Skills', system: 'System', + meetings: 'Meetings', + reminders: 'Reminders', + important: 'Important', }; function formatTime(ts: number): string { diff --git a/app/src/store/__tests__/notificationSlice.test.ts b/app/src/store/__tests__/notificationSlice.test.ts index 64841ecb5..8b76fee8e 100644 --- a/app/src/store/__tests__/notificationSlice.test.ts +++ b/app/src/store/__tests__/notificationSlice.test.ts @@ -1,3 +1,4 @@ +import { REHYDRATE } from 'redux-persist'; import { describe, expect, it } from 'vitest'; import reducer, { @@ -76,4 +77,31 @@ describe('notificationSlice', () => { expect(s.items).toHaveLength(200); expect(s.items[0].id).toBe('209'); }); + + describe('REHYDRATE', () => { + const rehydrate = (key: string, payload?: unknown) => ({ type: REHYDRATE, key, payload }); + + it('ignores REHYDRATE for a different persist key', () => { + const initial = reducer(undefined, setPreference({ category: 'meetings', enabled: false })); + const s = reducer(initial, rehydrate('other', { preferences: { meetings: true } })); + expect(s.preferences.meetings).toBe(false); + }); + + it('backfills missing preference keys on rehydration for the notifications key', () => { + const s = reducer( + undefined, + rehydrate('notifications', { preferences: { messages: false } }) + ); + expect(s.preferences.messages).toBe(false); + expect(s.preferences.meetings).toBe(true); + expect(s.preferences.reminders).toBe(true); + expect(s.preferences.important).toBe(true); + }); + + it('skips backfill when rehydrated payload has no preferences', () => { + const initial = reducer(undefined, setPreference({ category: 'meetings', enabled: false })); + const s = reducer(initial, rehydrate('notifications', {})); + expect(s.preferences.meetings).toBe(false); + }); + }); }); diff --git a/app/src/store/__tests__/settingsSlice.test.ts b/app/src/store/__tests__/settingsSlice.test.ts index 1abc403a5..6519f36c1 100644 --- a/app/src/store/__tests__/settingsSlice.test.ts +++ b/app/src/store/__tests__/settingsSlice.test.ts @@ -70,7 +70,15 @@ describe('Settings Reducers', () => { read: false, }, ], - preferences: { messages: true, agents: true, skills: true, system: true }, + preferences: { + messages: true, + agents: true, + skills: true, + system: true, + meetings: true, + reminders: true, + important: true, + }, integrationItems: [], integrationUnreadCount: 0, integrationLoading: false, diff --git a/app/src/store/notificationSlice.ts b/app/src/store/notificationSlice.ts index 652f61c39..6886e5912 100644 --- a/app/src/store/notificationSlice.ts +++ b/app/src/store/notificationSlice.ts @@ -1,9 +1,17 @@ import { createSlice, type PayloadAction } from '@reduxjs/toolkit'; +import { REHYDRATE } from 'redux-persist'; import type { IntegrationNotification } from '../types/notifications'; import { resetUserScopedState } from './resetActions'; -export type NotificationCategory = 'messages' | 'agents' | 'skills' | 'system'; +export type NotificationCategory = + | 'messages' + | 'agents' + | 'skills' + | 'system' + | 'meetings' + | 'reminders' + | 'important'; export interface NotificationItem { id: string; @@ -22,6 +30,9 @@ export interface NotificationPreferences { agents: boolean; skills: boolean; system: boolean; + meetings: boolean; + reminders: boolean; + important: boolean; } export interface NotificationState { @@ -37,7 +48,15 @@ const MAX_ITEMS = 200; const initialState: NotificationState = { items: [], - preferences: { messages: true, agents: true, skills: true, system: true }, + preferences: { + messages: true, + agents: true, + skills: true, + system: true, + meetings: true, + reminders: true, + important: true, + }, integrationItems: [], integrationUnreadCount: 0, integrationLoading: false, @@ -134,6 +153,23 @@ const notificationSlice = createSlice({ }, extraReducers: builder => { builder.addCase(resetUserScopedState, () => initialState); + // Backfill any new preference keys that may be absent on older persisted + // state (e.g. meetings/reminders/important added after initial release). + // This ensures state.preferences[item.category] never returns undefined + // for a valid NotificationCategory after rehydration. + builder.addCase(REHYDRATE, (state, action) => { + const rehydrateAction = action as { + type: typeof REHYDRATE; + key: string; + payload?: { preferences?: Partial }; + }; + // Only process the REHYDRATE action that belongs to this slice's persist key. + if (rehydrateAction.key !== 'notifications') return; + const payload = rehydrateAction.payload; + if (payload?.preferences) { + state.preferences = { ...initialState.preferences, ...payload.preferences }; + } + }); }, }); diff --git a/src/openhuman/config/schema/heartbeat_cron.rs b/src/openhuman/config/schema/heartbeat_cron.rs index 1b7bb2dd9..49430e654 100644 --- a/src/openhuman/config/schema/heartbeat_cron.rs +++ b/src/openhuman/config/schema/heartbeat_cron.rs @@ -18,12 +18,43 @@ pub struct HeartbeatConfig { /// Maximum token budget for the situation report (default 40k). #[serde(default = "default_context_budget")] pub context_budget_tokens: u32, + /// Enable proactive notifications for upcoming meetings. + #[serde(default = "default_true")] + pub notify_meetings: bool, + /// Enable proactive notifications for reminders and scheduled items. + #[serde(default = "default_true")] + pub notify_reminders: bool, + /// Enable proactive notifications for urgent/relevant events. + #[serde(default = "default_true")] + pub notify_relevant_events: bool, + /// Allow heartbeat proactive events to also deliver to active external channel. + /// Defaults to false and acts as an explicit consent gate. + #[serde(default)] + pub external_delivery_enabled: bool, + /// Maximum lookahead window for meeting notifications. + #[serde(default = "default_meeting_lookahead_minutes")] + pub meeting_lookahead_minutes: u32, + /// Maximum lookahead window for reminder notifications. + #[serde(default = "default_reminder_lookahead_minutes")] + pub reminder_lookahead_minutes: u32, } fn default_context_budget() -> u32 { 40_000 } +fn default_true() -> bool { + true +} + +fn default_meeting_lookahead_minutes() -> u32 { + 120 +} + +fn default_reminder_lookahead_minutes() -> u32 { + 30 +} + impl Default for HeartbeatConfig { fn default() -> Self { Self { @@ -31,6 +62,12 @@ impl Default for HeartbeatConfig { interval_minutes: 5, inference_enabled: true, context_budget_tokens: default_context_budget(), + notify_meetings: default_true(), + notify_reminders: default_true(), + notify_relevant_events: default_true(), + external_delivery_enabled: false, + meeting_lookahead_minutes: default_meeting_lookahead_minutes(), + reminder_lookahead_minutes: default_reminder_lookahead_minutes(), } } } diff --git a/src/openhuman/heartbeat/engine.rs b/src/openhuman/heartbeat/engine.rs index ce65e5298..77469cca3 100644 --- a/src/openhuman/heartbeat/engine.rs +++ b/src/openhuman/heartbeat/engine.rs @@ -42,7 +42,7 @@ impl HeartbeatEngine { let sleep_secs = u64::from(interval_mins) * 60; loop { - time::sleep(Duration::from_secs(sleep_secs)).await; + self.run_event_planner_tick().await; if self.config.inference_enabled { // Get the shared global engine (same instance as RPC handlers) @@ -86,6 +86,34 @@ impl HeartbeatEngine { } } } + + time::sleep(Duration::from_secs(sleep_secs)).await; + } + } + + async fn run_event_planner_tick(&self) { + match crate::openhuman::config::Config::load_or_init().await { + Ok(config) => { + if !config.heartbeat.enabled { + tracing::debug!("[heartbeat] planner skipped: heartbeat disabled"); + return; + } + let summary = crate::openhuman::heartbeat::planner::evaluate_and_dispatch( + &config, + chrono::Utc::now(), + ) + .await; + tracing::debug!( + source_events = summary.source_events, + deliveries_attempted = summary.deliveries_attempted, + deliveries_sent = summary.deliveries_sent, + deliveries_skipped_dedup = summary.deliveries_skipped_dedup, + "[heartbeat] planner tick summary" + ); + } + Err(error) => { + warn!("[heartbeat] planner skipped: failed to load config: {error}"); + } } } diff --git a/src/openhuman/heartbeat/mod.rs b/src/openhuman/heartbeat/mod.rs index 15e916d52..9909ab3c0 100644 --- a/src/openhuman/heartbeat/mod.rs +++ b/src/openhuman/heartbeat/mod.rs @@ -6,6 +6,8 @@ //! (memory, graph, skills) using the local Ollama model. pub mod engine; +pub mod planner; +pub mod rpc; mod schemas; pub use schemas::{ all_controller_schemas as all_heartbeat_controller_schemas, diff --git a/src/openhuman/heartbeat/planner/collectors.rs b/src/openhuman/heartbeat/planner/collectors.rs new file mode 100644 index 000000000..3797ebd3f --- /dev/null +++ b/src/openhuman/heartbeat/planner/collectors.rs @@ -0,0 +1,338 @@ +use chrono::{DateTime, Duration, Utc}; +use serde_json::json; + +use crate::openhuman::composio::build_composio_client; +use crate::openhuman::config::Config; +use crate::openhuman::cron; +use crate::openhuman::notifications::store as notifications_store; + +use super::types::{HeartbeatCategory, PendingEvent}; +use super::utils::{compute_overlap_key, sanitize_preview, stable_key}; + +pub(crate) fn collect_cron_reminders(config: &Config, now: DateTime) -> Vec { + let lookahead = Duration::minutes(i64::from( + config.heartbeat.reminder_lookahead_minutes.max(1), + )); + + let jobs = match cron::list_jobs(config) { + Ok(jobs) => jobs, + Err(error) => { + tracing::warn!(error = %error, "[heartbeat:planner] cron list_jobs failed"); + return Vec::new(); + } + }; + + jobs.into_iter() + .filter(|job| job.enabled) + .filter(|job| is_reminder_like_job(job)) + .filter(|job| { + let delta = job.next_run.signed_duration_since(now); + delta <= lookahead && delta >= Duration::minutes(-2) + }) + .map(|job| { + let title = job + .name + .clone() + .filter(|name| !name.trim().is_empty()) + .unwrap_or_else(|| "Reminder".to_string()); + let fingerprint = stable_key(&format!("cron:{}:{}", job.id, job.next_run.to_rfc3339())); + let body = format!( + "{} is scheduled at {}.", + title, + job.next_run.format("%H:%M") + ); + + PendingEvent { + category: HeartbeatCategory::Reminders, + source: "cron".to_string(), + source_event_id: job.id, + overlap_key: compute_overlap_key( + HeartbeatCategory::Reminders, + &title, + job.next_run, + ), + fingerprint, + title, + body, + deep_link: Some("/settings/cron-jobs".to_string()), + anchor_at: job.next_run, + } + }) + .collect() +} + +fn is_reminder_like_job(job: &cron::CronJob) -> bool { + if job.delivery.mode.eq_ignore_ascii_case("proactive") { + return true; + } + + let mut haystack = String::new(); + if let Some(name) = &job.name { + haystack.push_str(name); + haystack.push(' '); + } + if let Some(prompt) = &job.prompt { + haystack.push_str(prompt); + haystack.push(' '); + } + haystack.push_str(&job.command); + + let lowered = haystack.to_ascii_lowercase(); + lowered.contains("remind") + || lowered.contains("meeting") + || lowered.contains("standup") + || lowered.contains("follow up") +} + +pub(crate) async fn collect_calendar_meetings( + config: &Config, + now: DateTime, +) -> Vec { + let Some(client) = build_composio_client(config) else { + return Vec::new(); + }; + + let connections = match client.list_connections().await { + Ok(resp) => resp.connections, + Err(error) => { + tracing::warn!(error = %error, "[heartbeat:planner] composio list_connections failed"); + return Vec::new(); + } + }; + + let lookahead = Duration::minutes(i64::from(config.heartbeat.meeting_lookahead_minutes.max(1))); + let end_window = now + lookahead; + + let mut out = Vec::new(); + for conn in connections.into_iter().filter(|c| c.is_active()) { + let toolkit = conn.normalized_toolkit(); + if toolkit != "googlecalendar" && toolkit != "google_calendar" && toolkit != "calendar" { + continue; + } + + let arguments = json!({ + "connectionId": conn.id, + "timeMin": now.to_rfc3339(), + "timeMax": end_window.to_rfc3339(), + "maxResults": 20 + }); + + let resp = match client + .execute_tool("GOOGLECALENDAR_EVENTS_LIST", Some(arguments)) + .await + { + Ok(resp) => resp, + Err(error) => { + tracing::warn!( + toolkit = %toolkit, + connection_id = %conn.id, + error = %error, + "[heartbeat:planner] GOOGLECALENDAR_EVENTS_LIST failed" + ); + continue; + } + }; + + out.extend(extract_calendar_events( + &resp.data, &toolkit, &conn.id, now, end_window, + )); + } + + out +} + +pub(crate) fn extract_calendar_events( + value: &serde_json::Value, + toolkit: &str, + connection_id: &str, + start_window: DateTime, + end_window: DateTime, +) -> Vec { + let mut out = Vec::new(); + collect_calendar_events_recursive( + value, + toolkit, + connection_id, + start_window, + end_window, + &mut out, + ); + out +} + +fn collect_calendar_events_recursive( + value: &serde_json::Value, + toolkit: &str, + connection_id: &str, + start_window: DateTime, + end_window: DateTime, + out: &mut Vec, +) { + match value { + serde_json::Value::Array(items) => { + for item in items { + collect_calendar_events_recursive( + item, + toolkit, + connection_id, + start_window, + end_window, + out, + ); + } + } + serde_json::Value::Object(map) => { + if let Some(starts_at) = extract_datetime_from_map(map) { + if starts_at >= start_window && starts_at <= end_window { + let title = extract_title_from_map(map); + let source_event_id = map + .get("id") + .and_then(serde_json::Value::as_str) + .or_else(|| map.get("eventId").and_then(serde_json::Value::as_str)) + .or_else(|| map.get("icalUID").and_then(serde_json::Value::as_str)) + .unwrap_or("calendar-event") + .to_string(); + let deep_link = map + .get("htmlLink") + .and_then(serde_json::Value::as_str) + .or_else(|| map.get("hangoutLink").and_then(serde_json::Value::as_str)) + .map(ToString::to_string); + + let fingerprint = stable_key(&format!( + "{}:{}:{}:{}", + toolkit, + connection_id, + source_event_id, + starts_at.to_rfc3339() + )); + + out.push(PendingEvent { + category: HeartbeatCategory::Meetings, + source: format!("calendar:{toolkit}"), + source_event_id, + overlap_key: compute_overlap_key( + HeartbeatCategory::Meetings, + &title, + starts_at, + ), + fingerprint, + title: title.clone(), + body: format!("{} starts at {}.", title, starts_at.format("%H:%M")), + deep_link, + anchor_at: starts_at, + }); + } + } + + for child in map.values() { + collect_calendar_events_recursive( + child, + toolkit, + connection_id, + start_window, + end_window, + out, + ); + } + } + _ => {} + } +} + +fn extract_datetime_from_map( + map: &serde_json::Map, +) -> Option> { + // Only accept `start.dateTime` — never fall back to `start.date`. + // All-day events (birthdays, OOO, holidays) only have a `start.date` field + // and must not be surfaced as timed meetings. + let start = map.get("start").and_then(|start| match start { + serde_json::Value::Object(start_map) => start_map + .get("dateTime") + .and_then(serde_json::Value::as_str), + serde_json::Value::String(s) => Some(s.as_str()), + _ => None, + }); + + let direct = start + .or_else(|| map.get("start_time").and_then(serde_json::Value::as_str)) + .or_else(|| map.get("startTime").and_then(serde_json::Value::as_str)) + .or_else(|| map.get("starts_at").and_then(serde_json::Value::as_str)) + .or_else(|| map.get("startsAt").and_then(serde_json::Value::as_str)); + + direct.and_then(parse_datetime) +} + +fn extract_title_from_map(map: &serde_json::Map) -> String { + map.get("summary") + .and_then(serde_json::Value::as_str) + .or_else(|| map.get("title").and_then(serde_json::Value::as_str)) + .or_else(|| map.get("name").and_then(serde_json::Value::as_str)) + .map(|raw| sanitize_preview(raw, 80)) + .filter(|title| !title.is_empty()) + .unwrap_or_else(|| "Upcoming meeting".to_string()) +} + +fn parse_datetime(raw: &str) -> Option> { + chrono::DateTime::parse_from_rfc3339(raw) + .map(|dt| dt.with_timezone(&Utc)) + .ok() +} + +pub(crate) fn collect_relevant_notifications( + config: &Config, + now: DateTime, +) -> Vec { + // Do not apply an importance_score threshold here — urgent and action-worthy + // notifications may have a low or absent score. The downstream triage_action + // and raw_payload.urgent checks are the real gate. + let items = match notifications_store::list(config, 100, 0, None, None) { + Ok(items) => items, + Err(error) => { + tracing::warn!(error = %error, "[heartbeat:planner] notifications list failed"); + return Vec::new(); + } + }; + + items + .into_iter() + // Never re-escalate notifications we generated ourselves — that creates a + // feedback loop where each heartbeat tick spawns a new "Important event" + // with a fresh ID that bypasses the dedupe store. + .filter(|item| item.provider != "heartbeat") + .filter(|item| { + item.status == crate::openhuman::notifications::types::NotificationStatus::Unread + }) + .filter(|item| { + item.triage_action + .as_deref() + .map(|action| action == "escalate" || action == "react") + .unwrap_or(false) + || item + .raw_payload + .get("urgent") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false) + }) + .filter(|item| now.signed_duration_since(item.received_at) <= Duration::minutes(30)) + .map(|item| { + let title = format!("Important event from {}", item.provider); + let body = sanitize_preview(&item.title, 100); + + PendingEvent { + category: HeartbeatCategory::Important, + source: format!("notification:{}", item.provider), + source_event_id: item.id.clone(), + overlap_key: compute_overlap_key( + HeartbeatCategory::Important, + &title, + item.received_at, + ), + fingerprint: stable_key(&format!("notification:{}", item.id)), + title, + body, + deep_link: Some("/notifications".to_string()), + anchor_at: item.received_at, + } + }) + .collect() +} diff --git a/src/openhuman/heartbeat/planner/mod.rs b/src/openhuman/heartbeat/planner/mod.rs new file mode 100644 index 000000000..4eade2fd8 --- /dev/null +++ b/src/openhuman/heartbeat/planner/mod.rs @@ -0,0 +1,443 @@ +//! Heartbeat planner — evaluates upcoming events and dispatches proactive +//! notifications. +//! +//! # Module layout +//! +//! | File | Responsibility | +//! |------|----------------| +//! | `types.rs` | Shared data types (`HeartbeatCategory`, `PendingEvent`, …) | +//! | `collectors.rs` | Source-specific collectors (cron, calendar, notifications) | +//! | `plan.rs` | Delivery-window logic (`plan_delivery_for_event`) | +//! | `persistence.rs` | Durable notification persistence (`persist_heartbeat_alert`) | +//! | `utils.rs` | Pure helpers (`sanitize_preview`, `stable_key`) | +//! | `store.rs` | Dedupe store (`mark_sent`, `prune_old`) | + +mod collectors; +mod persistence; +mod plan; +mod store; +mod types; +mod utils; + +pub use types::PlannerRunSummary; + +use std::collections::HashSet; + +use chrono::{DateTime, Duration, Utc}; + +use crate::core::event_bus::{publish_global, DomainEvent}; +use crate::openhuman::config::Config; +use crate::openhuman::notifications::bus::publish_core_notification; +use crate::openhuman::notifications::types::CoreNotificationEvent; + +use collectors::{ + collect_calendar_meetings, collect_cron_reminders, collect_relevant_notifications, +}; +use persistence::persist_heartbeat_alert; +use plan::plan_delivery_for_event; +use utils::stable_key; + +/// Evaluate all configured notification categories and dispatch any events that +/// fall within their delivery windows and have not already been sent. +pub async fn evaluate_and_dispatch(config: &Config, now: DateTime) -> PlannerRunSummary { + let mut summary = PlannerRunSummary::empty(); + + if !(config.heartbeat.notify_meetings + || config.heartbeat.notify_reminders + || config.heartbeat.notify_relevant_events) + { + tracing::debug!("[heartbeat:planner] all categories disabled; skipping tick"); + return summary; + } + + let mut events = Vec::new(); + + if config.heartbeat.notify_reminders { + events.extend(collect_cron_reminders(config, now)); + } + + if config.heartbeat.notify_meetings { + events.extend(collect_calendar_meetings(config, now).await); + } + + if config.heartbeat.notify_relevant_events { + events.extend(collect_relevant_notifications(config, now)); + } + + summary.source_events = events.len(); + + let mut seen_keys: HashSet = HashSet::new(); + + for event in events { + let Some(plan) = plan_delivery_for_event(&event, config, now) else { + continue; + }; + + // Use `overlap_key` (content-based: category + title + time-bucket) so + // that identical underlying events surfaced by multiple sources + // (e.g. the same meeting visible in both cron reminders and a calendar + // connection) map to the same dedupe key and only one notification is + // delivered. + let dedupe_key = stable_key(&format!( + "{}|{}|{}", + event.category.as_str(), + event.overlap_key, + plan.stage + )); + + // Overlapping sources in the same tick should still dedupe before hitting disk. + if !seen_keys.insert(dedupe_key.clone()) { + summary.deliveries_skipped_dedup += 1; + continue; + } + + summary.deliveries_attempted += 1; + + let id = format!( + "heartbeat:{}:{}:{}", + event.category.as_str(), + plan.stage, + &dedupe_key[..12] + ); + + // Persist the durable notification record BEFORE marking dedupe, so a + // failed write doesn't permanently suppress future retries. + if let Err(error) = persist_heartbeat_alert(config, &event, &plan, now) { + tracing::warn!( + dedupe_key = %dedupe_key, + source = %event.source, + source_event_id = %event.source_event_id, + category = event.category.as_str(), + stage = plan.stage, + error = %error, + "[heartbeat:planner] failed to persist heartbeat alert; skipping delivery" + ); + continue; + } + + let inserted = match store::mark_sent( + config, + &store::SentMarker { + dedupe_key: &dedupe_key, + event_fingerprint: &event.fingerprint, + source: &event.source, + category: event.category.as_str(), + stage: plan.stage, + sent_at: now, + }, + ) { + Ok(v) => v, + Err(error) => { + tracing::warn!( + dedupe_key = %dedupe_key, + source = %event.source, + source_event_id = %event.source_event_id, + category = event.category.as_str(), + error = %error, + "[heartbeat:planner] failed to persist dedupe marker" + ); + continue; + } + }; + + if !inserted { + summary.deliveries_skipped_dedup += 1; + continue; + } + + publish_core_notification(CoreNotificationEvent { + id, + category: event.category.notification_category(), + title: plan.title, + body: plan.body, + deep_link: event.deep_link.clone(), + timestamp_ms: now.timestamp_millis().max(0) as u64, + }); + + if config.heartbeat.external_delivery_enabled && plan.allow_external { + publish_global(DomainEvent::ProactiveMessageRequested { + source: format!("heartbeat:{}", event.category.as_str()), + message: plan.proactive_message, + job_name: Some(format!("heartbeat-{}", event.category.as_str())), + }); + } + + summary.deliveries_sent += 1; + + tracing::debug!( + dedupe_key = %dedupe_key, + source = %event.source, + source_event_id = %event.source_event_id, + category = event.category.as_str(), + stage = plan.stage, + "[heartbeat:planner] delivery sent" + ); + } + + if let Err(error) = store::prune_old(config, now - Duration::days(14)) { + tracing::warn!(error = %error, "[heartbeat:planner] prune_old failed"); + } + + summary +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::config::Config; + use crate::openhuman::cron::{self, Schedule}; + use crate::openhuman::notifications::subscribe_core_notifications; + use chrono::TimeZone; + use serde_json::json; + use tempfile::TempDir; + + use collectors::extract_calendar_events; + use plan::plan_delivery_for_event; + use types::{HeartbeatCategory, PendingEvent}; + use utils::{compute_overlap_key, sanitize_preview}; + + #[test] + fn extract_calendar_events_reads_nested_payload() { + let now = Utc.with_ymd_and_hms(2026, 5, 8, 10, 0, 0).unwrap(); + let payload = json!({ + "items": [ + { + "id": "evt-1", + "summary": "Team sync", + "start": { "dateTime": "2026-05-08T10:20:00Z" }, + "htmlLink": "https://calendar.google.com/event?evt=1" + } + ] + }); + + let events = extract_calendar_events( + &payload, + "googlecalendar", + "conn-1", + now, + now + Duration::minutes(60), + ); + + assert_eq!(events.len(), 1); + assert_eq!(events[0].category, HeartbeatCategory::Meetings); + assert_eq!(events[0].source_event_id, "evt-1"); + assert_eq!(events[0].title, "Team sync"); + assert_eq!( + events[0].deep_link.as_deref(), + Some("https://calendar.google.com/event?evt=1") + ); + } + + #[test] + fn all_day_calendar_events_are_skipped() { + let now = Utc.with_ymd_and_hms(2026, 5, 8, 0, 0, 0).unwrap(); + let payload = json!({ + "items": [ + { + "id": "all-day-1", + "summary": "Birthday", + "start": { "date": "2026-05-08" } + } + ] + }); + + let events = extract_calendar_events( + &payload, + "googlecalendar", + "conn-1", + now, + now + Duration::hours(24), + ); + + assert_eq!( + events.len(), + 0, + "all-day events should not be promoted to meetings" + ); + } + + #[test] + fn reminder_stage_prioritizes_due_window() { + let mut config = Config::default(); + config.heartbeat.reminder_lookahead_minutes = 15; + let now = Utc.with_ymd_and_hms(2026, 5, 8, 10, 0, 0).unwrap(); + let event = PendingEvent { + category: HeartbeatCategory::Reminders, + source: "cron".to_string(), + source_event_id: "job-1".to_string(), + fingerprint: "fp-1".to_string(), + overlap_key: compute_overlap_key(HeartbeatCategory::Reminders, "Pay rent", now), + title: "Pay rent".to_string(), + body: String::new(), + deep_link: None, + anchor_at: now, + }; + + let plan = plan_delivery_for_event(&event, &config, now).expect("plan"); + assert_eq!(plan.stage, "due"); + assert!(plan.allow_external); + } + + #[test] + fn meeting_stage_uses_heads_up_for_longer_lead() { + let mut config = Config::default(); + config.heartbeat.meeting_lookahead_minutes = 120; + let now = Utc.with_ymd_and_hms(2026, 5, 8, 10, 0, 0).unwrap(); + let event = PendingEvent { + category: HeartbeatCategory::Meetings, + source: "calendar:googlecalendar".to_string(), + source_event_id: "evt-1".to_string(), + fingerprint: "fp-1".to_string(), + overlap_key: compute_overlap_key( + HeartbeatCategory::Meetings, + "Planning", + now + Duration::minutes(45), + ), + title: "Planning".to_string(), + body: String::new(), + deep_link: None, + anchor_at: now + Duration::minutes(45), + }; + + let plan = plan_delivery_for_event(&event, &config, now).expect("plan"); + assert_eq!(plan.stage, "heads_up"); + assert!(!plan.allow_external); + } + + #[test] + fn sanitize_preview_trims_and_normalizes_whitespace() { + let out = sanitize_preview(" hello world ", 30); + assert_eq!(out, "hello world"); + + let out = sanitize_preview("a very long sentence with many words", 10); + assert!(out.ends_with('…')); + assert!(out.chars().count() <= 10); + } + + fn test_config(tmp: &TempDir) -> Config { + Config { + workspace_dir: tmp.path().to_path_buf(), + config_path: tmp.path().join("config.toml"), + ..Config::default() + } + } + + #[tokio::test] + async fn evaluate_and_dispatch_dedupes_across_ticks() { + let tmp = TempDir::new().unwrap(); + let mut config = test_config(&tmp); + config.heartbeat.notify_meetings = false; + config.heartbeat.notify_relevant_events = false; + config.heartbeat.notify_reminders = true; + config.heartbeat.reminder_lookahead_minutes = 30; + + let now = Utc::now(); + let run_at = now + Duration::minutes(5); + let schedule = Schedule::At { at: run_at }; + let _job = cron::add_shell_job(&config, Some("remind_me".to_string()), schedule, "echo hi") + .expect("create cron reminder"); + + let mut rx = subscribe_core_notifications(); + while rx.try_recv().is_ok() {} + + let first = evaluate_and_dispatch(&config, now).await; + assert_eq!(first.deliveries_sent, 1); + + let second = evaluate_and_dispatch(&config, now).await; + assert_eq!(second.deliveries_sent, 0); + assert!(second.deliveries_skipped_dedup >= 1); + } + + #[tokio::test] + async fn heartbeat_provider_notifications_are_not_re_escalated() { + use crate::openhuman::notifications::store as notifications_store; + use crate::openhuman::notifications::types::{IntegrationNotification, NotificationStatus}; + + let tmp = TempDir::new().unwrap(); + let mut config = test_config(&tmp); + config.heartbeat.notify_meetings = false; + config.heartbeat.notify_reminders = false; + config.heartbeat.notify_relevant_events = true; + + let now = Utc::now(); + + // Simulate a previously-persisted heartbeat notification (triage_action="react", + // status=Unread, importance_score=0.9) — exactly what persist_heartbeat_alert writes. + let hb_notification = IntegrationNotification { + id: "heartbeat:meetings:final_call:abc123def456".to_string(), + provider: "heartbeat".to_string(), + account_id: None, + title: "Upcoming meeting: Team sync".to_string(), + body: "Starts in about 5 minutes.".to_string(), + raw_payload: serde_json::json!({"category": "meetings", "stage": "final_call"}), + importance_score: Some(0.9), + triage_action: Some("react".to_string()), + triage_reason: Some("heartbeat proactive event".to_string()), + status: NotificationStatus::Unread, + received_at: now, + scored_at: Some(now), + }; + notifications_store::insert_if_not_recent(&config, &hb_notification).unwrap(); + + // Planner must NOT re-escalate notifications it generated itself. + let summary = evaluate_and_dispatch(&config, now).await; + assert_eq!( + summary.deliveries_sent, 0, + "heartbeat provider notifications must not be re-escalated as Important events" + ); + } + + #[test] + fn overlap_key_same_for_cross_source_same_event() { + // Two different sources that surface the same meeting at the same time + // (within the 15-minute bucket) must produce the same overlap_key so + // only one notification is dispatched. + let anchor = Utc.with_ymd_and_hms(2026, 5, 8, 10, 0, 0).unwrap(); + + let key_from_calendar = + compute_overlap_key(HeartbeatCategory::Meetings, "Team Standup", anchor); + // A cron job with the same title and an anchor 2 minutes later (same + // 15-minute bucket) — different source, same underlying event. + let key_from_cron = compute_overlap_key( + HeartbeatCategory::Meetings, + "Team Standup", + anchor + Duration::minutes(2), + ); + + assert_eq!( + key_from_calendar, key_from_cron, + "cross-source events in the same 15-min bucket must share an overlap_key" + ); + } + + #[test] + fn overlap_key_differs_for_different_titles_or_times() { + let anchor = Utc.with_ymd_and_hms(2026, 5, 8, 10, 0, 0).unwrap(); + + // Different title → different key. + let key_a = compute_overlap_key(HeartbeatCategory::Meetings, "Team Standup", anchor); + let key_b = compute_overlap_key(HeartbeatCategory::Meetings, "1:1 With Manager", anchor); + assert_ne!( + key_a, key_b, + "different titles must produce different overlap keys" + ); + + // Same title but more than one bucket apart (>= 15 min) → different key. + let key_c = compute_overlap_key( + HeartbeatCategory::Meetings, + "Team Standup", + anchor + Duration::minutes(20), + ); + assert_ne!( + key_a, key_c, + "events in different time buckets must produce different overlap keys" + ); + + // Different category → different key even with same title and time. + let key_d = compute_overlap_key(HeartbeatCategory::Reminders, "Team Standup", anchor); + assert_ne!( + key_a, key_d, + "different categories must produce different overlap keys" + ); + } +} diff --git a/src/openhuman/heartbeat/planner/persistence.rs b/src/openhuman/heartbeat/planner/persistence.rs new file mode 100644 index 000000000..62acffc8c --- /dev/null +++ b/src/openhuman/heartbeat/planner/persistence.rs @@ -0,0 +1,52 @@ +use chrono::{DateTime, Utc}; + +use crate::openhuman::config::Config; +use crate::openhuman::notifications::store as notifications_store; +use crate::openhuman::notifications::types::{IntegrationNotification, NotificationStatus}; + +use super::types::{HeartbeatCategory, PendingEvent, PlannedDelivery}; +use super::utils::sanitize_preview; + +/// Durably persist a heartbeat alert into the notifications store. +/// +/// Returns an error if the store write fails. The caller should refrain from +/// marking the dedupe key until this returns `Ok`, so that a failed write does +/// not permanently suppress future retries. +pub(crate) fn persist_heartbeat_alert( + config: &Config, + event: &PendingEvent, + plan: &PlannedDelivery, + now: DateTime, +) -> anyhow::Result<()> { + let notification = IntegrationNotification { + id: format!( + "heartbeat:{}:{}:{}", + event.category.as_str(), + plan.stage, + &event.fingerprint[..12] + ), + provider: "heartbeat".to_string(), + account_id: Some(event.source_event_id.clone()), + title: sanitize_preview(&plan.title, 100), + body: sanitize_preview(&plan.body, 180), + raw_payload: serde_json::json!({ + "source": event.source, + "category": event.category.as_str(), + "stage": plan.stage, + "anchor_at": event.anchor_at.to_rfc3339(), + "deep_link": event.deep_link.clone(), + }), + importance_score: Some(match event.category { + HeartbeatCategory::Meetings => 0.8, + HeartbeatCategory::Reminders => 0.7, + HeartbeatCategory::Important => 0.9, + }), + triage_action: Some("react".to_string()), + triage_reason: Some("heartbeat proactive event".to_string()), + status: NotificationStatus::Unread, + received_at: now, + scored_at: Some(now), + }; + + notifications_store::insert_if_not_recent(config, ¬ification).map(|_| ()) +} diff --git a/src/openhuman/heartbeat/planner/plan.rs b/src/openhuman/heartbeat/planner/plan.rs new file mode 100644 index 000000000..21eda058c --- /dev/null +++ b/src/openhuman/heartbeat/planner/plan.rs @@ -0,0 +1,109 @@ +use chrono::{DateTime, Duration, Utc}; + +use crate::openhuman::config::Config; + +use super::types::{HeartbeatCategory, PendingEvent, PlannedDelivery}; + +/// Choose the correct notification stage and message text for `event` given +/// the current time and user config. Returns `None` when the event is outside +/// all delivery windows and should be skipped. +pub(crate) fn plan_delivery_for_event( + event: &PendingEvent, + config: &Config, + now: DateTime, +) -> Option { + let until = event.anchor_at.signed_duration_since(now); + let until_minutes = until.num_minutes(); + + match event.category { + HeartbeatCategory::Meetings => { + let lookahead = i64::from(config.heartbeat.meeting_lookahead_minutes.max(1)); + if until_minutes > 10 && until_minutes <= lookahead { + let mins = until_minutes.max(1); + return Some(PlannedDelivery { + stage: "heads_up", + title: format!("Meeting soon: {}", event.title), + body: format!("Starts in about {mins} minutes."), + proactive_message: format!( + "You have a meeting coming up in about {mins} minutes: {}.", + event.title + ), + allow_external: false, + }); + } + if until_minutes > 0 && until_minutes <= 10 { + let mins = until_minutes.max(1); + return Some(PlannedDelivery { + stage: "final_call", + title: format!("Upcoming meeting: {}", event.title), + body: format!("Starts in about {mins} minutes."), + proactive_message: format!( + "Your meeting starts in about {mins} minutes: {}.", + event.title + ), + allow_external: true, + }); + } + // Wider grace window: heartbeat runs every few minutes, so + // tiny post-start windows can miss real meetings. + if until_minutes <= 0 && until_minutes >= -10 { + return Some(PlannedDelivery { + stage: "starting_now", + title: format!("Meeting starting now: {}", event.title), + body: "This meeting should be starting now.".to_string(), + proactive_message: format!("Your meeting is starting now: {}.", event.title), + allow_external: true, + }); + } + None + } + HeartbeatCategory::Reminders => { + let lookahead = i64::from(config.heartbeat.reminder_lookahead_minutes.max(1)); + if until_minutes > 0 && until_minutes <= lookahead { + let mins = until_minutes.max(1); + return Some(PlannedDelivery { + stage: "soon", + title: format!("Reminder soon: {}", event.title), + body: format!("Scheduled in about {mins} minutes."), + proactive_message: format!( + "Reminder coming up in about {mins} minutes: {}.", + event.title + ), + allow_external: false, + }); + } + // Wider grace window for reminder due state to prevent misses + // from tick alignment. + if until_minutes <= 0 && until_minutes >= -10 { + return Some(PlannedDelivery { + stage: "due", + title: format!("Reminder due: {}", event.title), + body: "A scheduled reminder is due now.".to_string(), + proactive_message: format!("Reminder due now: {}.", event.title), + allow_external: true, + }); + } + None + } + HeartbeatCategory::Important => { + if now.signed_duration_since(event.anchor_at) <= Duration::minutes(10) { + return Some(PlannedDelivery { + stage: "important_now", + title: event.title.clone(), + body: if event.body.is_empty() { + "A time-sensitive event needs your attention.".to_string() + } else { + event.body.clone() + }, + proactive_message: if event.body.is_empty() { + "A time-sensitive event needs your attention.".to_string() + } else { + event.body.clone() + }, + allow_external: true, + }); + } + None + } + } +} diff --git a/src/openhuman/heartbeat/planner/store.rs b/src/openhuman/heartbeat/planner/store.rs new file mode 100644 index 000000000..8cedbbec9 --- /dev/null +++ b/src/openhuman/heartbeat/planner/store.rs @@ -0,0 +1,174 @@ +use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; +use rusqlite::{params, Connection}; + +use crate::openhuman::config::Config; + +const SCHEMA: &str = " +CREATE TABLE IF NOT EXISTS heartbeat_notification_state ( + dedupe_key TEXT PRIMARY KEY, + event_fingerprint TEXT NOT NULL, + source TEXT NOT NULL, + category TEXT NOT NULL, + stage TEXT NOT NULL, + sent_at TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_heartbeat_notification_state_sent_at + ON heartbeat_notification_state(sent_at); +"; + +pub struct SentMarker<'a> { + pub dedupe_key: &'a str, + pub event_fingerprint: &'a str, + pub source: &'a str, + pub category: &'a str, + pub stage: &'a str, + pub sent_at: DateTime, +} + +fn with_connection(config: &Config, f: impl FnOnce(&Connection) -> Result) -> Result { + let db_path = config + .workspace_dir + .join("heartbeat") + .join("heartbeat_state.db"); + + if let Some(parent) = db_path.parent() { + std::fs::create_dir_all(parent).with_context(|| { + format!( + "[heartbeat::store] failed to create DB dir {}", + parent.display() + ) + })?; + } + + let conn = Connection::open(&db_path).with_context(|| { + format!( + "[heartbeat::store] failed to open DB at {}", + db_path.display() + ) + })?; + conn.execute_batch(SCHEMA) + .context("[heartbeat::store] schema migration failed")?; + + f(&conn) +} + +pub fn mark_sent(config: &Config, marker: &SentMarker<'_>) -> Result { + with_connection(config, |conn| { + let changed = conn + .execute( + "INSERT OR IGNORE INTO heartbeat_notification_state + (dedupe_key, event_fingerprint, source, category, stage, sent_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + marker.dedupe_key, + marker.event_fingerprint, + marker.source, + marker.category, + marker.stage, + marker.sent_at.to_rfc3339(), + ], + ) + .context("[heartbeat::store] mark_sent insert failed")?; + + Ok(changed > 0) + }) +} + +pub fn prune_old(config: &Config, cutoff: DateTime) -> Result { + with_connection(config, |conn| { + let changed = conn + .execute( + "DELETE FROM heartbeat_notification_state WHERE sent_at < ?1", + params![cutoff.to_rfc3339()], + ) + .context("[heartbeat::store] prune_old delete failed")?; + Ok(changed) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn test_config(tmp: &TempDir) -> Config { + Config { + workspace_dir: tmp.path().to_path_buf(), + config_path: tmp.path().join("config.toml"), + ..Config::default() + } + } + + #[test] + fn mark_sent_dedupes_by_key() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let now = Utc::now(); + + let first = mark_sent( + &config, + &SentMarker { + dedupe_key: "a", + event_fingerprint: "fp", + source: "cron", + category: "reminders", + stage: "due", + sent_at: now, + }, + ) + .unwrap(); + + let second = mark_sent( + &config, + &SentMarker { + dedupe_key: "a", + event_fingerprint: "fp", + source: "cron", + category: "reminders", + stage: "due", + sent_at: now, + }, + ) + .unwrap(); + + assert!(first); + assert!(!second); + } + + #[test] + fn prune_old_removes_outdated_rows() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let now = Utc::now(); + + mark_sent( + &config, + &SentMarker { + dedupe_key: "old", + event_fingerprint: "fp-old", + source: "cron", + category: "reminders", + stage: "due", + sent_at: now - chrono::Duration::days(30), + }, + ) + .unwrap(); + + mark_sent( + &config, + &SentMarker { + dedupe_key: "new", + event_fingerprint: "fp-new", + source: "cron", + category: "reminders", + stage: "due", + sent_at: now, + }, + ) + .unwrap(); + + let removed = prune_old(&config, now - chrono::Duration::days(14)).unwrap(); + assert_eq!(removed, 1); + } +} diff --git a/src/openhuman/heartbeat/planner/types.rs b/src/openhuman/heartbeat/planner/types.rs new file mode 100644 index 000000000..aa8c8d402 --- /dev/null +++ b/src/openhuman/heartbeat/planner/types.rs @@ -0,0 +1,75 @@ +use chrono::{DateTime, Utc}; +use serde::Serialize; + +use crate::openhuman::notifications::types::CoreNotificationCategory; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum HeartbeatCategory { + Meetings, + Reminders, + Important, +} + +impl HeartbeatCategory { + pub(crate) fn as_str(&self) -> &'static str { + match self { + Self::Meetings => "meetings", + Self::Reminders => "reminders", + Self::Important => "important", + } + } + + pub(crate) fn notification_category(&self) -> CoreNotificationCategory { + match self { + Self::Meetings => CoreNotificationCategory::Meetings, + Self::Reminders => CoreNotificationCategory::Reminders, + Self::Important => CoreNotificationCategory::Important, + } + } +} + +#[derive(Debug, Clone)] +pub(crate) struct PendingEvent { + pub category: HeartbeatCategory, + pub source: String, + pub source_event_id: String, + /// Source-specific fingerprint — unique within a single source. + pub fingerprint: String, + /// Content-based overlap key — identical events from different sources + /// (e.g. the same meeting appearing in both a cron job and a calendar + /// connection) hash to the same value and are deduplicated across sources. + /// Derived from `category + normalized_title + time_bucket`. + pub overlap_key: String, + pub title: String, + pub body: String, + pub deep_link: Option, + pub anchor_at: DateTime, +} + +#[derive(Debug, Clone)] +pub(crate) struct PlannedDelivery { + pub stage: &'static str, + pub title: String, + pub body: String, + pub proactive_message: String, + pub allow_external: bool, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PlannerRunSummary { + pub source_events: usize, + pub deliveries_attempted: usize, + pub deliveries_sent: usize, + pub deliveries_skipped_dedup: usize, +} + +impl PlannerRunSummary { + pub(crate) fn empty() -> Self { + Self { + source_events: 0, + deliveries_attempted: 0, + deliveries_sent: 0, + deliveries_skipped_dedup: 0, + } + } +} diff --git a/src/openhuman/heartbeat/planner/utils.rs b/src/openhuman/heartbeat/planner/utils.rs new file mode 100644 index 000000000..9821995ec --- /dev/null +++ b/src/openhuman/heartbeat/planner/utils.rs @@ -0,0 +1,56 @@ +use chrono::{DateTime, Utc}; +use sha2::{Digest, Sha256}; + +use super::types::HeartbeatCategory; + +/// Truncate `raw` to at most `max_chars` characters, normalizing internal +/// whitespace and appending '…' if truncated. +pub(crate) fn sanitize_preview(raw: &str, max_chars: usize) -> String { + let clean = raw.split_whitespace().collect::>().join(" "); + if clean.chars().count() <= max_chars { + return clean; + } + let mut trimmed: String = clean.chars().take(max_chars.saturating_sub(1)).collect(); + trimmed.push('…'); + trimmed +} + +/// Return a stable hex-encoded SHA-256 of `seed`. +pub(crate) fn stable_key(seed: &str) -> String { + let mut hasher = Sha256::new(); + hasher.update(seed.as_bytes()); + hex::encode(hasher.finalize()) +} + +/// Compute an overlap key for cross-source deduplication. +/// +/// Events from different sources (e.g. a cron reminder and a calendar event) +/// representing the same underlying occurrence should produce the same overlap +/// key so that only one notification is dispatched regardless of which source +/// surfaces it first. +/// +/// The key is derived from: +/// - `category` — so meetings, reminders, and important events never collide. +/// - `normalized_title` — lowercased, whitespace-normalized title. +/// - `time_bucket` — `anchor_at` rounded down to the nearest 15-minute slot, +/// giving a small window of tolerance for sources that report slightly +/// different start times for the same event. +pub(crate) fn compute_overlap_key( + category: HeartbeatCategory, + title: &str, + anchor_at: DateTime, +) -> String { + let normalized_title = title.to_ascii_lowercase(); + let normalized_title = normalized_title + .split_whitespace() + .collect::>() + .join(" "); + // Round down to nearest 15-minute bucket to tolerate minor time skew across sources. + let bucket_minutes = (anchor_at.timestamp() / 60) / 15 * 15; + stable_key(&format!( + "{}|{}|{}", + category.as_str(), + normalized_title, + bucket_minutes + )) +} diff --git a/src/openhuman/heartbeat/rpc.rs b/src/openhuman/heartbeat/rpc.rs new file mode 100644 index 000000000..160f4e660 --- /dev/null +++ b/src/openhuman/heartbeat/rpc.rs @@ -0,0 +1,130 @@ +use chrono::Utc; +use serde::{Deserialize, Serialize}; +use serde_json::json; +use tracing::{debug, warn}; + +use crate::openhuman::config::{self, Config}; +use crate::rpc::RpcOutcome; + +use super::planner; + +#[derive(Debug, Clone, Deserialize)] +pub struct HeartbeatSettingsPatch { + pub enabled: Option, + pub interval_minutes: Option, + pub inference_enabled: Option, + pub notify_meetings: Option, + pub notify_reminders: Option, + pub notify_relevant_events: Option, + pub external_delivery_enabled: Option, + pub meeting_lookahead_minutes: Option, + pub reminder_lookahead_minutes: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct HeartbeatSettingsView { + pub enabled: bool, + pub interval_minutes: u32, + pub inference_enabled: bool, + pub notify_meetings: bool, + pub notify_reminders: bool, + pub notify_relevant_events: bool, + pub external_delivery_enabled: bool, + pub meeting_lookahead_minutes: u32, + pub reminder_lookahead_minutes: u32, +} + +pub async fn settings_get() -> Result, String> { + debug!("[heartbeat][rpc] settings_get: entry"); + let config = config::rpc::load_config_with_timeout().await.map_err(|e| { + warn!("[heartbeat][rpc] settings_get: load_config failed: {e}"); + e + })?; + debug!("[heartbeat][rpc] settings_get: exit ok"); + Ok(RpcOutcome::single_log( + json!({ "settings": view(&config) }), + "heartbeat settings loaded", + )) +} + +pub async fn settings_set( + patch: HeartbeatSettingsPatch, +) -> Result, String> { + debug!("[heartbeat][rpc] settings_set: entry"); + let mut config = config::rpc::load_config_with_timeout().await.map_err(|e| { + warn!("[heartbeat][rpc] settings_set: load_config failed: {e}"); + e + })?; + + if let Some(enabled) = patch.enabled { + config.heartbeat.enabled = enabled; + } + if let Some(interval_minutes) = patch.interval_minutes { + // Clamp to the 5-minute minimum that HeartbeatEngine::run enforces at runtime. + config.heartbeat.interval_minutes = interval_minutes.max(5); + } + if let Some(inference_enabled) = patch.inference_enabled { + config.heartbeat.inference_enabled = inference_enabled; + } + if let Some(notify_meetings) = patch.notify_meetings { + config.heartbeat.notify_meetings = notify_meetings; + } + if let Some(notify_reminders) = patch.notify_reminders { + config.heartbeat.notify_reminders = notify_reminders; + } + if let Some(notify_relevant_events) = patch.notify_relevant_events { + config.heartbeat.notify_relevant_events = notify_relevant_events; + } + if let Some(external_delivery_enabled) = patch.external_delivery_enabled { + config.heartbeat.external_delivery_enabled = external_delivery_enabled; + } + if let Some(meeting_lookahead_minutes) = patch.meeting_lookahead_minutes { + config.heartbeat.meeting_lookahead_minutes = meeting_lookahead_minutes.max(1); + } + if let Some(reminder_lookahead_minutes) = patch.reminder_lookahead_minutes { + config.heartbeat.reminder_lookahead_minutes = reminder_lookahead_minutes.max(1); + } + + config.save().await.map_err(|e| { + warn!("[heartbeat][rpc] settings_set: config.save failed: {e}"); + e.to_string() + })?; + + debug!("[heartbeat][rpc] settings_set: exit ok"); + Ok(RpcOutcome::single_log( + json!({ "settings": view(&config) }), + "heartbeat settings saved", + )) +} + +pub async fn tick_now() -> Result, String> { + debug!("[heartbeat][rpc] tick_now: entry"); + let config = config::rpc::load_config_with_timeout().await.map_err(|e| { + warn!("[heartbeat][rpc] tick_now: load_config failed: {e}"); + e + })?; + let summary = planner::evaluate_and_dispatch(&config, Utc::now()).await; + debug!( + source_events = summary.source_events, + deliveries_sent = summary.deliveries_sent, + "[heartbeat][rpc] tick_now: exit ok" + ); + Ok(RpcOutcome::single_log( + json!({ "summary": summary }), + "heartbeat planner tick completed", + )) +} + +fn view(config: &Config) -> HeartbeatSettingsView { + HeartbeatSettingsView { + enabled: config.heartbeat.enabled, + interval_minutes: config.heartbeat.interval_minutes, + inference_enabled: config.heartbeat.inference_enabled, + notify_meetings: config.heartbeat.notify_meetings, + notify_reminders: config.heartbeat.notify_reminders, + notify_relevant_events: config.heartbeat.notify_relevant_events, + external_delivery_enabled: config.heartbeat.external_delivery_enabled, + meeting_lookahead_minutes: config.heartbeat.meeting_lookahead_minutes, + reminder_lookahead_minutes: config.heartbeat.reminder_lookahead_minutes, + } +} diff --git a/src/openhuman/heartbeat/schemas.rs b/src/openhuman/heartbeat/schemas.rs index ecb39697d..4d4d3a3c7 100644 --- a/src/openhuman/heartbeat/schemas.rs +++ b/src/openhuman/heartbeat/schemas.rs @@ -1,10 +1,159 @@ -use crate::core::all::RegisteredController; -use crate::core::ControllerSchema; +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; pub fn all_controller_schemas() -> Vec { - Vec::new() + vec![ + schemas("settings_get"), + schemas("settings_set"), + schemas("tick_now"), + ] } pub fn all_registered_controllers() -> Vec { - Vec::new() + vec![ + RegisteredController { + schema: schemas("settings_get"), + handler: handle_settings_get, + }, + RegisteredController { + schema: schemas("settings_set"), + handler: handle_settings_set, + }, + RegisteredController { + schema: schemas("tick_now"), + handler: handle_tick_now, + }, + ] +} + +pub fn schemas(function: &str) -> ControllerSchema { + match function { + "settings_get" => ControllerSchema { + namespace: "heartbeat", + function: "settings_get", + description: "Read heartbeat proactive notification settings.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "settings", + ty: TypeSchema::Json, + comment: "Current heartbeat settings.", + required: true, + }], + }, + "settings_set" => ControllerSchema { + namespace: "heartbeat", + function: "settings_set", + description: "Update heartbeat proactive notification settings.", + inputs: vec![ + optional_bool("enabled", "Enable or disable heartbeat loop."), + optional_u64("interval_minutes", "Tick interval in minutes."), + optional_bool( + "inference_enabled", + "Enable subconscious inference during heartbeat ticks.", + ), + optional_bool( + "notify_meetings", + "Enable proactive notifications for upcoming meetings.", + ), + optional_bool( + "notify_reminders", + "Enable proactive notifications for reminders.", + ), + optional_bool( + "notify_relevant_events", + "Enable proactive notifications for urgent/relevant events.", + ), + optional_bool( + "external_delivery_enabled", + "Allow proactive delivery to external active channels.", + ), + optional_u64( + "meeting_lookahead_minutes", + "Max lookahead window (minutes) for meeting notifications.", + ), + optional_u64( + "reminder_lookahead_minutes", + "Max lookahead window (minutes) for reminder notifications.", + ), + ], + outputs: vec![FieldSchema { + name: "settings", + ty: TypeSchema::Json, + comment: "Updated heartbeat settings.", + required: true, + }], + }, + "tick_now" => ControllerSchema { + namespace: "heartbeat", + function: "tick_now", + description: + "Run one immediate heartbeat planner tick for proactive event notifications.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "summary", + ty: TypeSchema::Json, + comment: "Planner tick result summary.", + required: true, + }], + }, + _ => ControllerSchema { + namespace: "heartbeat", + function: "unknown", + description: "Unknown heartbeat controller.", + inputs: vec![], + outputs: vec![FieldSchema { + name: "error", + ty: TypeSchema::String, + comment: "Lookup error details.", + required: true, + }], + }, + } +} + +fn handle_settings_get(_params: Map) -> ControllerFuture { + Box::pin(async move { + crate::openhuman::heartbeat::rpc::settings_get() + .await? + .into_cli_compatible_json() + }) +} + +fn handle_settings_set(params: Map) -> ControllerFuture { + Box::pin(async move { + let patch: crate::openhuman::heartbeat::rpc::HeartbeatSettingsPatch = + serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("invalid heartbeat settings_set params: {e}"))?; + crate::openhuman::heartbeat::rpc::settings_set(patch) + .await? + .into_cli_compatible_json() + }) +} + +fn handle_tick_now(_params: Map) -> ControllerFuture { + Box::pin(async move { + crate::openhuman::heartbeat::rpc::tick_now() + .await? + .into_cli_compatible_json() + }) +} + +fn optional_bool(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::Bool)), + comment, + required: false, + } +} + +fn optional_u64(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment, + required: false, + } } diff --git a/src/openhuman/notifications/types.rs b/src/openhuman/notifications/types.rs index 4bcadf46a..7af264933 100644 --- a/src/openhuman/notifications/types.rs +++ b/src/openhuman/notifications/types.rs @@ -15,6 +15,9 @@ pub enum CoreNotificationCategory { Agents, Skills, System, + Meetings, + Reminders, + Important, } /// Wire payload emitted on the `core_notification` socket event. Short,