From c7052709e78d64904ede1dfbfba1205bb2b44f12 Mon Sep 17 00:00:00 2001 From: Srinivas Vaddi <38348871+vaddisrinivas@users.noreply.github.com> Date: Sun, 17 May 2026 04:35:55 -0400 Subject: [PATCH] Add background loop controls and usage diagnostics (#1965) Co-authored-by: Steven Enamakel --- .../components/settings/panels/AIPanel.tsx | 845 ++++++++++++++++++ .../panels/__tests__/AIPanel.test.tsx | 232 ++++- app/src/utils/tauriCommands/heartbeat.test.ts | 95 ++ app/src/utils/tauriCommands/heartbeat.ts | 61 ++ app/src/utils/tauriCommands/index.ts | 1 + src/openhuman/config/schema/heartbeat_cron.rs | 107 ++- src/openhuman/heartbeat/engine.rs | 125 ++- src/openhuman/heartbeat/planner/collectors.rs | 154 +++- src/openhuman/heartbeat/rpc.rs | 20 + src/openhuman/heartbeat/schemas.rs | 4 + src/openhuman/subconscious/global.rs | 33 +- 11 files changed, 1606 insertions(+), 71 deletions(-) create mode 100644 app/src/utils/tauriCommands/heartbeat.test.ts create mode 100644 app/src/utils/tauriCommands/heartbeat.ts diff --git a/app/src/components/settings/panels/AIPanel.tsx b/app/src/components/settings/panels/AIPanel.tsx index 42a99aaf9..d11055361 100644 --- a/app/src/components/settings/panels/AIPanel.tsx +++ b/app/src/components/settings/panels/AIPanel.tsx @@ -11,6 +11,8 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { LuCheck, LuCircleAlert } from 'react-icons/lu'; +import { listConnections as listComposioConnections } from '../../../lib/composio/composioApi'; +import type { ComposioConnection } from '../../../lib/composio/types'; import { type AISettings as ApiAISettings, type ProviderRef as ApiProviderRef, @@ -22,7 +24,20 @@ import { saveAISettings, setCloudProviderKey, } from '../../../services/api/aiSettingsApi'; +import { + creditsApi, + type CreditTransaction, + type TeamUsage, +} from '../../../services/api/creditsApi'; import type { AuthStyle } from '../../../utils/tauriCommands/config'; +import { + type HeartbeatPlannerSummary, + type HeartbeatSettings, + type HeartbeatSettingsPatch, + openhumanHeartbeatSettingsGet, + openhumanHeartbeatSettingsSet, + openhumanHeartbeatTickNow, +} from '../../../utils/tauriCommands/heartbeat'; import SettingsHeader from '../components/SettingsHeader'; import { useSettingsNavigation } from '../hooks/useSettingsNavigation'; @@ -496,6 +511,834 @@ const ProviderKeyDialog = ({ ); }; +// ───────────────────────────────────────────────────────────────────────────── +// Background loop controls + usage diagnostics +// ───────────────────────────────────────────────────────────────────────────── + +const USD = new Intl.NumberFormat('en-US', { + style: 'currency', + currency: 'USD', + minimumFractionDigits: 4, + maximumFractionDigits: 6, +}); + +const WEEK_MINUTES = 7 * 24 * 60; +const COMPOSIO_PERIODIC_TICK_MINUTES = 20; +const LEARNING_REBUILD_MINUTES = 30; +const MEMORY_WORKERS = 4; +const MEMORY_POLL_SECONDS = 5; + +const formatUsd = (value: number): string => USD.format(Number.isFinite(value) ? value : 0); + +const spendAmount = (tx: CreditTransaction): number => { + const amount = Number(tx.amountUsd); + return Number.isFinite(amount) ? Math.abs(amount) : 0; +}; + +const formatCount = (value: number): string => + new Intl.NumberFormat('en-US', { maximumFractionDigits: 0 }).format( + Number.isFinite(value) ? value : 0 + ); + +const formatDateTime = (value: string | null | undefined): string => { + if (!value) return 'n/a'; + const date = new Date(value); + if (Number.isNaN(date.getTime())) return 'n/a'; + return date.toLocaleString([], { + month: 'short', + day: 'numeric', + hour: 'numeric', + minute: '2-digit', + }); +}; + +const activeConnection = (connection: ComposioConnection): boolean => { + const status = connection.status.toUpperCase(); + return status === 'ACTIVE' || status === 'CONNECTED'; +}; + +const normalizedToolkit = (connection: ComposioConnection): string => + connection.toolkit.toLowerCase().replace(/[^a-z0-9]/g, ''); + +const isCalendarConnection = (connection: ComposioConnection): boolean => { + const toolkit = normalizedToolkit(connection); + return toolkit === 'googlecalendar' || toolkit === 'calendar'; +}; + +function summarizeSpendByAction( + transactions: CreditTransaction[] +): Array<[string, number, number]> { + const byAction = new Map(); + for (const tx of transactions) { + if (tx.type !== 'SPEND') continue; + const key = tx.action || 'SPEND'; + const prev = byAction.get(key) ?? { count: 0, total: 0 }; + prev.count += 1; + prev.total += spendAmount(tx); + byAction.set(key, prev); + } + return Array.from(byAction.entries()) + .map(([action, value]) => [action, value.count, value.total] as [string, number, number]) + .sort((a, b) => b[2] - a[2]) + .slice(0, 4); +} + +function summarizeSpendByHour(transactions: CreditTransaction[]): Array<[string, number]> { + const byHour = new Map(); + for (const tx of transactions) { + if (tx.type !== 'SPEND') continue; + const date = new Date(tx.createdAt); + if (Number.isNaN(date.getTime())) continue; + date.setMinutes(0, 0, 0); + const key = date.toLocaleString([], { month: 'short', day: 'numeric', hour: 'numeric' }); + byHour.set(key, (byHour.get(key) ?? 0) + spendAmount(tx)); + } + return Array.from(byHour.entries()) + .sort((a, b) => b[1] - a[1]) + .slice(0, 4); +} + +function summarizeSpendSample(transactions: CreditTransaction[]) { + const rows = transactions + .filter(tx => tx.type === 'SPEND') + .sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()); + const total = rows.reduce((sum, tx) => sum + spendAmount(tx), 0); + const avgRowUsd = rows.length > 0 ? total / rows.length : 0; + const times = rows + .map(tx => new Date(tx.createdAt).getTime()) + .filter(time => !Number.isNaN(time)) + .sort((a, b) => a - b); + const sampleHours = + times.length >= 2 ? Math.max((times[times.length - 1] - times[0]) / 3_600_000, 1 / 60) : 0; + const spendPerHour = sampleHours > 0 ? total / sampleHours : 0; + const rowsPerHour = sampleHours > 0 ? rows.length / sampleHours : 0; + return { rows, total, avgRowUsd, sampleHours, spendPerHour, rowsPerHour }; +} + +function describeProvider(ref: ProviderRef, providers: CloudProvider[]): string { + if (ref.kind === 'openhuman') return 'OpenHuman'; + if (ref.kind === 'local') return `Local ${ref.model}`; + const provider = providers.find(p => p.slug === ref.providerSlug); + return `${provider?.label ?? ref.providerSlug} ${ref.model || 'custom model'}`; +} + +const LoopToggle = ({ + label, + description, + checked, + busy, + onToggle, +}: { + label: string; + description: string; + checked: boolean; + busy: boolean; + onToggle: () => void; +}) => ( +
+
+
{label}
+
{description}
+
+ +
+); + +const MetricTile = ({ + label, + value, + detail, +}: { + label: string; + value: string; + detail?: string; +}) => ( +
+
{label}
+
{value}
+ {detail ?
{detail}
: null} +
+); + +const FormulaRow = ({ label, value, detail }: { label: string; value: string; detail: string }) => ( +
+
+ {label} + {value} +
+
{detail}
+
+); + +const BackgroundLoopControls = ({ + routing, + cloudProviders, +}: { + routing: RoutingMap; + cloudProviders: CloudProvider[]; +}) => { + const [settings, setSettings] = useState(null); + const [usage, setUsage] = useState(null); + const [transactions, setTransactions] = useState([]); + const [connections, setConnections] = useState([]); + const [loading, setLoading] = useState(true); + const [saving, setSaving] = useState(null); + const [runningTick, setRunningTick] = useState(false); + const [plannerSummary, setPlannerSummary] = useState(null); + const [error, setError] = useState(''); + const settingsRef = useRef(null); + const patchRequestIdRef = useRef(0); + + const commitSettings = useCallback((nextSettings: HeartbeatSettings | null) => { + settingsRef.current = nextSettings; + setSettings(nextSettings); + }, []); + + const refresh = useCallback(async () => { + setLoading(true); + setError(''); + const [heartbeatResult, usageResult, transactionsResult, connectionsResult] = + await Promise.allSettled([ + openhumanHeartbeatSettingsGet(), + creditsApi.getTeamUsage(), + creditsApi.getTransactions(200, 0), + listComposioConnections(), + ]); + + if (heartbeatResult.status === 'fulfilled') { + commitSettings(heartbeatResult.value.result.settings); + } else { + setError( + heartbeatResult.reason instanceof Error ? heartbeatResult.reason.message : 'Load failed' + ); + } + + if (usageResult.status === 'fulfilled') { + setUsage(usageResult.value); + } + + if (transactionsResult.status === 'fulfilled') { + setTransactions(transactionsResult.value.transactions ?? []); + } + + if (connectionsResult.status === 'fulfilled') { + setConnections(connectionsResult.value.connections ?? []); + } + setLoading(false); + }, [commitSettings]); + + useEffect(() => { + void refresh(); + }, [refresh]); + + const applyHeartbeatPatch = useCallback( + async (patch: HeartbeatSettingsPatch) => { + const requestId = patchRequestIdRef.current + 1; + patchRequestIdRef.current = requestId; + const savingKey = Object.keys(patch).join(','); + const previous = settingsRef.current; + setError(''); + setSaving(savingKey); + if (!previous) { + // No baseline to patch against — abandon this request. + if (patchRequestIdRef.current === requestId) { + setSaving(null); + } + return; + } + commitSettings({ ...previous, ...patch }); + try { + const response = await openhumanHeartbeatSettingsSet(patch); + // Stale response — a newer patch superseded us; drop this result. + if (patchRequestIdRef.current !== requestId) return; + commitSettings(response.result.settings); + } catch (err) { + if (patchRequestIdRef.current !== requestId) return; + commitSettings(previous); + setError(err instanceof Error ? err.message : String(err)); + } finally { + if (patchRequestIdRef.current === requestId) { + setSaving(null); + } + } + }, + [commitSettings] + ); + + const runPlannerNow = useCallback(async () => { + setRunningTick(true); + setError(''); + try { + const response = await openhumanHeartbeatTickNow(); + setPlannerSummary(response.result.summary); + await refresh(); + } catch (err) { + setError(err instanceof Error ? err.message : String(err)); + } finally { + setRunningTick(false); + } + }, [refresh]); + + const spendSample = summarizeSpendSample(transactions); + const spendRows = spendSample.rows; + const actionSummary = summarizeSpendByAction(transactions); + const hourSummary = summarizeSpendByHour(transactions); + const latestSpend = spendRows[0] ?? null; + const heartbeatIntervalMinutes = settings ? Math.max(settings.interval_minutes, 5) : 5; + const heartbeatTicksPerWeek = settings?.enabled + ? Math.ceil(WEEK_MINUTES / heartbeatIntervalMinutes) + : 0; + const activeConnections = connections.filter(activeConnection); + const activeCalendarConnections = activeConnections.filter(isCalendarConnection); + const maxCalendarConnectionsPerTick = settings + ? Math.max(settings.max_calendar_connections_per_tick ?? 2, 1) + : 2; + const calendarConnectionsPolled = settings?.notify_meetings + ? Math.min(activeCalendarConnections.length, maxCalendarConnectionsPerTick) + : 0; + const calendarConnectionsSkipped = settings?.notify_meetings + ? Math.max(activeCalendarConnections.length - calendarConnectionsPolled, 0) + : 0; + const calendarPlannerCallsPerTick = settings?.notify_meetings ? 1 + calendarConnectionsPolled : 0; + const calendarPlannerCallsPerWeek = heartbeatTicksPerWeek * calendarPlannerCallsPerTick; + const subconsciousModelCallsPerWeek = + settings?.enabled && settings.inference_enabled ? heartbeatTicksPerWeek : 0; + const composioPeriodicTicksPerWeek = Math.ceil(WEEK_MINUTES / COMPOSIO_PERIODIC_TICK_MINUTES); + const learningTicksPerWeek = Math.ceil(WEEK_MINUTES / LEARNING_REBUILD_MINUTES); + const memoryPollsPerWeek = Math.ceil((WEEK_MINUTES * 60 * MEMORY_WORKERS) / MEMORY_POLL_SECONDS); + const composioConnectionScansPerWeek = composioPeriodicTicksPerWeek * activeConnections.length; + const backgroundApiReadsPerWeek = calendarPlannerCallsPerWeek + composioConnectionScansPerWeek; + const backgroundWakeupsPerWeek = + heartbeatTicksPerWeek + + composioPeriodicTicksPerWeek + + learningTicksPerWeek + + memoryPollsPerWeek; + const scheduledCallsPerRemainingDollar = + usage && usage.remainingUsd > 0 ? backgroundApiReadsPerWeek / usage.remainingUsd : null; + const estimatedRowsLeft = + usage && spendSample.avgRowUsd > 0 + ? Math.floor(usage.remainingUsd / spendSample.avgRowUsd) + : null; + const estimatedRowsPerBudget = + usage && spendSample.avgRowUsd > 0 + ? Math.floor(usage.cycleBudgetUsd / spendSample.avgRowUsd) + : null; + const projectedHoursLeft = + usage && spendSample.spendPerHour > 0 ? usage.remainingUsd / spendSample.spendPerHour : null; + const projectionAnchorMs = latestSpend ? new Date(latestSpend.createdAt).getTime() : Number.NaN; + const projectedExhaustAt = + projectedHoursLeft !== null && Number.isFinite(projectionAnchorMs) + ? new Date(projectionAnchorMs + projectedHoursLeft * 3_600_000).toLocaleString([], { + month: 'short', + day: 'numeric', + hour: 'numeric', + minute: '2-digit', + }) + : 'n/a'; + + const loops = [ + { + name: 'Heartbeat planner', + enabled: Boolean(settings?.enabled), + cadence: `${settings?.interval_minutes ?? 5} min`, + route: describeProvider(routing.heartbeat, cloudProviders), + work: 'Runs proactive collectors: cron reminders, calendar meetings, relevant notifications.', + risk: settings?.notify_meetings + ? `${calendarPlannerCallsPerTick} Composio read call(s)/tick; ${calendarConnectionsSkipped} calendar link(s) over cap skipped.` + : 'Calendar collector off; planner reads only local enabled categories.', + }, + { + name: 'Subconscious tick', + enabled: Boolean(settings?.enabled && settings?.inference_enabled), + cadence: `${settings?.interval_minutes ?? 5} min`, + route: describeProvider(routing.subconscious, cloudProviders), + work: 'Evaluates subconscious tasks/reflections through kind=subconscious_tick.', + risk: + subconsciousModelCallsPerWeek > 0 + ? `${formatCount(subconsciousModelCallsPerWeek)} model call(s)/week at current interval.` + : 'Inference off; no scheduled subconscious model calls.', + }, + { + name: 'Memory tree workers', + enabled: true, + cadence: 'queue', + route: describeProvider(routing.memory, cloudProviders), + work: 'Extracts chunks, seals branches, runs daily digests, routes topics.', + risk: `${MEMORY_WORKERS} workers poll every ${MEMORY_POLL_SECONDS}s; LLM calls only when queue has extract/seal/digest/topic jobs.`, + }, + { + name: 'Reflection rebuild', + enabled: true, + cadence: '30 min', + route: describeProvider(routing.learning, cloudProviders), + work: 'Refreshes reflection state after memory activity.', + risk: `${formatCount(learningTicksPerWeek)} wakeups/week; LLM work only when rebuild needs reflection.`, + }, + { + name: 'Composio sync', + enabled: true, + cadence: '20 min', + route: 'Integration APIs', + work: 'Polls connected tools when provider sync is due.', + risk: `${formatCount(composioPeriodicTicksPerWeek)} wakeups/week; scans ${activeConnections.length} active connection(s).`, + }, + ]; + + return ( +
+
+

Background loops

+

+ See what runs without a chat message, pause heartbeat work, and inspect recent credit + ledger rows. +

+
+ + {error && ( +
+ {error} +
+ )} + +
+
+
+
+
+
Heartbeat controls
+
+ Defaults off. Enabling starts the loop; disabling aborts the running task. +
+
+ +
+ + {settings ? ( +
+ void applyHeartbeatPatch({ enabled: !settings.enabled })} + /> + + void applyHeartbeatPatch({ inference_enabled: !settings.inference_enabled }) + } + /> + + void applyHeartbeatPatch({ notify_meetings: !settings.notify_meetings }) + } + /> +
+ + + +
+ + void applyHeartbeatPatch({ notify_reminders: !settings.notify_reminders }) + } + /> + + void applyHeartbeatPatch({ + notify_relevant_events: !settings.notify_relevant_events, + }) + } + /> + + void applyHeartbeatPatch({ + external_delivery_enabled: !settings.external_delivery_enabled, + }) + } + /> + +
+ + + +
+ + {plannerSummary && ( +
+ Planner: {plannerSummary.source_events} source events,{' '} + {plannerSummary.deliveries_sent} sent, {plannerSummary.deliveries_skipped_dedup}{' '} + deduped. +
+ )} +
+ ) : ( +
+ {loading ? 'Loading heartbeat controls...' : 'Heartbeat controls unavailable.'} +
+ )} +
+ +
+
+ Loop map +
+
+ {loops.map(loop => ( +
+
+
{loop.name}
+
+ {loop.enabled ? 'on' : 'off'} + {loop.cadence} +
+
+
+
{loop.work}
+
+ route: {loop.route} +
+
{loop.risk}
+
+
+ ))} +
+
+
+ +
+
+
+
Recent usage ledger
+
+ Backend rows expose action/time today; source tags need backend support. +
+
+ +
+ +
+ + + + 0 ? formatUsd(spendSample.avgRowUsd) : 'n/a'} + detail={`${spendRows.length} recent spend rows`} + /> + + +
+ +
+
+ Budget math +
+
+ + + 0 ? `${formatUsd(spendSample.spendPerHour)}/hr` : 'n/a' + } + detail={ + spendSample.sampleHours > 0 + ? `${formatCount(spendSample.rowsPerHour)} rows/hr across ${spendSample.sampleHours.toFixed(1)}h sample` + : 'Need timestamps from at least two spend rows.' + } + /> + + +
+
+ +
+
+ Loop call budget +
+
+ + + + + + + +
+
+ + {latestSpend && ( +
+ Latest spend: {formatUsd(spendAmount(latestSpend))} at{' '} + {new Date(latestSpend.createdAt).toLocaleString()} ({latestSpend.action}) +
+ )} + +
+
+
+ Top actions +
+
+ {actionSummary.length > 0 ? ( + actionSummary.map(([action, count, total]) => ( +
+ {action} + + {count} / {formatUsd(total)} + +
+ )) + ) : ( +
No spend rows loaded.
+ )} +
+
+ +
+
+ Top hours +
+
+ {hourSummary.length > 0 ? ( + hourSummary.map(([hour, total]) => ( +
+ {hour} + {formatUsd(total)} +
+ )) + ) : ( +
No hourly spend yet.
+ )} +
+
+
+
+
+
+ ); +}; + // CloudProviderCard was removed alongside the list-based auth UI. The new // chip layout (ProviderToggleChip) covers the same affordances with less // chrome. CloudProviderEditor still exists for the advanced add/edit flow, @@ -1042,6 +1885,8 @@ const AIPanel = ({ embedded = false }: AIPanelProps = {}) => { {/* end of Routing section */} + + {isDirty && ( diff --git a/app/src/components/settings/panels/__tests__/AIPanel.test.tsx b/app/src/components/settings/panels/__tests__/AIPanel.test.tsx index 987bbe50a..7b857c99a 100644 --- a/app/src/components/settings/panels/__tests__/AIPanel.test.tsx +++ b/app/src/components/settings/panels/__tests__/AIPanel.test.tsx @@ -1,13 +1,20 @@ -import { fireEvent, screen, waitFor } from '@testing-library/react'; +import { fireEvent, screen, waitFor, within } from '@testing-library/react'; import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { listConnections as listComposioConnections } from '../../../../lib/composio/composioApi'; import { loadAISettings, loadLocalProviderSnapshot, saveAISettings, setCloudProviderKey, } from '../../../../services/api/aiSettingsApi'; +import { creditsApi } from '../../../../services/api/creditsApi'; import { renderWithProviders } from '../../../../test/test-utils'; +import { + openhumanHeartbeatSettingsGet, + openhumanHeartbeatSettingsSet, + openhumanHeartbeatTickNow, +} from '../../../../utils/tauriCommands/heartbeat'; import AIPanel from '../AIPanel'; vi.mock('../../../../services/api/aiSettingsApi', () => ({ @@ -44,6 +51,18 @@ vi.mock('../../hooks/useSettingsNavigation', () => ({ }), })); +vi.mock('../../../../utils/tauriCommands/heartbeat', () => ({ + openhumanHeartbeatSettingsGet: vi.fn(), + openhumanHeartbeatSettingsSet: vi.fn(), + openhumanHeartbeatTickNow: vi.fn(), +})); + +vi.mock('../../../../services/api/creditsApi', () => ({ + creditsApi: { getTeamUsage: vi.fn(), getTransactions: vi.fn() }, +})); + +vi.mock('../../../../lib/composio/composioApi', () => ({ listConnections: vi.fn() })); + const baseSettings = { cloudProviders: [ { @@ -69,11 +88,95 @@ const baseSettings = { const baseLocalSnapshot = { status: null, diagnostics: null, presets: null, installedModels: [] }; +const baseHeartbeatSettings = { + enabled: true, + interval_minutes: 15, + inference_enabled: true, + notify_meetings: true, + notify_reminders: true, + notify_relevant_events: false, + external_delivery_enabled: false, + meeting_lookahead_minutes: 60, + max_calendar_connections_per_tick: 2, + reminder_lookahead_minutes: 30, +}; + +const baseUsage = { + remainingUsd: 1.5, + cycleBudgetUsd: 10, + cycleLimit5hr: 0.12, + cycleLimit7day: 8.5, + fiveHourCapUsd: 1, + fiveHourResetsAt: '2026-05-17T08:00:00.000Z', + cycleStartDate: '2026-05-14T00:00:00.000Z', + cycleEndsAt: '2026-05-21T00:00:00.000Z', +}; + +const baseTransactions = [ + { + id: 'older', + type: 'SPEND' as const, + action: 'SPEND:USAGE_DEDUCTION:USER', + amountUsd: -0.25, + balanceAfterUsd: 9.75, + createdAt: '2026-05-17T01:00:00.000Z', + }, + { + id: 'earn', + type: 'EARN' as const, + action: 'TOPUP', + amountUsd: 1, + balanceAfterUsd: 10.75, + createdAt: '2026-05-17T02:00:00.000Z', + }, + { + id: 'latest', + type: 'SPEND' as const, + action: 'HEARTBEAT', + amountUsd: -0.5, + balanceAfterUsd: 9.25, + createdAt: '2026-05-17T03:00:00.000Z', + }, +]; + +const baseConnections = [ + { id: 'cal-1', toolkit: 'googlecalendar', status: 'ACTIVE' }, + { id: 'cal-2', toolkit: 'calendar', status: 'CONNECTED' }, + { id: 'cal-3', toolkit: 'google_calendar', status: 'ACTIVE' }, + { id: 'slack-1', toolkit: 'slack', status: 'ACTIVE' }, + { id: 'pending-cal', toolkit: 'googlecalendar', status: 'PENDING' }, +]; + describe('AIPanel', () => { beforeEach(() => { vi.clearAllMocks(); vi.mocked(loadAISettings).mockResolvedValue(baseSettings); vi.mocked(loadLocalProviderSnapshot).mockResolvedValue(baseLocalSnapshot); + vi.mocked(openhumanHeartbeatSettingsGet).mockResolvedValue({ + result: { settings: baseHeartbeatSettings }, + logs: [], + }); + vi.mocked(openhumanHeartbeatSettingsSet).mockResolvedValue({ + result: { settings: baseHeartbeatSettings }, + logs: [], + }); + vi.mocked(openhumanHeartbeatTickNow).mockResolvedValue({ + result: { + summary: { + source_events: 3, + deliveries_attempted: 2, + deliveries_sent: 1, + deliveries_skipped_dedup: 1, + }, + }, + logs: [], + }); + vi.mocked(creditsApi.getTeamUsage).mockResolvedValue(baseUsage); + vi.mocked(creditsApi.getTransactions).mockResolvedValue({ + transactions: baseTransactions, + total: baseTransactions.length, + }); + vi.mocked(listComposioConnections).mockResolvedValue({ connections: baseConnections }); }); it('renders the LLM Providers + Routing top-level section headers', async () => { @@ -316,4 +419,131 @@ describe('AIPanel', () => { // Specifically: no "Disconnect OpenAI" switch (chip is still in off state). expect(screen.queryByRole('switch', { name: /Disconnect OpenAI/i })).not.toBeInTheDocument(); }); + + it('renders background loop diagnostics with newest spend row and budget math', async () => { + renderWithProviders(); + + await waitFor(() => expect(screen.getByText('Background loops')).toBeInTheDocument()); + + expect(screen.getByText('Heartbeat controls')).toBeInTheDocument(); + expect(screen.getByText('Recent usage ledger')).toBeInTheDocument(); + expect(screen.getByText('Loop map')).toBeInTheDocument(); + expect(screen.getByText('Heartbeat planner')).toBeInTheDocument(); + expect(screen.getByText('Subconscious tick')).toBeInTheDocument(); + expect(screen.getByText('Memory tree workers')).toBeInTheDocument(); + expect(screen.getByText('Reflection rebuild')).toBeInTheDocument(); + expect(screen.getByText('Composio sync')).toBeInTheDocument(); + + expect(screen.getByText('Week budget')).toBeInTheDocument(); + expect(screen.getByText('$10.0000')).toBeInTheDocument(); + expect(screen.getByText('Week remaining')).toBeInTheDocument(); + expect(screen.getByText('$1.5000')).toBeInTheDocument(); + expect(screen.getByText('Avg spend row')).toBeInTheDocument(); + expect(screen.getByText('Bg API reads')).toBeInTheDocument(); + expect(screen.getByText('Bg wakeups')).toBeInTheDocument(); + + expect(screen.getByText('Rows left')).toBeInTheDocument(); + expect(screen.getByText('Rows per full week budget')).toBeInTheDocument(); + expect(screen.getByText('Sample burn rate')).toBeInTheDocument(); + expect(screen.getByText('Projected empty')).toBeInTheDocument(); + expect(screen.getByText('API reads per $ remaining')).toBeInTheDocument(); + expect(screen.getByText('Loop call budget')).toBeInTheDocument(); + expect(screen.getByText('Calendar fanout cap')).toBeInTheDocument(); + expect(screen.getByText('Subconscious model calls')).toBeInTheDocument(); + expect(screen.getByText('Composio sync scans')).toBeInTheDocument(); + expect(screen.getByText('Memory worker polls')).toBeInTheDocument(); + + expect(screen.getByText(/3 Composio read call\(s\)\/tick/)).toBeInTheDocument(); + expect(screen.getByText(/1 calendar link\(s\) over cap skipped/)).toBeInTheDocument(); + expect(screen.getByText(/2\/3 conn\/tick/)).toBeInTheDocument(); + expect(screen.getByText('HEARTBEAT')).toBeInTheDocument(); + expect(screen.getByText('SPEND:USAGE_DEDUCTION:USER')).toBeInTheDocument(); + expect(screen.getByText(/Latest spend: \$0\.5000/)).toBeInTheDocument(); + }); + + it('patches heartbeat controls and runs a manual planner tick', async () => { + let currentSettings = { ...baseHeartbeatSettings }; + vi.mocked(openhumanHeartbeatSettingsGet).mockImplementation(async () => ({ + result: { settings: currentSettings }, + logs: [], + })); + vi.mocked(openhumanHeartbeatSettingsSet).mockImplementation(async patch => { + currentSettings = { ...currentSettings, ...patch }; + return { result: { settings: currentSettings }, logs: [] }; + }); + + renderWithProviders(); + await waitFor(() => expect(screen.getByText('Heartbeat controls')).toBeInTheDocument()); + + const clickToggle = async (label: string, expectedPatch: Record) => { + const row = screen.getByText(label).parentElement!.parentElement!; + fireEvent.click(within(row).getByRole('switch')); + await waitFor(() => + expect(vi.mocked(openhumanHeartbeatSettingsSet)).toHaveBeenLastCalledWith(expectedPatch) + ); + }; + + await clickToggle('Heartbeat loop', { enabled: false }); + await clickToggle('Subconscious inference', { inference_enabled: false }); + await clickToggle('Calendar meeting checks', { notify_meetings: false }); + await clickToggle('Cron reminder checks', { notify_reminders: false }); + await clickToggle('Relevant notification checks', { notify_relevant_events: true }); + await clickToggle('External delivery', { external_delivery_enabled: true }); + + fireEvent.change(screen.getByLabelText('Calendar cap'), { target: { value: '3' } }); + await waitFor(() => + expect(vi.mocked(openhumanHeartbeatSettingsSet)).toHaveBeenLastCalledWith({ + max_calendar_connections_per_tick: 3, + }) + ); + + fireEvent.change(screen.getByLabelText('Meeting lookahead'), { target: { value: '120' } }); + await waitFor(() => + expect(vi.mocked(openhumanHeartbeatSettingsSet)).toHaveBeenLastCalledWith({ + meeting_lookahead_minutes: 120, + }) + ); + + fireEvent.change(screen.getByLabelText('Reminder lookahead'), { target: { value: '60' } }); + await waitFor(() => + expect(vi.mocked(openhumanHeartbeatSettingsSet)).toHaveBeenLastCalledWith({ + reminder_lookahead_minutes: 60, + }) + ); + + fireEvent.change(screen.getByLabelText('Interval'), { target: { value: '30' } }); + await waitFor(() => + expect(vi.mocked(openhumanHeartbeatSettingsSet)).toHaveBeenLastCalledWith({ + interval_minutes: 30, + }) + ); + + fireEvent.click(screen.getByRole('button', { name: 'Planner tick now' })); + await waitFor(() => expect(vi.mocked(openhumanHeartbeatTickNow)).toHaveBeenCalled()); + await waitFor(() => expect(screen.getByText(/Planner: 3 source events/)).toBeInTheDocument()); + + fireEvent.click(screen.getByRole('button', { name: 'Refresh' })); + fireEvent.click(screen.getByRole('button', { name: 'Reload' })); + await waitFor(() => expect(vi.mocked(openhumanHeartbeatSettingsGet)).toHaveBeenCalled()); + }); + + it('shows heartbeat load and planner errors without crashing diagnostics', async () => { + vi.mocked(openhumanHeartbeatSettingsGet).mockRejectedValueOnce(new Error('heartbeat offline')); + vi.mocked(openhumanHeartbeatTickNow).mockRejectedValueOnce(new Error('tick failed')); + + renderWithProviders(); + + await waitFor(() => expect(screen.getByText('heartbeat offline')).toBeInTheDocument()); + expect(screen.getByText('Heartbeat controls unavailable.')).toBeInTheDocument(); + + vi.mocked(openhumanHeartbeatSettingsGet).mockResolvedValueOnce({ + result: { settings: baseHeartbeatSettings }, + logs: [], + }); + fireEvent.click(screen.getByRole('button', { name: 'Refresh' })); + await waitFor(() => expect(screen.getByText('Heartbeat controls')).toBeInTheDocument()); + + fireEvent.click(screen.getByRole('button', { name: 'Planner tick now' })); + await waitFor(() => expect(screen.getByText('tick failed')).toBeInTheDocument()); + }); }); diff --git a/app/src/utils/tauriCommands/heartbeat.test.ts b/app/src/utils/tauriCommands/heartbeat.test.ts new file mode 100644 index 000000000..c3093f5b1 --- /dev/null +++ b/app/src/utils/tauriCommands/heartbeat.test.ts @@ -0,0 +1,95 @@ +import { isTauri } from '@tauri-apps/api/core'; +import { beforeEach, describe, expect, type Mock, test, vi } from 'vitest'; + +import { callCoreRpc } from '../../services/coreRpcClient'; + +vi.mock('@tauri-apps/api/core', () => ({ isTauri: vi.fn() })); +vi.mock('../../services/coreRpcClient', () => ({ callCoreRpc: vi.fn() })); + +describe('tauriCommands/heartbeat', () => { + const mockIsTauri = isTauri as Mock; + const mockCallCoreRpc = callCoreRpc as Mock; + + beforeEach(() => { + vi.clearAllMocks(); + mockIsTauri.mockReturnValue(true); + Object.defineProperty(window, '__TAURI_INTERNALS__', { + configurable: true, + value: { invoke: vi.fn() }, + }); + }); + + test('reads heartbeat settings', async () => { + const { openhumanHeartbeatSettingsGet } = await import('./heartbeat'); + mockCallCoreRpc.mockResolvedValue({ + result: { + settings: { + enabled: false, + interval_minutes: 5, + inference_enabled: false, + notify_meetings: false, + notify_reminders: false, + notify_relevant_events: false, + external_delivery_enabled: false, + meeting_lookahead_minutes: 120, + max_calendar_connections_per_tick: 2, + reminder_lookahead_minutes: 30, + }, + }, + logs: [], + }); + + const out = await openhumanHeartbeatSettingsGet(); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ method: 'openhuman.heartbeat_settings_get' }); + expect(out.result.settings.enabled).toBe(false); + }); + + test('saves heartbeat settings patch', async () => { + const { openhumanHeartbeatSettingsSet } = await import('./heartbeat'); + mockCallCoreRpc.mockResolvedValue({ result: { settings: { enabled: true } }, logs: [] }); + + await openhumanHeartbeatSettingsSet({ enabled: true, interval_minutes: 15 }); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.heartbeat_settings_set', + params: { enabled: true, interval_minutes: 15 }, + }); + }); + + test('runs a planner tick now', async () => { + const { openhumanHeartbeatTickNow } = await import('./heartbeat'); + mockCallCoreRpc.mockResolvedValue({ + result: { + summary: { + source_events: 3, + deliveries_attempted: 2, + deliveries_sent: 1, + deliveries_skipped_dedup: 1, + }, + }, + logs: [], + }); + + const out = await openhumanHeartbeatTickNow(); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ method: 'openhuman.heartbeat_tick_now' }); + expect(out.result.summary.deliveries_sent).toBe(1); + }); + + test('rejects when not running in Tauri', async () => { + mockIsTauri.mockReturnValue(false); + const { + openhumanHeartbeatSettingsGet, + openhumanHeartbeatSettingsSet, + openhumanHeartbeatTickNow, + } = await import('./heartbeat'); + + await expect(openhumanHeartbeatSettingsGet()).rejects.toThrow('Not running in Tauri'); + await expect(openhumanHeartbeatSettingsSet({ enabled: true })).rejects.toThrow( + 'Not running in Tauri' + ); + await expect(openhumanHeartbeatTickNow()).rejects.toThrow('Not running in Tauri'); + expect(mockCallCoreRpc).not.toHaveBeenCalled(); + }); +}); diff --git a/app/src/utils/tauriCommands/heartbeat.ts b/app/src/utils/tauriCommands/heartbeat.ts new file mode 100644 index 000000000..4fe21836e --- /dev/null +++ b/app/src/utils/tauriCommands/heartbeat.ts @@ -0,0 +1,61 @@ +/** + * Heartbeat loop commands. + */ +import { callCoreRpc } from '../../services/coreRpcClient'; +import { type CommandResponse, isTauri } from './common'; + +export interface HeartbeatSettings { + enabled: boolean; + interval_minutes: number; + inference_enabled: boolean; + notify_meetings: boolean; + notify_reminders: boolean; + notify_relevant_events: boolean; + external_delivery_enabled: boolean; + meeting_lookahead_minutes: number; + max_calendar_connections_per_tick: number; + reminder_lookahead_minutes: number; +} + +export type HeartbeatSettingsPatch = Partial; + +export interface HeartbeatPlannerSummary { + source_events: number; + deliveries_attempted: number; + deliveries_sent: number; + deliveries_skipped_dedup: number; +} + +export async function openhumanHeartbeatSettingsGet(): Promise< + CommandResponse<{ settings: HeartbeatSettings }> +> { + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + return await callCoreRpc>({ + method: 'openhuman.heartbeat_settings_get', + }); +} + +export async function openhumanHeartbeatSettingsSet( + patch: HeartbeatSettingsPatch +): Promise> { + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + return await callCoreRpc>({ + method: 'openhuman.heartbeat_settings_set', + params: patch, + }); +} + +export async function openhumanHeartbeatTickNow(): Promise< + CommandResponse<{ summary: HeartbeatPlannerSummary }> +> { + if (!isTauri()) { + throw new Error('Not running in Tauri'); + } + return await callCoreRpc>({ + method: 'openhuman.heartbeat_tick_now', + }); +} diff --git a/app/src/utils/tauriCommands/index.ts b/app/src/utils/tauriCommands/index.ts index 27bf41122..02c4cb52b 100644 --- a/app/src/utils/tauriCommands/index.ts +++ b/app/src/utils/tauriCommands/index.ts @@ -14,6 +14,7 @@ export * from './subconscious'; export * from './localAi'; export * from './config'; export * from './cron'; +export * from './heartbeat'; export * from './service'; export * from './accessibility'; export * from './autocomplete'; diff --git a/src/openhuman/config/schema/heartbeat_cron.rs b/src/openhuman/config/schema/heartbeat_cron.rs index e03ac9e29..de5226895 100644 --- a/src/openhuman/config/schema/heartbeat_cron.rs +++ b/src/openhuman/config/schema/heartbeat_cron.rs @@ -4,31 +4,32 @@ use schemars::JsonSchema; use serde::{Deserialize, Serialize}; /// Heartbeat configuration — periodic background loop that evaluates -/// HEARTBEAT.md tasks against workspace state using local model inference. +/// HEARTBEAT.md tasks and proactive notification sources. #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(default)] pub struct HeartbeatConfig { - /// Enable the heartbeat loop. - #[serde(default = "default_true")] + /// Enable the heartbeat loop. Opt-in because ticks may call hosted models + /// and integration APIs depending on routing and enabled collectors. + #[serde(default)] pub enabled: bool, /// Tick interval in minutes (minimum 5). #[serde(default = "default_interval_minutes")] pub interval_minutes: u32, - /// Enable subconscious inference (local model evaluation). - /// When false, the heartbeat only counts tasks without reasoning. - #[serde(default = "default_true")] + /// Enable subconscious inference. When false, heartbeat only counts tasks + /// without model-backed reasoning. + #[serde(default)] pub inference_enabled: bool, /// 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")] + #[serde(default)] pub notify_meetings: bool, /// Enable proactive notifications for reminders and scheduled items. - #[serde(default = "default_true")] + #[serde(default)] pub notify_reminders: bool, /// Enable proactive notifications for urgent/relevant events. - #[serde(default = "default_true")] + #[serde(default)] 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. @@ -37,6 +38,12 @@ pub struct HeartbeatConfig { /// Maximum lookahead window for meeting notifications. #[serde(default = "default_meeting_lookahead_minutes")] pub meeting_lookahead_minutes: u32, + /// Maximum active calendar connections polled per heartbeat planner tick. + #[serde( + default = "default_max_calendar_connections_per_tick", + deserialize_with = "deserialize_calendar_connection_limit" + )] + pub max_calendar_connections_per_tick: u32, /// Maximum lookahead window for reminder notifications. #[serde(default = "default_reminder_lookahead_minutes")] pub reminder_lookahead_minutes: u32, @@ -46,10 +53,6 @@ fn default_context_budget() -> u32 { 40_000 } -fn default_true() -> bool { - true -} - fn default_interval_minutes() -> u32 { 5 } @@ -58,6 +61,20 @@ fn default_meeting_lookahead_minutes() -> u32 { 120 } +fn default_max_calendar_connections_per_tick() -> u32 { + 2 +} + +fn deserialize_calendar_connection_limit<'de, D>(deserializer: D) -> Result +where + D: serde::Deserializer<'de>, +{ + let value = Option::::deserialize(deserializer)?; + Ok(value + .unwrap_or_else(default_max_calendar_connections_per_tick) + .max(1)) +} + fn default_reminder_lookahead_minutes() -> u32 { 30 } @@ -65,20 +82,74 @@ fn default_reminder_lookahead_minutes() -> u32 { impl Default for HeartbeatConfig { fn default() -> Self { Self { - enabled: default_true(), + enabled: false, interval_minutes: default_interval_minutes(), - inference_enabled: default_true(), + inference_enabled: false, context_budget_tokens: default_context_budget(), - notify_meetings: default_true(), - notify_reminders: default_true(), - notify_relevant_events: default_true(), + notify_meetings: false, + notify_reminders: false, + notify_relevant_events: false, external_delivery_enabled: false, meeting_lookahead_minutes: default_meeting_lookahead_minutes(), + max_calendar_connections_per_tick: default_max_calendar_connections_per_tick(), reminder_lookahead_minutes: default_reminder_lookahead_minutes(), } } } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn heartbeat_defaults_are_opt_in() { + let config = HeartbeatConfig::default(); + assert!(!config.enabled); + assert!(!config.inference_enabled); + assert!(!config.notify_meetings); + assert!(!config.notify_reminders); + assert!(!config.notify_relevant_events); + assert!(!config.external_delivery_enabled); + assert_eq!(config.interval_minutes, 5); + assert_eq!(config.max_calendar_connections_per_tick, 2); + } + + #[test] + fn heartbeat_deserialization_fills_opt_in_defaults() { + let config: HeartbeatConfig = serde_json::from_str("{}").unwrap(); + assert!(!config.enabled); + assert!(!config.inference_enabled); + assert!(!config.notify_meetings); + assert!(!config.notify_reminders); + assert!(!config.notify_relevant_events); + assert!(!config.external_delivery_enabled); + assert_eq!(config.interval_minutes, 5); + assert_eq!(config.max_calendar_connections_per_tick, 2); + assert_eq!(config.meeting_lookahead_minutes, 120); + assert_eq!(config.reminder_lookahead_minutes, 30); + + let partial: HeartbeatConfig = + serde_json::from_str(r#"{"enabled":true,"interval_minutes":15}"#).unwrap(); + assert!(partial.enabled); + assert_eq!(partial.interval_minutes, 15); + assert!(!partial.inference_enabled); + assert!(!partial.notify_meetings); + assert_eq!(partial.max_calendar_connections_per_tick, 2); + + let zero_cap: HeartbeatConfig = + serde_json::from_str(r#"{"max_calendar_connections_per_tick":0}"#).unwrap(); + assert_eq!(zero_cap.max_calendar_connections_per_tick, 1); + + let null_cap: HeartbeatConfig = + serde_json::from_str(r#"{"max_calendar_connections_per_tick":null}"#).unwrap(); + assert_eq!(null_cap.max_calendar_connections_per_tick, 2); + + let explicit_cap: HeartbeatConfig = + serde_json::from_str(r#"{"max_calendar_connections_per_tick":4}"#).unwrap(); + assert_eq!(explicit_cap.max_calendar_connections_per_tick, 4); + } +} + #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(default)] pub struct CronConfig { diff --git a/src/openhuman/heartbeat/engine.rs b/src/openhuman/heartbeat/engine.rs index 77469cca3..2c8247873 100644 --- a/src/openhuman/heartbeat/engine.rs +++ b/src/openhuman/heartbeat/engine.rs @@ -1,12 +1,12 @@ -use crate::openhuman::config::HeartbeatConfig; +use crate::openhuman::config::{Config, HeartbeatConfig}; use crate::openhuman::subconscious::global::get_or_init_engine; use anyhow::Result; use std::path::Path; use tokio::time::{self, Duration}; use tracing::{info, warn}; -/// Heartbeat engine — periodic scheduler that delegates to the subconscious -/// loop for task-driven evaluation via local model inference. +/// Heartbeat engine — periodic scheduler that delegates to planner collectors +/// and optional subconscious model inference. pub struct HeartbeatEngine { config: HeartbeatConfig, workspace_dir: std::path::PathBuf, @@ -21,30 +21,73 @@ impl HeartbeatEngine { } /// Start the heartbeat loop (runs until cancelled). - /// On each tick, delegates to the shared global subconscious engine. + /// Sleeps before the first tick so fresh login never burns budget + /// immediately, then reloads config before every tick so UI changes apply + /// without an app restart. pub async fn run(&self) -> Result<()> { - if !self.config.enabled { + let mut current = self.config.clone(); + + if !current.enabled { info!("[heartbeat] disabled"); return Ok(()); } - let interval_mins = self.config.interval_minutes.max(5); + let mut logged_settings = ( + current.interval_minutes.max(5), + current.inference_enabled, + current.notify_meetings, + current.notify_reminders, + current.notify_relevant_events, + ); info!( - "[heartbeat] started: every {} minutes, subconscious inference {}", - interval_mins, - if self.config.inference_enabled { - "enabled" - } else { - "disabled (task counting only)" - } + interval_minutes = logged_settings.0, + subconscious_inference = logged_settings.1, + notify_meetings = logged_settings.2, + notify_reminders = logged_settings.3, + notify_relevant_events = logged_settings.4, + "[heartbeat] started" ); - let sleep_secs = u64::from(interval_mins) * 60; - loop { - self.run_event_planner_tick().await; + let sleep_secs = u64::from(current.interval_minutes.max(5)) * 60; + time::sleep(Duration::from_secs(sleep_secs)).await; - if self.config.inference_enabled { + let config = match Config::load_or_init().await { + Ok(config) => config, + Err(error) => { + warn!("[heartbeat] tick skipped: failed to load config: {error}"); + continue; + } + }; + + current = config.heartbeat.clone(); + if !current.enabled { + info!("[heartbeat] stopped: disabled in config"); + return Ok(()); + } + + let next_settings = ( + current.interval_minutes.max(5), + current.inference_enabled, + current.notify_meetings, + current.notify_reminders, + current.notify_relevant_events, + ); + if next_settings != logged_settings { + logged_settings = next_settings; + info!( + interval_minutes = logged_settings.0, + subconscious_inference = logged_settings.1, + notify_meetings = logged_settings.2, + notify_reminders = logged_settings.3, + notify_relevant_events = logged_settings.4, + "[heartbeat] settings reloaded" + ); + } + + self.run_event_planner_tick_for_config(&config).await; + + if current.inference_enabled { // Get the shared global engine (same instance as RPC handlers) let lock = match get_or_init_engine().await { Ok(l) => l, @@ -86,35 +129,33 @@ 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}"); - } + async fn run_event_planner_tick_for_config(&self, config: &Config) { + if !config.heartbeat.enabled { + tracing::debug!("[heartbeat] planner skipped: heartbeat disabled"); + return; } + + if !(config.heartbeat.notify_meetings + || config.heartbeat.notify_reminders + || config.heartbeat.notify_relevant_events) + { + tracing::debug!("[heartbeat] planner skipped: notification categories 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" + ); } /// Read HEARTBEAT.md and return all parsed tasks. diff --git a/src/openhuman/heartbeat/planner/collectors.rs b/src/openhuman/heartbeat/planner/collectors.rs index 27e45af35..2ddae239c 100644 --- a/src/openhuman/heartbeat/planner/collectors.rs +++ b/src/openhuman/heartbeat/planner/collectors.rs @@ -4,7 +4,7 @@ use serde_json::json; use crate::openhuman::composio::client::{ create_composio_client, direct_execute, direct_list_connections, ComposioClientKind, }; -use crate::openhuman::composio::types::ComposioExecuteResponse; +use crate::openhuman::composio::types::{ComposioConnection, ComposioExecuteResponse}; use crate::openhuman::config::Config; use crate::openhuman::cron; use crate::openhuman::notifications::store as notifications_store; @@ -87,6 +87,66 @@ fn is_reminder_like_job(job: &cron::CronJob) -> bool { || lowered.contains("follow up") } +fn is_calendar_connection(connection: &ComposioConnection) -> bool { + if !connection.is_active() { + return false; + } + + let toolkit = connection.normalized_toolkit(); + toolkit == "googlecalendar" || toolkit == "google_calendar" || toolkit == "calendar" +} + +fn select_calendar_connections_for_tick( + connections: Vec, + limit: usize, + now: DateTime, + interval_minutes: u32, +) -> Vec { + let eligible: Vec<_> = connections + .into_iter() + .filter(is_calendar_connection) + .collect(); + let eligible_count = eligible.len(); + let selected_count = eligible_count.min(limit.max(1)); + + if selected_count == 0 { + tracing::debug!( + target: "composio", + eligible = eligible_count, + cap = limit.max(1), + selected = 0, + "[heartbeat:planner] calendar-fanout: eligible=0 cap={} selected=0", + limit.max(1) + ); + return Vec::new(); + } + + let interval_seconds = i64::from(interval_minutes.max(5)) * 60; + let tick_index = now.timestamp().div_euclid(interval_seconds); + let offset = tick_index.rem_euclid(eligible_count as i64) as usize; + let selected = eligible + .iter() + .cycle() + .skip(offset) + .take(selected_count) + .cloned() + .collect::>(); + + tracing::debug!( + target: "composio", + eligible = eligible_count, + cap = limit.max(1), + selected = selected_count, + offset, + "[heartbeat:planner] calendar-fanout: eligible={} cap={} selected={}", + eligible_count, + limit.max(1), + selected_count + ); + + selected +} + pub(crate) async fn collect_calendar_meetings( config: &Config, now: DateTime, @@ -151,11 +211,15 @@ pub(crate) async fn collect_calendar_meetings( let end_window = now + lookahead; let mut out = Vec::new(); - for conn in connections.into_iter().filter(|c| c.is_active()) { + let calendar_connection_limit = + config.heartbeat.max_calendar_connections_per_tick.max(1) as usize; + for conn in select_calendar_connections_for_tick( + connections, + calendar_connection_limit, + now, + config.heartbeat.interval_minutes, + ) { let toolkit = conn.normalized_toolkit(); - if toolkit != "googlecalendar" && toolkit != "google_calendar" && toolkit != "calendar" { - continue; - } // Build base args, then let the shared transformer fill in // `timeZone` + `singleEvents` so this poller behaves identically @@ -429,3 +493,83 @@ pub(crate) fn collect_relevant_notifications( }) .collect() } + +#[cfg(test)] +mod tests { + use chrono::TimeZone; + + use super::*; + + fn conn(id: &str, toolkit: &str, status: &str) -> ComposioConnection { + ComposioConnection { + id: id.to_string(), + toolkit: toolkit.to_string(), + status: status.to_string(), + created_at: None, + } + } + + #[test] + fn calendar_selection_rotates_across_tick_buckets() { + let connections = vec![ + conn("cal-1", "googlecalendar", "ACTIVE"), + conn("cal-2", "google_calendar", "CONNECTED"), + conn("cal-3", "calendar", "ACTIVE"), + ]; + let first_tick = Utc.timestamp_opt(0, 0).single().unwrap(); + let second_tick = Utc.timestamp_opt(300, 0).single().unwrap(); + + let first = select_calendar_connections_for_tick(connections.clone(), 2, first_tick, 5) + .into_iter() + .map(|c| c.id) + .collect::>(); + let second = select_calendar_connections_for_tick(connections, 2, second_tick, 5) + .into_iter() + .map(|c| c.id) + .collect::>(); + + assert_eq!(first, vec!["cal-1", "cal-2"]); + assert_eq!(second, vec!["cal-2", "cal-3"]); + } + + #[test] + fn calendar_selection_uses_heartbeat_interval_floor() { + let connections = vec![ + conn("cal-1", "googlecalendar", "ACTIVE"), + conn("cal-2", "google_calendar", "CONNECTED"), + conn("cal-3", "calendar", "ACTIVE"), + ]; + let one_minute_later = Utc.timestamp_opt(60, 0).single().unwrap(); + let five_minutes_later = Utc.timestamp_opt(300, 0).single().unwrap(); + + let first = + select_calendar_connections_for_tick(connections.clone(), 2, one_minute_later, 1) + .into_iter() + .map(|c| c.id) + .collect::>(); + let second = select_calendar_connections_for_tick(connections, 2, five_minutes_later, 1) + .into_iter() + .map(|c| c.id) + .collect::>(); + + assert_eq!(first, vec!["cal-1", "cal-2"]); + assert_eq!(second, vec!["cal-2", "cal-3"]); + } + + #[test] + fn calendar_selection_filters_inactive_and_non_calendar_connections() { + let connections = vec![ + conn("slack", "slack", "ACTIVE"), + conn("pending-cal", "googlecalendar", "PENDING"), + conn("active-cal", "googlecalendar", "ACTIVE"), + ]; + let now = Utc.timestamp_opt(0, 0).single().unwrap(); + + let selected = select_calendar_connections_for_tick(connections, 10, now, 5) + .into_iter() + .map(|c| c.id) + .collect::>(); + + assert_eq!(selected, vec!["active-cal"]); + } +} diff --git a/src/openhuman/heartbeat/rpc.rs b/src/openhuman/heartbeat/rpc.rs index 160f4e660..f499111d6 100644 --- a/src/openhuman/heartbeat/rpc.rs +++ b/src/openhuman/heartbeat/rpc.rs @@ -18,6 +18,7 @@ pub struct HeartbeatSettingsPatch { pub notify_relevant_events: Option, pub external_delivery_enabled: Option, pub meeting_lookahead_minutes: Option, + pub max_calendar_connections_per_tick: Option, pub reminder_lookahead_minutes: Option, } @@ -31,6 +32,7 @@ pub struct HeartbeatSettingsView { pub notify_relevant_events: bool, pub external_delivery_enabled: bool, pub meeting_lookahead_minutes: u32, + pub max_calendar_connections_per_tick: u32, pub reminder_lookahead_minutes: u32, } @@ -81,6 +83,10 @@ pub async fn settings_set( if let Some(meeting_lookahead_minutes) = patch.meeting_lookahead_minutes { config.heartbeat.meeting_lookahead_minutes = meeting_lookahead_minutes.max(1); } + if let Some(max_calendar_connections_per_tick) = patch.max_calendar_connections_per_tick { + config.heartbeat.max_calendar_connections_per_tick = + max_calendar_connections_per_tick.max(1); + } if let Some(reminder_lookahead_minutes) = patch.reminder_lookahead_minutes { config.heartbeat.reminder_lookahead_minutes = reminder_lookahead_minutes.max(1); } @@ -90,6 +96,19 @@ pub async fn settings_set( e.to_string() })?; + if config.heartbeat.enabled { + debug!("[heartbeat][rpc] settings_set: enabling — running bootstrap_after_login()"); + if let Err(error) = crate::openhuman::subconscious::global::bootstrap_after_login().await { + warn!("[heartbeat][rpc] settings_set: heartbeat bootstrap failed: {error}"); + return Err(format!( + "heartbeat settings saved, but failed to start heartbeat loop: {error}" + )); + } + } else { + debug!("[heartbeat][rpc] settings_set: disabling — stopping heartbeat loop"); + crate::openhuman::subconscious::global::stop_heartbeat_loop().await; + } + debug!("[heartbeat][rpc] settings_set: exit ok"); Ok(RpcOutcome::single_log( json!({ "settings": view(&config) }), @@ -125,6 +144,7 @@ fn view(config: &Config) -> HeartbeatSettingsView { notify_relevant_events: config.heartbeat.notify_relevant_events, external_delivery_enabled: config.heartbeat.external_delivery_enabled, meeting_lookahead_minutes: config.heartbeat.meeting_lookahead_minutes, + max_calendar_connections_per_tick: config.heartbeat.max_calendar_connections_per_tick, reminder_lookahead_minutes: config.heartbeat.reminder_lookahead_minutes, } } diff --git a/src/openhuman/heartbeat/schemas.rs b/src/openhuman/heartbeat/schemas.rs index 4d4d3a3c7..16a8902db 100644 --- a/src/openhuman/heartbeat/schemas.rs +++ b/src/openhuman/heartbeat/schemas.rs @@ -73,6 +73,10 @@ pub fn schemas(function: &str) -> ControllerSchema { "meeting_lookahead_minutes", "Max lookahead window (minutes) for meeting notifications.", ), + optional_u64( + "max_calendar_connections_per_tick", + "Max active calendar connections polled per planner tick.", + ), optional_u64( "reminder_lookahead_minutes", "Max lookahead window (minutes) for reminder notifications.", diff --git a/src/openhuman/subconscious/global.rs b/src/openhuman/subconscious/global.rs index 23763da2a..12fd6a793 100644 --- a/src/openhuman/subconscious/global.rs +++ b/src/openhuman/subconscious/global.rs @@ -86,6 +86,7 @@ pub async fn bootstrap_after_login() -> Result<(), String> { if !config.heartbeat.enabled { tracing::info!("[subconscious] heartbeat disabled in config — bootstrap skipped"); + BOOTSTRAPPED.store(false, Ordering::SeqCst); return Ok(()); } @@ -122,6 +123,32 @@ pub async fn bootstrap_after_login() -> Result<(), String> { Ok(()) } +/// Stop only the heartbeat loop. Keep the engine cache intact so manual +/// subconscious RPCs can still inspect state, while a future enable can spawn +/// a fresh loop. +pub async fn stop_heartbeat_loop() { + if let Some(handle) = heartbeat_slot().lock().await.take() { + handle.abort(); + // Await the aborted task so it fully releases its engine reference + // before we let bootstrap_after_login spawn a fresh loop. Without this + // the old task can still be executing engine.run_tick() against a + // replaced engine state. + match handle.await { + Ok(()) => { + tracing::debug!("[heartbeat] loop exited before abort completed"); + } + Err(join_err) if join_err.is_cancelled() => { + tracing::info!("[heartbeat] loop aborted"); + } + Err(join_err) => { + tracing::warn!(error = %join_err, "[heartbeat] loop abort join failed"); + } + } + } + + BOOTSTRAPPED.store(false, Ordering::SeqCst); +} + /// Tear down the engine + heartbeat loop so the next login rebuilds them /// against the new user's workspace. Call on logout or account switch. /// @@ -129,15 +156,11 @@ pub async fn bootstrap_after_login() -> Result<(), String> { /// user's `workspace_dir` and subsequent ticks / RPC queries would leak /// into the wrong DB. pub async fn reset_engine_for_user_switch() { - if let Some(handle) = heartbeat_slot().lock().await.take() { - handle.abort(); - tracing::info!("[heartbeat] loop aborted for user switch"); - } + stop_heartbeat_loop().await; let lock = engine_lock(); let mut guard = lock.lock().await; *guard = None; - BOOTSTRAPPED.store(false, Ordering::SeqCst); tracing::info!("[subconscious] engine reset for user switch"); }