feat(heartbeat): deliver durable proactive meeting/reminder notifications with dedup + category controls (#1369)

2026-07-27 21:08:00 +00:00 · 2026-05-08 19:03:31 -07:00
parent 9c1df8d91b
commit 0a749f5cec
20 changed files with 1715 additions and 8 deletions
@@ -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 = () => {
@@ -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');
@@ -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)',
@@ -18,6 +18,9 @@ const CATEGORY_LABEL: Record<NotificationCategory, string> = {
  agents: 'Agents',
  skills: 'Skills',
  system: 'System',
+  meetings: 'Meetings',
+  reminders: 'Reminders',
+  important: 'Important',
 };

 function formatTime(ts: number): string {
@@ -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);
+    });
+  });
 });
@@ -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,
@@ -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<NotificationPreferences> };
+      };
+      // 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 };
+      }
+    });
  },
 });

@@ -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(),
        }
    }
 }
@@ -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}");
+            }
        }
    }

@@ -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,
@@ -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<Utc>) -> Vec<PendingEvent> {
+    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<Utc>,
+) -> Vec<PendingEvent> {
+    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<Utc>,
+    end_window: DateTime<Utc>,
+) -> Vec<PendingEvent> {
+    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<Utc>,
+    end_window: DateTime<Utc>,
+    out: &mut Vec<PendingEvent>,
+) {
+    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<String, serde_json::Value>,
+) -> Option<DateTime<Utc>> {
+    // 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, serde_json::Value>) -> 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<DateTime<Utc>> {
+    chrono::DateTime::parse_from_rfc3339(raw)
+        .map(|dt| dt.with_timezone(&Utc))
+        .ok()
+}
+
+pub(crate) fn collect_relevant_notifications(
+    config: &Config,
+    now: DateTime<Utc>,
+) -> Vec<PendingEvent> {
+    // 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()
+}
@@ -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<Utc>) -> 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<String> = 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"
+        );
+    }
+}
@@ -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<Utc>,
+) -> 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, &notification).map(|_| ())
+}
@@ -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<Utc>,
+) -> Option<PlannedDelivery> {
+    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
+        }
+    }
+}
@@ -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<Utc>,
+}
+
+fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
+    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<bool> {
+    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<Utc>) -> Result<usize> {
+    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);
+    }
+}
@@ -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<String>,
+    pub anchor_at: DateTime<Utc>,
+}
+
+#[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,
+        }
+    }
+}
@@ -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::<Vec<_>>().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<Utc>,
+) -> String {
+    let normalized_title = title.to_ascii_lowercase();
+    let normalized_title = normalized_title
+        .split_whitespace()
+        .collect::<Vec<_>>()
+        .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
+    ))
+}
@@ -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<bool>,
+    pub interval_minutes: Option<u32>,
+    pub inference_enabled: Option<bool>,
+    pub notify_meetings: Option<bool>,
+    pub notify_reminders: Option<bool>,
+    pub notify_relevant_events: Option<bool>,
+    pub external_delivery_enabled: Option<bool>,
+    pub meeting_lookahead_minutes: Option<u32>,
+    pub reminder_lookahead_minutes: Option<u32>,
+}
+
+#[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<RpcOutcome<serde_json::Value>, 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<RpcOutcome<serde_json::Value>, 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<RpcOutcome<serde_json::Value>, 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,
+    }
+}
@@ -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<ControllerSchema> {
-    Vec::new()
+    vec![
+        schemas("settings_get"),
+        schemas("settings_set"),
+        schemas("tick_now"),
+    ]
 }

 pub fn all_registered_controllers() -> Vec<RegisteredController> {
-    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<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        crate::openhuman::heartbeat::rpc::settings_get()
+            .await?
+            .into_cli_compatible_json()
+    })
+}
+
+fn handle_settings_set(params: Map<String, Value>) -> 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<String, Value>) -> 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,
+    }
 }
@@ -15,6 +15,9 @@ pub enum CoreNotificationCategory {
    Agents,
    Skills,
    System,
+    Meetings,
+    Reminders,
+    Important,
 }

 /// Wire payload emitted on the `core_notification` socket event. Short,