From 058b79c5f79c6336ccfd5abc6f20e8510e4a5acb Mon Sep 17 00:00:00 2001 From: Shanu Date: Fri, 4 Sep 2026 07:58:13 +0530 Subject: [PATCH 1/5] fix(memory-sources): honour the sync depth, say why a sync stopped, repair older memories, release leases on quit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-ups from tracing why a Gmail source that said "100 items synced" stopped picking up new mail, plus the two gaps the memory-stack review found. Sync depth reaches Gmail. The per-source "Sync depth (days)" cap was logged and nothing else: the connector request had no field for it. tinyconnectors 1.8 adds `depth_days`; `run_sync_pass` now resolves the cap from the memory-sources registry (one place, for all five callers — `max_items` showed what open-coding the rule in each does) and sends it. The module turns it into Gmail's own `after:` search term. Zero and absent read as unbounded, which is what every earlier release did. A stopped sync says why. The module reports when the day's request budget is spent; the host dropped it, so the completed detail read "ingested 0 item(s)" and the row said "Up to date" — the opposite of what happened. The note now rides after the count (the UI's parse contract holds) and the row shows "budget spent" or "more to sync", as the run actually ended. No more "Maxed" badge. It compared chunks against an item cap; one email is several chunks, so it lit long before the cap. It returns when the status carries an item count. "Repair older memories". The backfill RPC from openhuman#6012 had no entry point in the app: no button, and the agent's tool list is fixed. A button on the Sources panel previews with a dry run and asks before the real pass, because the real pass embeds every document it files. Leases released on quit (tinymemory#133, host half). The embedded server's graceful path is a cancellation token, not SIGTERM, so the core's shutdown hooks never ran and nothing called the memory driver's `shutdown`; every launch waited the job leases out. `memory::exit::shutdown_for_exit` runs in the server's post-drain block, per-driver bounded, and the Tauri shell gives the server a bounded moment to drain before aborting it. Also: the status prefix normalises the toolkit the way the engine keys the rows. vendor/tinyconnectors points at the fix commit for now; it moves to the release tag, with the registry re-pin, before this merges. --- app/src-tauri/src/core_process.rs | 34 +++- .../intelligence/MemorySourceRow.tsx | 44 +++-- .../intelligence/MemorySourcesRegistry.tsx | 151 +++++++++++++++- .../intelligence/SourceSettingsPanel.tsx | 32 +--- .../MemorySourcesRegistry.sync.test.tsx | 169 ++++++++++++++++++ .../intelligence/memorySourcesSyncTypes.ts | 12 ++ app/src/lib/i18n/ar.ts | 14 +- app/src/lib/i18n/bn.ts | 15 +- app/src/lib/i18n/de.ts | 17 +- app/src/lib/i18n/en.ts | 15 +- app/src/lib/i18n/es.ts | 16 +- app/src/lib/i18n/fr.ts | 17 +- app/src/lib/i18n/hi.ts | 15 +- app/src/lib/i18n/id.ts | 15 +- app/src/lib/i18n/it.ts | 16 +- app/src/lib/i18n/ko.ts | 15 +- app/src/lib/i18n/pl.ts | 16 +- app/src/lib/i18n/pt.ts | 17 +- app/src/lib/i18n/ru.ts | 17 +- app/src/lib/i18n/zh-CN.ts | 14 +- .../utils/tauriCommands/memoryTree.test.ts | 49 +++++ app/src/utils/tauriCommands/memoryTree.ts | 62 +++++++ src/core/runtime/builder.rs | 7 + src/core/shutdown.rs | 11 ++ .../integrations/composio/ops/mod.rs | 1 + .../composio/ops/providers_ops.rs | 120 ++++++++++++- .../composio/ops_tests_part_03_tests.rs | 5 +- .../composio/ops_tests_part_04_tests.rs | 52 ++++++ src/openhuman/memory/binding.rs | 15 ++ src/openhuman/memory/exit.rs | 47 +++++ src/openhuman/memory/exit_tests.rs | 30 ++++ src/openhuman/memory/mod.rs | 3 + src/openhuman/memory/sources/status.rs | 10 +- vendor/tinyconnectors | 2 +- 34 files changed, 1003 insertions(+), 72 deletions(-) create mode 100644 src/openhuman/memory/exit.rs create mode 100644 src/openhuman/memory/exit_tests.rs diff --git a/app/src-tauri/src/core_process.rs b/app/src-tauri/src/core_process.rs index 76638f1d3a..1c1a0c8865 100644 --- a/app/src-tauri/src/core_process.rs +++ b/app/src-tauri/src/core_process.rs @@ -683,14 +683,40 @@ impl CoreProcessHandle { /// Synchronous-friendly shutdown for `RunEvent::ExitRequested`. /// - /// Aborts the embedded server task so any background tokio tasks the - /// server spawned stop driving I/O before CEF's teardown runs. Cheap - /// and non-blocking on the UI thread — `JoinHandle::abort` returns - /// immediately. + /// Cancels the embedded server's token, gives it a bounded moment to + /// drain, then aborts whatever is left so any background tokio tasks the + /// server spawned stop driving I/O before CEF's teardown runs. + /// + /// The moment is what makes the server's post-drain teardown real. The + /// memory engine releases its job leases there, and an immediate abort + /// skipped it on every normal quit, so every next launch waited the + /// leases out (tinymemory#133). The wait is the same shape as the gateway + /// shutdown beside it: short, bounded, and worth the last moment of the + /// UI thread. A server that does not finish in time is aborted as before. pub async fn send_terminate_signal(&self) { self.cancel_shutdown_token(" on app shutdown").await; + self.drain_task_briefly().await; self.abort_task(" on app shutdown").await; } + + /// Wait a bounded moment for the server task to finish on its own after + /// its token was cancelled, so the teardown inside it runs. + async fn drain_task_briefly(&self) { + const BUDGET: Duration = Duration::from_millis(2_500); + let mut task_guard = self.task.lock().await; + let Some(task) = task_guard.as_mut() else { + return; + }; + match timeout(BUDGET, task).await { + Ok(_) => { + task_guard.take(); + log::info!("[core] embedded core server task drained on app shutdown"); + } + Err(_) => log::warn!( + "[core] embedded core server task did not drain within {BUDGET:?}; aborting" + ), + } + } } /// A non-OpenHuman process holding the core RPC port. Surfaced to the frontend diff --git a/app/src/components/intelligence/MemorySourceRow.tsx b/app/src/components/intelligence/MemorySourceRow.tsx index e348c004c6..933c9a863e 100644 --- a/app/src/components/intelligence/MemorySourceRow.tsx +++ b/app/src/components/intelligence/MemorySourceRow.tsx @@ -66,6 +66,19 @@ interface SourceRowProps { onSignIn: () => void; } +/** + * The line shown beside a successful count when the run stopped short. A + * spent budget with a zero count is already the main text (there is nothing + * else to say), so it is not repeated here. + */ +function syncNoteKey(result: SyncResult): string | null { + if (result.note === 'more_pending') return 'memorySources.sync.morePending'; + if (result.note === 'budget_spent' && result.items && result.items > 0) { + return 'memorySources.sync.budgetSpent'; + } + return null; +} + export function MemorySourceRow({ source, status, @@ -97,6 +110,7 @@ export function MemorySourceRow({ // renders live progress / a terminal chip instead, and `chunks_pending` is // legitimately transient, so we suppress the warning until things settle. const settled = !progress && !result; + const noteKey = result ? syncNoteKey(result) : null; const health: SourcePipelineHealth = settled ? deriveSourcePipelineHealth(status, pipeline) : { state: 'none', issues: [], authRelated: false }; @@ -164,12 +178,23 @@ export function MemorySourceRow({ {!progress && result && (
{result.kind === 'success' ? ( - - - {result.items && result.items > 0 - ? `${result.items.toLocaleString()} ${t('memorySources.sync.itemsSynced')}` - : t('memorySources.sync.upToDate')} - + <> + + + {result.items && result.items > 0 + ? `${result.items.toLocaleString()} ${t('memorySources.sync.itemsSynced')}` + : result.note === 'budget_spent' + ? t('memorySources.sync.budgetSpent') + : t('memorySources.sync.upToDate')} + + {noteKey && ( + + {t(noteKey)} + + )} + ) : (
- + diff --git a/app/src/components/intelligence/MemorySourcesRegistry.tsx b/app/src/components/intelligence/MemorySourcesRegistry.tsx index 6d7906690d..b6643c91a2 100644 --- a/app/src/components/intelligence/MemorySourcesRegistry.tsx +++ b/app/src/components/intelligence/MemorySourcesRegistry.tsx @@ -28,6 +28,8 @@ import type { ToastNotification, } from '../../types/intelligence'; import { + type BackfillConnectorTreesResponse, + memoryTreeBackfillConnectorTrees, memoryTreeFlushSource, memoryTreePipelineStatus, type MemoryTreePipelineStatus, @@ -41,6 +43,7 @@ import { AllInIcon, PlusIcon } from './memorySourcesIcons'; import { sourceTreeScope } from './memorySourcesRowHelpers'; import { STAGE_FALLBACK_PERCENT, + type SyncNote, type SyncProgress, type SyncResult, } from './memorySourcesSyncTypes'; @@ -89,6 +92,23 @@ export function parseIngestedCount(detail: string | null): number | null { return null; } +/** + * Why a completed run stopped short, from the remainder the core writes after + * the count: `", more pending — Sync again to continue"` when the per-run cap + * left more to read, `"; today's provider request budget is spent"` when the + * day's budget did. The number alone made both read as a finished sync, and a + * spent budget with zero new items read as "Up to date" — the opposite of what + * happened. The budget wins when both appear: it is the reason nothing more + * will arrive today. + */ +export function parseSyncNote(detail: string | null): SyncNote | null { + if (!detail) return null; + const lower = detail.toLowerCase(); + if (lower.includes('budget')) return 'budget_spent'; + if (lower.includes('more pending')) return 'more_pending'; + return null; +} + export function MemorySourcesRegistry({ onToast, pollIntervalMs = 5000, @@ -120,6 +140,14 @@ export function MemorySourcesRegistry({ const [allInModalOpen, setAllInModalOpen] = useState(false); const [applyingAllIn, setApplyingAllIn] = useState(false); const allInInFlightRef = useRef(false); + // "Repair older memories" (openhuman#6012): the backfill RPC has no other + // entry point in the app. Two steps — a dry run that only counts, then the + // real pass behind a confirmation — because the real pass embeds every + // document it files, and that spends credits. + const [repairModalOpen, setRepairModalOpen] = useState(false); + const [repairPreview, setRepairPreview] = useState(null); + const [repairing, setRepairing] = useState(false); + const repairInFlightRef = useRef(false); const [expandedSettingsId, setExpandedSettingsId] = useState(null); // Refs let the (intentionally dep-free) sync-stage listener fire accurate @@ -177,18 +205,25 @@ export function MemorySourcesRegistry({ // Success: record + toast the item count parsed from the detail // ("ingested N item(s)"). 0 new items → "up to date" (#3295). const items = parseIngestedCount(data?.detail ?? null); + const note = parseSyncNote(data?.detail ?? null); setSyncResults(prev => { const next = new Map(prev); - next.set(rowId, { kind: 'success', items, reason: null }); + next.set(rowId, { kind: 'success', items, reason: null, note }); return next; }); + const counted = + items && items > 0 + ? `${items} ${tt('memorySources.sync.itemsSynced')}` + : note === 'budget_spent' + ? tt('memorySources.sync.budgetSpent') + : tt('memorySources.sync.upToDate'); onToastRef.current?.({ - type: 'success', + type: note === 'budget_spent' ? 'warning' : 'success', title: `${tt('memorySources.sync.completeTitle')} ${label}`, message: - items && items > 0 - ? `${items} ${tt('memorySources.sync.itemsSynced')}` - : tt('memorySources.sync.upToDate'), + note === 'more_pending' + ? `${counted} — ${tt('memorySources.sync.morePending')}` + : counted, }); } else { // Failure: surface the reason on the row + a toast. The core already @@ -196,7 +231,7 @@ export function MemorySourcesRegistry({ const reason = data?.detail ?? null; setSyncResults(prev => { const next = new Map(prev); - next.set(rowId, { kind: 'failed', items: null, reason }); + next.set(rowId, { kind: 'failed', items: null, reason, note: null }); return next; }); onToastRef.current?.({ @@ -381,7 +416,7 @@ export function MemorySourcesRegistry({ }); setSyncResults(prev => { const next = new Map(prev); - next.set(source.id, { kind: 'failed', items: null, reason }); + next.set(source.id, { kind: 'failed', items: null, reason, note: null }); return next; }); onToast?.({ @@ -473,6 +508,70 @@ export function MemorySourcesRegistry({ } }, [onToast, t]); + const handleRepairClick = useCallback(async () => { + if (repairInFlightRef.current) return; + repairInFlightRef.current = true; + setRepairing(true); + try { + // Preview first: the dry run counts what a pass would examine and + // writes nothing, so the confirmation can name a number before any + // credit is spent. + const preview = await memoryTreeBackfillConnectorTrees({ dryRun: true }); + setRepairPreview(preview); + if (preview.scanned === 0) { + onToast?.({ type: 'success', title: t('memorySources.repair.nothing') }); + return; + } + setRepairModalOpen(true); + } catch (err) { + onToast?.({ + type: 'error', + title: t('memorySources.repair.failed'), + message: err instanceof Error ? err.message : String(err), + }); + } finally { + repairInFlightRef.current = false; + setRepairing(false); + } + }, [onToast, t]); + + const handleConfirmRepair = useCallback(async () => { + if (repairInFlightRef.current) return; + repairInFlightRef.current = true; + setRepairing(true); + setRepairModalOpen(false); + try { + const result = await memoryTreeBackfillConnectorTrees({ dryRun: false }); + const summary = t('memorySources.repair.success') + .replace('{ingested}', String(result.ingested)) + .replace('{already}', String(result.already_present)) + .replace('{skipped}', String(result.skipped)); + // The driver files up to its per-call limit and says when documents + // remain; the pass is idempotent, so "run it again" is the whole + // resume story. + if (result.more_pending) { + onToast?.({ + type: 'warning', + title: summary, + message: t('memorySources.repair.morePending'), + }); + } else { + onToast?.({ type: 'success', title: summary }); + } + void refresh(); + } catch (err) { + onToast?.({ + type: 'error', + title: t('memorySources.repair.failed'), + message: err instanceof Error ? err.message : String(err), + }); + } finally { + repairInFlightRef.current = false; + setRepairing(false); + setRepairPreview(null); + } + }, [onToast, refresh, t]); + const handleSettingsSaved = useCallback((updated: MemorySourceEntry) => { setSources(prev => prev.map(s => (s.id === updated.id ? updated : s))); }, []); @@ -496,11 +595,39 @@ export function MemorySourcesRegistry({ }, }; + const repairModal: ConfirmationModalType = { + isOpen: repairModalOpen, + title: t('memorySources.repair.title'), + message: t('memorySources.repair.message').replace( + '{scanned}', + String(repairPreview?.scanned ?? 0) + ), + confirmText: t('memorySources.repair.confirm'), + cancelText: t('memorySources.repair.cancel'), + destructive: false, + onConfirm: () => { + void handleConfirmRepair(); + }, + onCancel: () => { + setRepairModalOpen(false); + setRepairPreview(null); + }, + }; + return (

{t('memorySources.title')}

+