{fields.map(field => {
- const cap = Number(values[field]);
const isUnlimited = (values[field] ?? '') === '';
- const isMaxed =
- COUNT_FIELDS.has(field) &&
- !isUnlimited &&
- Number.isFinite(cap) &&
- typeof syncedCount === 'number' &&
- syncedCount >= cap;
+ // No "Maxed" badge. The only count a row has is *chunks*, and a cap
+ // is *items* (emails, issues, commits); one email is several
+ // chunks, so the badge lit long before the cap was reached. It
+ // comes back when the status carries an item count (openhuman#6012).
return (
{t(FIELD_LABEL_KEYS[field])}
- {isMaxed && (
-
- {t('memorySources.settings.maxed')}
-
- )}
{isUnlimited && (
({
// ── tauriCommands mock ────────────────────────────────────────────────────────
// memoryTreePipelineStatus is polled for downstream pipeline health (GH-4690);
// default to a healthy running snapshot so rows keep their clean synced state.
+const mockTrackAnalyticsEvent = vi.fn();
+vi.mock('../../analytics', () => ({
+ trackAnalyticsEvent: (...args: unknown[]) => mockTrackAnalyticsEvent(...args),
+}));
+
vi.mock('../../../utils/tauriCommands/memoryTree', () => ({
+ memoryTreeBackfillConnectorTrees: vi.fn(),
memoryTreeFlushSource: vi.fn().mockResolvedValue({ seals_fired: 0 }),
memoryTreePipelineStatus: vi
.fn()
@@ -135,6 +142,32 @@ describe('parseIngestedCount', () => {
});
});
+// ── parseSyncNote unit tests ──────────────────────────────────────────────────
+
+describe('parseSyncNote', () => {
+ it('reads "more pending" from the remainder after the count', () => {
+ expect(parseSyncNote('ingested 100 item(s), more pending — Sync again to continue')).toBe(
+ 'more_pending'
+ );
+ });
+
+ it('reads a spent budget, and prefers it when both appear', () => {
+ expect(parseSyncNote("ingested 0 item(s); today's provider request budget is spent")).toBe(
+ 'budget_spent'
+ );
+ expect(
+ parseSyncNote(
+ "ingested 0 item(s), more pending — Sync again to continue; today's provider request budget is spent"
+ )
+ ).toBe('budget_spent');
+ });
+
+ it('returns null for a plain count or no detail', () => {
+ expect(parseSyncNote('ingested 5 item(s)')).toBeNull();
+ expect(parseSyncNote(null)).toBeNull();
+ });
+});
+
// ── MemorySourcesRegistry integration tests ───────────────────────────────────
describe('MemorySourcesRegistry', () => {
@@ -485,4 +518,208 @@ describe('MemorySourcesRegistry', () => {
// The optimistic syncing state is cleared after the RPC rejection.
expect(screen.queryByText('sync.syncing')).not.toBeInTheDocument();
});
+ it('says the budget is spent instead of "Up to date" when zero items arrive for that reason', async () => {
+ // The core writes why a run stopped after the count; a spent budget with
+ // zero new items used to read as "Up to date" — the opposite of what
+ // happened (openhuman#6012 follow-up).
+ const sources = [makeSource('src-budget')];
+ listMemorySources.mockResolvedValue(sources);
+ memorySourcesStatusList.mockResolvedValue([]);
+ const onToast = vi.fn();
+
+ renderWithProviders(;
+ backfill
+ .mockResolvedValueOnce({
+ executed: false,
+ scanned: 42,
+ ingested: 0,
+ already_present: 0,
+ skipped: 0,
+ more_pending: false,
+ notes: [],
+ })
+ .mockResolvedValueOnce({
+ executed: true,
+ scanned: 42,
+ ingested: 40,
+ already_present: 1,
+ skipped: 1,
+ more_pending: false,
+ notes: [],
+ });
+ listMemorySources.mockResolvedValue([makeSource('src-a')]);
+ memorySourcesStatusList.mockResolvedValue([]);
+ const onToast = vi.fn();
+
+ renderWithProviders(;
+ backfill.mockReset();
+ backfill.mockResolvedValueOnce({
+ executed: false,
+ scanned: 0,
+ ingested: 0,
+ already_present: 0,
+ skipped: 0,
+ more_pending: false,
+ notes: [],
+ });
+ listMemorySources.mockResolvedValue([makeSource('src-b')]);
+ memorySourcesStatusList.mockResolvedValue([]);
+ const onToast = vi.fn();
+
+ renderWithProviders( {
return out;
}
+/** Response shape for `memory_tree_backfill_connector_trees`. */
+export interface BackfillConnectorTreesResponse {
+ /** False for the dry-run preview: it counted and wrote nothing. */
+ executed: boolean;
+ /** Documents examined. */
+ scanned: number;
+ /** Documents that produced new memory-tree rows. */
+ ingested: number;
+ /** Documents the tree already held. */
+ already_present: number;
+ /** Documents left alone rather than filed under a guess. */
+ skipped: number;
+ /** The pass stopped on its limit with documents unexamined — run again. */
+ more_pending: boolean;
+ /** Bounded, human-readable reasons behind `skipped`. */
+ notes: string[];
+}
+
+/**
+ * A real pass reads and embeds up to the driver's per-call limit of documents
+ * (500) before it answers; the budget has to cover the whole pass.
+ */
+const BACKFILL_RPC_TIMEOUT_MS = 10 * 60 * 1_000;
+
+/**
+ * Re-file connector documents stored before the memory-tree routing fix
+ * (openhuman#6007) into the tree. Backed by
+ * `openhuman.memory_tree_backfill_connector_trees` (openhuman#6012).
+ *
+ * `dryRun: true` counts what a pass would examine and writes nothing; the real
+ * pass embeds every document it files, which spends credits, so the UI always
+ * previews first and asks. Idempotent: documents already in the tree come
+ * back as `already_present`, never filed twice.
+ */
+export async function memoryTreeBackfillConnectorTrees(opts: {
+ dryRun: boolean;
+ limit?: number;
+}): Promise {
+ console.debug(
+ '[memory-tree-rpc] memoryTreeBackfillConnectorTrees: entry dry_run=%s',
+ opts.dryRun
+ );
+ const resp = await callCoreRpc<
+ BackfillConnectorTreesResponse | ResultEnvelope
+ >({
+ method: 'openhuman.memory_tree_backfill_connector_trees',
+ params: { dry_run: opts.dryRun, ...(opts.limit ? { limit: opts.limit } : {}) },
+ timeoutMs: BACKFILL_RPC_TIMEOUT_MS,
+ });
+ const out = unwrapResult(resp);
+ console.debug(
+ '[memory-tree-rpc] memoryTreeBackfillConnectorTrees: exit executed=%s scanned=%d ingested=%d already=%d skipped=%d more=%s',
+ out.executed,
+ out.scanned,
+ out.ingested,
+ out.already_present,
+ out.skipped,
+ out.more_pending
+ );
+ return out;
+}
+
/** Response shape for `memory_tree_flush_now`. */
export interface FlushNowResponse {
enqueued: boolean;
diff --git a/src/core/runtime/builder.rs b/src/core/runtime/builder.rs
index f69a0cbcb0..ce8341e972 100644
--- a/src/core/runtime/builder.rs
+++ b/src/core/runtime/builder.rs
@@ -841,7 +841,16 @@ impl CoreRuntime {
});
}
- if let Some(shutdown_token) = shutdown_token {
+ // Arms memory's exit gate for the eventual exit (and clears one a
+ // previous server in this process may have left): from here on a
+ // memory binding built during exit is refused rather than missed.
+ crate::openhuman::memory::exit::server_starting();
+
+ // The serve result is held, not propagated, until the exit work below
+ // has run. A `?` here on a server error would skip the memory teardown
+ // on exactly the exits where a wedged store is likeliest, and the
+ // callers only forward the error — nobody else runs the cleanup.
+ let served = if let Some(shutdown_token) = shutdown_token {
log::info!(
"[core] embedded server waiting on cancellation token for graceful shutdown"
);
@@ -849,13 +858,26 @@ impl CoreRuntime {
.with_graceful_shutdown(async move {
shutdown_token.cancelled().await;
})
- .await?;
+ .await
} else {
axum::serve(listener, app)
.with_graceful_shutdown(crate::core::shutdown::signal())
- .await?;
+ .await
+ };
+ if let Err(error) = &served {
+ log::warn!(
+ "[core] embedded server ended with an error; running exit cleanup before \
+ reporting it: {error}"
+ );
}
+ // Memory first. The engine's queue worker holds leases on in-flight
+ // jobs, and releasing them is a write to the store, so it has to happen
+ // while the store is still open and before anything else on the way
+ // out (tinymemory#133). Bounded inside, on one shared deadline: a
+ // wedged store costs at most that budget, never the exit.
+ crate::openhuman::memory::exit::shutdown_for_exit().await;
+
// Server has stopped accepting and in-flight requests drained. Kill any
// `ollama serve` openhuman itself spawned (no-op when externally
// managed) so the next launch doesn't try to reclaim a dead daemon.
@@ -876,6 +898,7 @@ impl CoreRuntime {
}
}
+ served?;
Ok(())
}
diff --git a/src/core/shutdown.rs b/src/core/shutdown.rs
index 4d569f4d74..aa1e14a438 100644
--- a/src/core/shutdown.rs
+++ b/src/core/shutdown.rs
@@ -52,6 +52,27 @@ async fn run_hooks() {
}
}
+/// Run every registered hook now, once, without waiting for a signal.
+///
+/// For the embedded server, whose graceful path is a cancellation token rather
+/// than SIGTERM: [`signal`] never resolves there, so the hooks it would have
+/// run — the memory engine releasing its queue leases, above all — never ran
+/// on a normal quit. Drains the registry, so a later call, or a signal landing
+/// mid-teardown, finds nothing to run twice.
+///
+/// Side by side, unlike the signal path. The caller runs this under a
+/// deadline, and in sequence one hook that never answers would keep every hook
+/// registered after it — the lease release among them — from so much as
+/// starting before the deadline dropped the lot. Run together, a hanging hook
+/// costs only itself.
+pub async fn run_hooks_now() {
+ let hooks: Vec = {
+ let mut guard = HOOKS.lock().expect("shutdown hooks poisoned");
+ std::mem::take(&mut *guard)
+ };
+ futures::future::join_all(hooks.iter().map(|hook| hook())).await;
+}
+
/// Returns a future that resolves when the process receives a termination
/// signal (SIGINT on all platforms, plus SIGTERM on Unix), then runs all
/// registered shutdown hooks.
diff --git a/src/openhuman/integrations/composio/ops/mod.rs b/src/openhuman/integrations/composio/ops/mod.rs
index a91dc8f23d..cb16fcf708 100644
--- a/src/openhuman/integrations/composio/ops/mod.rs
+++ b/src/openhuman/integrations/composio/ops/mod.rs
@@ -43,6 +43,7 @@ pub use execute::composio_execute;
#[cfg(test)]
pub(crate) use providers_ops::{
completed_sync_detail, completed_sync_detail_for_test, next_pass_budget,
+ pick_source_sync_depth_days,
};
pub use providers_ops::{
composio_get_user_profile, composio_refresh_all_identities, composio_sync,
diff --git a/src/openhuman/integrations/composio/ops/providers_ops.rs b/src/openhuman/integrations/composio/ops/providers_ops.rs
index b4394d1a2e..fa38c0d5fe 100644
--- a/src/openhuman/integrations/composio/ops/providers_ops.rs
+++ b/src/openhuman/integrations/composio/ops/providers_ops.rs
@@ -71,12 +71,27 @@ pub(crate) fn next_pass_budget(source_max_items: Option, total_written: u64
/// A parse contract, not prose: the Sources UI extracts the count with
/// `/ingested\s+(\d+)\s+item/i` and falls back to a generic "up to date"
/// when it cannot (#3295). Pinned by a unit test against that exact pattern.
-pub(crate) fn completed_sync_detail(total_written: u64, more_pending: bool) -> String {
- if more_pending {
+///
+/// `note` is what the module said about a run that stopped short — today's
+/// request budget being spent, above all. It rides *after* the count, never
+/// inside it, so the regex keeps matching and everything past the count is
+/// free text the UI can show. Without it a spent budget wrote zero items and
+/// read back as "Up to date", the opposite of what happened.
+pub(crate) fn completed_sync_detail(
+ total_written: u64,
+ more_pending: bool,
+ note: Option<&str>,
+) -> String {
+ let mut detail = if more_pending {
format!("ingested {total_written} item(s), more pending — Sync again to continue")
} else {
format!("ingested {total_written} item(s)")
+ };
+ if let Some(note) = note.map(str::trim).filter(|note| !note.is_empty()) {
+ detail.push_str("; ");
+ detail.push_str(note);
}
+ detail
}
/// Aggregate result of [`composio_refresh_all_identities`].
@@ -325,6 +340,11 @@ pub async fn composio_sync_budgeted(
let mut total_written: u64 = 0;
let mut passes = 0usize;
let mut more_pending = false;
+ // The module's own account of a pass that stopped short (the day's
+ // request budget, typically). The last pass's word wins, present or
+ // absent: it describes the state the run ended in, and a pass that
+ // completes cleanly must not inherit the note of an earlier one.
+ let mut stop_note: Option = None;
let outcome = loop {
passes += 1;
// The source's configured per-run cap wins over the pass ceiling:
@@ -346,6 +366,12 @@ pub async fn composio_sync_budgeted(
Ok(pass) => {
total_written = total_written.saturating_add(u64::from(pass.written));
more_pending = pass.more_pending;
+ stop_note = pass
+ .message
+ .as_deref()
+ .map(str::trim)
+ .filter(|message| !message.is_empty())
+ .map(str::to_string);
tracing::info!(
toolkit = %toolkit_for_log,
connection_id = %connection_for_log,
@@ -378,7 +404,11 @@ pub async fn composio_sync_budgeted(
// falls back to a generic "up to date" when it cannot (#3295).
publish_stage(
"completed",
- Some(completed_sync_detail(total_written, more_pending)),
+ Some(completed_sync_detail(
+ total_written,
+ more_pending,
+ stop_note.as_deref(),
+ )),
);
}
Err(error) => {
@@ -427,6 +457,10 @@ pub(crate) struct SyncPassOutcome {
/// Whether the module has more to read — the caller decides whether to
/// call again.
pub more_pending: bool,
+ /// What the module said about a run that stopped short — today's request
+ /// budget being spent, above all. Carried so the completed-stage detail
+ /// can say *why* zero items arrived instead of reading as "up to date".
+ pub message: Option,
}
/// Read one connection through the module and ingest what it returns.
@@ -450,6 +484,13 @@ pub(crate) async fn run_sync_pass(
) -> Result {
// Sync pages the whole account inside the call; the default 30s bus
// deadline reported failure on runs the module then finished successfully.
+ // The per-source "Sync depth (days)" cap. Until contract 1.8 gave the
+ // request a field for it, the setting reached the log and nothing else;
+ // the module now turns it into Gmail's own `after:` search term, so a
+ // bounded first sync costs one page of recent mail rather than a walk
+ // through the years. `None` reads unbounded, as every earlier release did.
+ let depth_days = source_sync_depth_days(config, toolkit, connection_id);
+
let response = connectors::call_slow::<_, ConnectorSyncResponse>(
config,
methods::SYNC,
@@ -462,6 +503,7 @@ pub(crate) async fn run_sync_pass(
// pass loop bounds nothing (review finding). complete=false at the
// budget → more_pending → the loop (or the next click) resumes.
max_items: Some(pass_budget),
+ depth_days,
..ConnectorSyncRequest::default()
},
)
@@ -474,6 +516,7 @@ pub(crate) async fn run_sync_pass(
written: 0,
already_ingested: false,
more_pending: !response.batch.complete,
+ message: response.message,
});
}
@@ -593,9 +636,80 @@ pub(crate) async fn run_sync_pass(
written: outcome.written,
already_ingested: outcome.already_ingested,
more_pending: !response.batch.complete,
+ message: response.message,
})
}
+/// The per-source "Sync depth (days)" cap for one connection, from the
+/// memory-sources registry the pass's own `config` names.
+///
+/// Resolved here rather than threaded through every caller because there are
+/// five of them (the row button, All In, the periodic loop, the connection
+/// bootstrap, Slack's own RPC), and `max_items` already showed what happens
+/// when each open-codes the same rule: two of the five disagreed
+/// (openhuman#6007). Read through `config`, not the process environment: a
+/// pass is bound to one workspace, and the global registry path would answer a
+/// caller bound to workspace B with workspace A's rows — the cross-workspace
+/// leak the registry's `_in` variants exist to prevent. `None` when the row is
+/// missing, carries no cap, or the registry cannot be read — each means "no
+/// lower bound", which is what every release before the field existed did, so
+/// a registry hiccup degrades to the old behaviour rather than to a failed
+/// sync.
+fn source_sync_depth_days(config: &Config, toolkit: &str, connection_id: &str) -> Option {
+ let sources = match crate::openhuman::memory::sources::registry::list_sources_in(config) {
+ Ok(sources) => sources,
+ Err(error) => {
+ tracing::warn!(
+ toolkit = %toolkit,
+ connection_id = %connection_id,
+ error = %error,
+ "[composio] memory-sources registry unreadable for the sync depth; \
+ syncing without a lower bound"
+ );
+ return None;
+ }
+ };
+ pick_source_sync_depth_days(
+ sources
+ .iter()
+ .filter(|source| source.kind == crate::openhuman::memory::sources::SourceKind::Composio)
+ .map(|source| {
+ (
+ source.toolkit.as_deref(),
+ source.connection_id.as_deref(),
+ source.sync_depth_days,
+ )
+ }),
+ toolkit,
+ connection_id,
+ )
+}
+
+/// The cap of the row matching `toolkit` and `connection_id`, if any.
+///
+/// Matched the way the engine keys the rows — toolkit case-insensitively and
+/// trimmed, connection trimmed — and a cap of zero reads as none: the settings
+/// field stores "unlimited" as an empty value, and a zero typed by hand would
+/// otherwise ask Gmail for mail newer than today.
+pub(crate) fn pick_source_sync_depth_days<'a>(
+ rows: impl IntoIterator- , Option<&'a str>, Option
)>,
+ toolkit: &str,
+ connection_id: &str,
+) -> Option {
+ let toolkit = toolkit.trim();
+ let connection_id = connection_id.trim();
+ rows.into_iter()
+ .find_map(|(row_toolkit, row_connection, depth)| {
+ let same_toolkit =
+ row_toolkit.is_some_and(|slug| slug.trim().eq_ignore_ascii_case(toolkit));
+ let same_connection = row_connection.is_some_and(|id| id.trim() == connection_id);
+ (same_toolkit && same_connection)
+ .then_some(depth)
+ .flatten()
+ .filter(|days| *days > 0)
+ })
+}
+
/// Parse the optional `reason` parameter into a [`SyncReason`].
///
/// `None` and the explicit `"manual"` value both map to
@@ -617,5 +731,5 @@ pub(crate) fn parse_sync_reason(raw: Option<&str>) -> OpResult {
/// name; this avoids re-plumbing the cfg(test) re-export).
#[cfg(test)]
pub(crate) fn completed_sync_detail_for_test(total: u64, more: bool) -> String {
- completed_sync_detail(total, more)
+ completed_sync_detail(total, more, None)
}
diff --git a/src/openhuman/integrations/composio/ops_tests_part_03_tests.rs b/src/openhuman/integrations/composio/ops_tests_part_03_tests.rs
index a31901326e..80236334f9 100644
--- a/src/openhuman/integrations/composio/ops_tests_part_03_tests.rs
+++ b/src/openhuman/integrations/composio/ops_tests_part_03_tests.rs
@@ -648,8 +648,9 @@ async fn composio_list_connections_returns_empty_when_direct_mode_no_key() {
fn completed_sync_detail_matches_the_ui_parse_contract() {
let re = regex::Regex::new(r"(?i)ingested\s+(\d+)\s+item").expect("ui parse regex");
for count in [0u64, 1, 200, 25_000] {
- let detail =
- crate::openhuman::integrations::composio::ops::completed_sync_detail(count, false);
+ let detail = crate::openhuman::integrations::composio::ops::completed_sync_detail(
+ count, false, None,
+ );
let caps = re
.captures(&detail)
.unwrap_or_else(|| panic!("detail must parse: {detail}"));
diff --git a/src/openhuman/integrations/composio/ops_tests_part_04_tests.rs b/src/openhuman/integrations/composio/ops_tests_part_04_tests.rs
index 805182af9d..5790018ac8 100644
--- a/src/openhuman/integrations/composio/ops_tests_part_04_tests.rs
+++ b/src/openhuman/integrations/composio/ops_tests_part_04_tests.rs
@@ -643,3 +643,55 @@ async fn enrich_leaves_unmatched_connection_unchanged() {
"connection with no cached profile must remain unenriched"
);
}
+
+/// A run that wrote nothing because the day's request budget was spent must
+/// say so: the UI shows "Up to date" for a zero count, and a spent budget is
+/// the opposite. The note rides after the count so the parse contract holds,
+/// and a blank note is no separator with nothing behind it.
+#[test]
+fn completed_detail_carries_the_module_note_after_the_count() {
+ let re = regex::Regex::new(r"(?i)ingested\s+(\d+)\s+item").expect("ui parse regex");
+ let detail = crate::openhuman::integrations::composio::ops::completed_sync_detail(
+ 0,
+ true,
+ Some("today's provider request budget is spent"),
+ );
+ let caps = re.captures(&detail).expect("detail still parses");
+ assert_eq!(&caps[1], "0");
+ assert!(
+ detail.ends_with("; today's provider request budget is spent"),
+ "{detail}"
+ );
+ let bare =
+ crate::openhuman::integrations::composio::ops::completed_sync_detail(3, false, Some(" "));
+ assert_eq!(bare, "ingested 3 item(s)");
+}
+
+/// The per-source depth cap is matched the way the engine keys the rows and a
+/// zero reads as "no cap": the settings field stores unlimited as empty, and a
+/// zero typed by hand must not ask for mail newer than today.
+#[test]
+fn source_depth_matches_the_row_and_treats_zero_as_unbounded() {
+ use crate::openhuman::integrations::composio::ops::pick_source_sync_depth_days;
+ let rows = [
+ (Some("gmail"), Some("conn-1"), Some(30)),
+ (Some("gmail"), Some("conn-2"), Some(0)),
+ (Some("notion"), Some("conn-3"), Some(14)),
+ (Some("gmail"), None, Some(7)),
+ ];
+ assert_eq!(
+ pick_source_sync_depth_days(rows, "gmail", "conn-1"),
+ Some(30)
+ );
+ assert_eq!(
+ pick_source_sync_depth_days(rows, " GMAIL ", "conn-1 "),
+ Some(30)
+ );
+ assert_eq!(pick_source_sync_depth_days(rows, "gmail", "conn-2"), None);
+ assert_eq!(pick_source_sync_depth_days(rows, "gmail", "conn-9"), None);
+ assert_eq!(
+ pick_source_sync_depth_days(rows, "notion", "conn-3"),
+ Some(14)
+ );
+ assert_eq!(pick_source_sync_depth_days([], "gmail", "conn-1"), None);
+}
diff --git a/src/openhuman/memory/binding.rs b/src/openhuman/memory/binding.rs
index fd7353216a..1601763ddd 100644
--- a/src/openhuman/memory/binding.rs
+++ b/src/openhuman/memory/binding.rs
@@ -467,6 +467,46 @@ pub(crate) fn bind_provider_for_test(
type BindingCacheKey = (PathBuf, String, MemorySubsystemConfig);
static BINDINGS: OnceLock>>> = OnceLock::new();
+/// Every binding this process has built so far, for the exit path.
+///
+/// A snapshot rather than a handle to the map: exit runs while other tasks may
+/// still be resolving bindings, and holding the lock across a driver's
+/// `shutdown` would queue them behind it.
+pub(crate) fn cached_bindings() -> Vec> {
+ let Some(cache) = BINDINGS.get() else {
+ return Vec::new();
+ };
+ match cache.read() {
+ Ok(map) => map.values().cloned().collect(),
+ Err(poisoned) => poisoned.into_inner().values().cloned().collect(),
+ }
+}
+
+/// Drop `shut_down` from the cache, so a server that starts again in this
+/// process binds fresh drivers instead of the ones exit has already torn down.
+///
+/// The shell restarts the embedded server in place (a permission refresh, an
+/// app update), and exit runs every cached driver's `shutdown` on the way out.
+/// Left in the cache, those drivers would be handed straight back to the next
+/// server — their workers stopped and their hooks already drained — and memory
+/// would stay dark until the whole desktop process restarted. Matched by
+/// identity, not by key: only what exit actually asked to shut down leaves.
+/// With the exit gate up nothing else can be in the cache by then; without it
+/// (a bare call, in tests) a binding built beside the exit stays.
+pub(crate) fn evict_bindings(shut_down: &[Arc]) {
+ if shut_down.is_empty() {
+ return;
+ }
+ let Some(cache) = BINDINGS.get() else {
+ return;
+ };
+ let mut map = match cache.write() {
+ Ok(map) => map,
+ Err(poisoned) => poisoned.into_inner(),
+ };
+ map.retain(|_, binding| !shut_down.iter().any(|gone| Arc::ptr_eq(gone, binding)));
+}
+
/// The bound memory driver for `workspace_dir`, constructing it on first use.
///
/// The same workspace always resolves to the same cached `Arc` (so
@@ -666,6 +706,16 @@ pub fn for_subtree(
let mut guard = cache
.write()
.map_err(|e| format!("[memory:binding] cache write lock poisoned: {e}"))?;
+ // Under the same lock the exit snapshot is taken under: once memory is on
+ // its way out, a driver bound now would never be asked to shut down, so
+ // it is not bound at all. The check sits inside the lock on purpose — a
+ // builder that passed it inserted before the snapshot, and one that did
+ // not is refused; there is no third case (memory/exit.rs).
+ if crate::openhuman::memory::exit::exiting() {
+ return Err(
+ "[memory:binding] memory is shutting down; not binding a new driver".to_string(),
+ );
+ }
// Re-check under the write lock: a racing caller may have bound the same
// workspace while we were building. Reuse theirs so one workspace never has
// two live drivers (kernel.md §3.1) and `capabilities()` stays asked once.
diff --git a/src/openhuman/memory/exit.rs b/src/openhuman/memory/exit.rs
new file mode 100644
index 0000000000..d2de4622b6
--- /dev/null
+++ b/src/openhuman/memory/exit.rs
@@ -0,0 +1,127 @@
+//! What memory does on the way out of the process.
+//!
+//! The engine registers exactly one shutdown hook — its queue worker releasing
+//! the leases on in-flight jobs, so the next launch re-claims that work instead
+//! of waiting the leases out (tinymemory#133). In module mode the engine banks
+//! that hook and drains it when the host calls `Shutdown` (tinymemory#137).
+//! The host never did. The embedded server's graceful path is a cancellation
+//! token, not SIGTERM, so [`crate::core::shutdown::signal`] never resolved
+//! there, and nothing else ever called the bound provider's `shutdown`: every
+//! normal quit left the leases held, and every next launch took the slow path.
+//!
+//! This is the other half. It runs from the server's post-drain block, and it
+//! is bounded, because a quit that hangs on a wedged store is worse than a
+//! lease that expires on its own.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
+
+use futures::future::join_all;
+use tokio::time::Instant;
+
+/// The whole of memory's exit work — every bound driver's `shutdown`, then
+/// the host's hook registry — must fit in this, together.
+///
+/// One deadline rather than one per driver. The Tauri shell gives the server
+/// a bounded moment to drain before it aborts the task, and a per-driver
+/// budget multiplied by the number of bindings could outrun that moment with
+/// a hook still pending. Releasing a lease is one write per in-flight job;
+/// anything slower is a store that is not going to answer, and its leases
+/// expire by themselves. The shell sizes its drain budget from this constant
+/// plus the ollama cleanup that follows it in `serve_http`.
+pub const EXIT_BUDGET: Duration = Duration::from_secs(2);
+
+/// The least the hook registry gets even when the drivers spent the budget.
+/// In-process engines (dev runs, tests) release their leases through a hook,
+/// not a driver, so the hooks are never skipped outright.
+const HOOKS_FLOOR: Duration = Duration::from_millis(250);
+
+/// Whether a real server has started in this process.
+///
+/// The exit gate below engages only then. Unit tests call
+/// [`shutdown_for_exit`] without ever serving, and a process-wide refusal to
+/// bind memory would reach every other test in the same process.
+static SERVING: AtomicBool = AtomicBool::new(false);
+
+/// Set when memory's exit work begins, cleared when a server starts.
+///
+/// Read by the binding cache *inside its write lock* before every insert. That
+/// ordering is what makes the exit snapshot complete: the flag goes up, then
+/// the snapshot is taken under the same lock, so a builder either inserted
+/// before the snapshot (and is in it) or sees the flag and is refused. No
+/// number of re-snapshots could say that.
+static EXITING: AtomicBool = AtomicBool::new(false);
+
+/// Called by the embedded server as it starts serving. Arms the exit gate for
+/// the eventual exit and clears one left by a previous server in this process
+/// (the shell can respawn the core task without restarting the process).
+pub fn server_starting() {
+ EXITING.store(false, Ordering::SeqCst);
+ SERVING.store(true, Ordering::SeqCst);
+}
+
+/// Whether memory is on its way out: new bindings are refused.
+pub(crate) fn exiting() -> bool {
+ EXITING.load(Ordering::SeqCst)
+}
+
+/// Test seam: put both flags back so one test's exit cannot leak into another.
+#[cfg(test)]
+pub(crate) fn reset_gate_for_tests() {
+ EXITING.store(false, Ordering::SeqCst);
+ SERVING.store(false, Ordering::SeqCst);
+}
+
+/// Ask every bound memory driver to shut down, then run the hooks the
+/// in-process engine registered with the host.
+///
+/// Providers first, concurrently and on the shared deadline: a module drains
+/// its own banked hook inside `Shutdown`, and the host's registry is where the
+/// in-process engine registers instead. The snapshot of the binding cache is
+/// complete rather than merely recent: the exit gate goes up first, and the
+/// cache checks it under its write lock before every insert, so nothing can
+/// appear after the snapshot. Both halves are idempotent — a provider's second
+/// `shutdown` is a no-op and the registry drains — so a signal landing
+/// mid-teardown, or the app-update restart path calling this twice, repeats
+/// nothing. The drivers it shut down leave the cache afterwards, so the
+/// server the shell starts next in this same process binds anew instead of
+/// reusing a driver whose workers are stopped.
+pub async fn shutdown_for_exit() {
+ if SERVING.load(Ordering::SeqCst) {
+ EXITING.store(true, Ordering::SeqCst);
+ }
+ let deadline = Instant::now() + EXIT_BUDGET;
+
+ let bindings = crate::openhuman::memory::binding::cached_bindings();
+ if !bindings.is_empty() {
+ let shutdowns = join_all(bindings.iter().map(|binding| async move {
+ let driver = binding.driver_id().to_string();
+ match binding.provider().shutdown().await {
+ Ok(()) => log::debug!("[memory:exit] driver '{driver}' shut down"),
+ Err(error) => {
+ log::warn!("[memory:exit] driver '{driver}' shutdown failed: {error}");
+ }
+ }
+ }));
+ if tokio::time::timeout_at(deadline, shutdowns).await.is_err() {
+ log::warn!(
+ "[memory:exit] driver shutdown exceeded the {EXIT_BUDGET:?} exit budget; \
+ proceeding with exit"
+ );
+ }
+ // Out of the cache either way — answered or timed out, these drivers
+ // have been told to stop. The shell restarts the embedded server in
+ // place; a server starting again in this process must bind fresh
+ // drivers, never be handed back ones whose workers exit already
+ // stopped.
+ crate::openhuman::memory::binding::evict_bindings(&bindings);
+ }
+
+ let hooks_deadline = deadline.max(Instant::now() + HOOKS_FLOOR);
+ if tokio::time::timeout_at(hooks_deadline, crate::core::shutdown::run_hooks_now())
+ .await
+ .is_err()
+ {
+ log::warn!("[memory:exit] shutdown hooks exceeded the exit budget; proceeding with exit");
+ }
+}
diff --git a/src/openhuman/memory/exit_tests.rs b/src/openhuman/memory/exit_tests.rs
new file mode 100644
index 0000000000..74172637d9
--- /dev/null
+++ b/src/openhuman/memory/exit_tests.rs
@@ -0,0 +1,120 @@
+//! Tests for the memory exit path.
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use super::binding::{cached_bindings, for_subtree};
+use super::exit::{exiting, reset_gate_for_tests, server_starting, shutdown_for_exit, EXIT_BUDGET};
+use crate::openhuman::config::schema::MemorySubsystemConfig;
+
+/// The hook registry is process-global, so the two tests below must not see
+/// each other's hooks: each registers and drains under this lock.
+static REGISTRY: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+/// With no binding ever built, the exit path is the hook registry alone — and
+/// the registry drains, so the hooks run once however many times exit is asked
+/// for (the app-update restart path asks twice).
+#[tokio::test]
+async fn exit_runs_registered_hooks_once() {
+ let _serial = REGISTRY.lock().await;
+ let runs = Arc::new(AtomicUsize::new(0));
+ let counter = Arc::clone(&runs);
+ crate::core::shutdown::register(move || {
+ let counter = Arc::clone(&counter);
+ async move {
+ counter.fetch_add(1, Ordering::SeqCst);
+ }
+ });
+
+ shutdown_for_exit().await;
+ shutdown_for_exit().await;
+
+ assert_eq!(
+ runs.load(Ordering::SeqCst),
+ 1,
+ "the registry drains on the first exit; the second finds nothing"
+ );
+}
+
+/// The gate that refuses new bindings goes up only once a real server has
+/// started — a bare exit (every unit test in this process) must never leave
+/// memory refusing to bind — and a server starting again takes it down.
+#[tokio::test]
+async fn the_exit_gate_engages_only_behind_a_server() {
+ let _serial = REGISTRY.lock().await;
+ reset_gate_for_tests();
+
+ shutdown_for_exit().await;
+ assert!(
+ !exiting(),
+ "no server ever started, so exit must not gate bindings"
+ );
+
+ server_starting();
+ shutdown_for_exit().await;
+ assert!(exiting(), "behind a server, exit refuses new bindings");
+
+ server_starting();
+ assert!(!exiting(), "a server starting again takes the gate down");
+
+ reset_gate_for_tests();
+}
+
+/// A hook that never answers must not hold the quit: exit returns within its
+/// budget (plus the hooks' floor) and the process goes on without it.
+#[tokio::test]
+async fn exit_is_bounded_even_when_a_hook_hangs() {
+ let _serial = REGISTRY.lock().await;
+ crate::core::shutdown::register(|| async {
+ tokio::time::sleep(Duration::from_secs(30)).await;
+ });
+
+ let started = std::time::Instant::now();
+ shutdown_for_exit().await;
+
+ assert!(
+ started.elapsed() < EXIT_BUDGET + Duration::from_secs(2),
+ "exit took {:?}; the hanging hook held it past the budget",
+ started.elapsed()
+ );
+}
+
+/// The shell restarts the embedded server in place (a permission refresh, an
+/// app update). Exit shuts every cached driver down, so the server that starts
+/// next must not be handed those drivers back: exit evicts them, and the next
+/// bind builds anew.
+#[tokio::test]
+async fn exit_evicts_the_drivers_it_shut_down() {
+ let _serial = REGISTRY.lock().await;
+ reset_gate_for_tests();
+ let workspace = tempfile::tempdir().expect("tempdir");
+ let cfg = MemorySubsystemConfig {
+ driver: "null".into(),
+ ..Default::default()
+ };
+ let before = for_subtree(workspace.path(), "memory", &cfg).expect("binds the null driver");
+ assert!(
+ cached_bindings()
+ .iter()
+ .any(|cached| Arc::ptr_eq(cached, &before)),
+ "the binding is cached before exit"
+ );
+
+ server_starting();
+ shutdown_for_exit().await;
+ assert!(
+ !cached_bindings()
+ .iter()
+ .any(|cached| Arc::ptr_eq(cached, &before)),
+ "exit must drop the driver it shut down"
+ );
+
+ server_starting();
+ let after = for_subtree(workspace.path(), "memory", &cfg).expect("binds again after a restart");
+ assert!(
+ !Arc::ptr_eq(&before, &after),
+ "a restarted server must not be handed back a driver exit already stopped"
+ );
+ reset_gate_for_tests();
+}
diff --git a/src/openhuman/memory/mod.rs b/src/openhuman/memory/mod.rs
index c1a68508a7..00d43ee661 100644
--- a/src/openhuman/memory/mod.rs
+++ b/src/openhuman/memory/mod.rs
@@ -36,6 +36,7 @@ pub mod agent;
pub mod api;
pub mod binding;
pub mod driver;
+pub mod exit;
pub mod guard;
pub mod host;
/// Host implementations of the seam traits the ENGINE declares.
@@ -100,6 +101,8 @@ mod bypass_allowlist_tests;
#[cfg(test)]
mod direct_engine_refs_tests;
#[cfg(test)]
+mod exit_tests;
+#[cfg(test)]
mod profile_conn_guard_tests;
#[cfg(test)]
mod seam_integration_tests;
diff --git a/src/openhuman/memory/sources/status.rs b/src/openhuman/memory/sources/status.rs
index 3563927c3a..37a3d508bf 100644
--- a/src/openhuman/memory/sources/status.rs
+++ b/src/openhuman/memory/sources/status.rs
@@ -126,7 +126,15 @@ pub(crate) fn source_id_prefix(source: &MemorySourceEntry) -> String {
match source.kind {
SourceKind::Composio => {
match (source.toolkit.as_deref(), source.connection_id.as_deref()) {
- (Some(toolkit), Some(connection_id)) => format!("{toolkit}:{connection_id}:"),
+ // Normalised exactly as the engine keys the rows
+ // (`ingest_connector_item_into_tree`: toolkit trimmed and
+ // lower-cased, connection trimmed), so a slug that ever arrives
+ // in another case still counts its own chunks instead of none.
+ (Some(toolkit), Some(connection_id)) => format!(
+ "{}:{}:",
+ toolkit.trim().to_ascii_lowercase(),
+ connection_id.trim()
+ ),
// A connection-less entry gets an unmatchable prefix, not the
// bare `{toolkit}:` — that widened prefix matched *every*
// connection of the toolkit, so a malformed or legacy Gmail
diff --git a/src/openhuman/modules/registry_part_02.rs b/src/openhuman/modules/registry_part_02.rs
index 31f347a748..66c64d0db1 100644
--- a/src/openhuman/modules/registry_part_02.rs
+++ b/src/openhuman/modules/registry_part_02.rs
@@ -91,63 +91,63 @@ const TINYCONNECTORS: ModuleRecord = ModuleRecord {
description: "OAuth connector integrations: accounts, actions, triggers, and record sync",
bus_name: "ai.tinyhumans.connectors.Composio",
object_path: "/ai/tinyhumans/connectors/Composio",
- version: "0.7.1",
- release_url: "https://github.com/tinyhumansai/tinyconnectors/releases/tag/v0.7.1",
+ version: "0.8.0",
+ release_url: "https://github.com/tinyhumansai/tinyconnectors/releases/tag/v0.8.0",
assets: &[
PlatformAsset {
host_key: "ubuntu-24.04-x86_64",
- archive: "tinyconnectors-0.7.1-ubuntu-24.04-x86_64.tar.gz",
- sha256: "31f0cfb402b0788b59d35a9f40d8a1ed514f9da0084259f61a58c03e4fe0d8ff",
+ archive: "tinyconnectors-0.8.0-ubuntu-24.04-x86_64.tar.gz",
+ sha256: "3cdd2c4b119b2da3ce0082bada68eeff89ee8554b2f1959065ca97d0eb6f1596",
},
PlatformAsset {
host_key: "ubuntu-24.04-arm64",
- archive: "tinyconnectors-0.7.1-ubuntu-24.04-arm64.tar.gz",
- sha256: "137e37be064e764585750f90dc310ea1c374aa454f46db73546a14355fc0cc97",
+ archive: "tinyconnectors-0.8.0-ubuntu-24.04-arm64.tar.gz",
+ sha256: "f2e63c9042ea75e11134b06a9ddff0a5f5d8edf24a6a300a723957cd5765c027",
},
PlatformAsset {
host_key: "ubuntu-22.04-x86_64",
- archive: "tinyconnectors-0.7.1-ubuntu-22.04-x86_64.tar.gz",
- sha256: "ca18131395fa146dc30cd662dee62b986c2a4b4bbdaa99ba17c9f62e628102cb",
+ archive: "tinyconnectors-0.8.0-ubuntu-22.04-x86_64.tar.gz",
+ sha256: "a29759a86d76b788ca58ff1a24f47de54a4fe82402e3c51161ed333a63932231",
},
PlatformAsset {
host_key: "ubuntu-22.04-arm64",
- archive: "tinyconnectors-0.7.1-ubuntu-22.04-arm64.tar.gz",
- sha256: "828878f7012897ac85b35a6caaabb5ad867aaca267afb7d15c2bd17d4a4fd907",
+ archive: "tinyconnectors-0.8.0-ubuntu-22.04-arm64.tar.gz",
+ sha256: "0cd3fd23cbf62a0ba9f4932c7d94f4cf3553e6d137b2ba3600c234486b0c18d5",
},
PlatformAsset {
host_key: "macos-26-arm64",
- archive: "tinyconnectors-0.7.1-macos-26-arm64.tar.gz",
- sha256: "64d057a49178863a2e4289b890117bd921dc915d6044e69786c90aca19878287",
+ archive: "tinyconnectors-0.8.0-macos-26-arm64.tar.gz",
+ sha256: "e9a97f70620b811ee63d458f0f72f06eb57cc3efa94bbd656088a2f961555f6a",
},
PlatformAsset {
host_key: "macos-26-x86_64",
- archive: "tinyconnectors-0.7.1-macos-26-x86_64.tar.gz",
- sha256: "4a908f1b598634bca38adaba93272ae1bd4b2b30b6472cf0f551bb36fd0637d9",
+ archive: "tinyconnectors-0.8.0-macos-26-x86_64.tar.gz",
+ sha256: "e27f2ac2f34f943dc3d3ae9fdfd3ae8b1742e94569bad6cdbfce54a52186c1d9",
},
PlatformAsset {
host_key: "macos-15-arm64",
- archive: "tinyconnectors-0.7.1-macos-15-arm64.tar.gz",
- sha256: "6c1b8ed910fd1d14bb570c4df6dbe6976605aa9c74b4551e867ace87e605520e",
+ archive: "tinyconnectors-0.8.0-macos-15-arm64.tar.gz",
+ sha256: "19c6fbc6dc5b9504424b35dd8600be7d80071bbeb5172b99588edce74d45ef5f",
},
PlatformAsset {
host_key: "macos-15-x86_64",
- archive: "tinyconnectors-0.7.1-macos-15-x86_64.tar.gz",
- sha256: "f0a13813fe460e6a8cc359b0cee6b8a854bb03c9fe0acc96c9d190a64035a5c2",
+ archive: "tinyconnectors-0.8.0-macos-15-x86_64.tar.gz",
+ sha256: "d6022e8d32834162ed269230eb50ea1e7263b6dd41d333b83f017eee17cc0bc3",
},
PlatformAsset {
host_key: "windows-2025-x86_64",
- archive: "tinyconnectors-0.7.1-windows-2025-x86_64.zip",
- sha256: "71356fb76f975736e2b7a74961eef4a2ad59842265575bfbfdb7e576836ce9df",
+ archive: "tinyconnectors-0.8.0-windows-2025-x86_64.zip",
+ sha256: "ace5366027828436ae2d3d78fe45dc4907fd4385fcc7b351e6f0cceb9dc58cdd",
},
PlatformAsset {
host_key: "windows-2022-x86_64",
- archive: "tinyconnectors-0.7.1-windows-2022-x86_64.zip",
- sha256: "5740c96c0a546f133e48ef3409550d17aee2fb4877bbd799ba1954867c29ba6a",
+ archive: "tinyconnectors-0.8.0-windows-2022-x86_64.zip",
+ sha256: "a2db1ca277c2287d26fcb0cb99ad2e2c5000cbe3992adf801dc33de7ec78d291",
},
PlatformAsset {
host_key: "windows-11-arm64",
- archive: "tinyconnectors-0.7.1-windows-11-arm64.zip",
- sha256: "f63002ffeaa7dd6c7b0fa096ab2e175bfba2d26d23eccef8b0bbd54e00171b4e",
+ archive: "tinyconnectors-0.8.0-windows-11-arm64.zip",
+ sha256: "f10a348ec43beea290ec42a836a49575cdcb970b7355712286464e90fa40f9c4",
},
],
// Lazy: a user with no connected accounts should not pay to load it, and
diff --git a/vendor/tinyconnectors b/vendor/tinyconnectors
index 402fc649ec..8a5045df3a 160000
--- a/vendor/tinyconnectors
+++ b/vendor/tinyconnectors
@@ -1 +1 @@
-Subproject commit 402fc649ecbaa78878b0dc88ea8960969d12e1bb
+Subproject commit 8a5045df3adb816fac7f0487eaacc3c84ca58fc2