From 78cd8b73a20e7414b87c53f2177f56e31100d914 Mon Sep 17 00:00:00 2001 From: Shanu Date: Thu, 10 Sep 2026 02:08:54 +0530 Subject: [PATCH] test(module): drive a source sync over the bus, not just its routing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `every_declared_method_is_actually_routed` walks the module's declared method list and checks each one dispatches. Six sync members are in that list and none was ever called. Routing is not behaviour, and the sync pipeline is what a connector-backed install spends its time doing. Measured before writing this: 143 unique declared members, 96 driven over the bus, 47 listed-only — ten of them sync. The pipeline itself is not short of tests; there are ~397 across this workspace. Every one runs in-process. What none of them reaches is the same pipeline over the bus, and that is a real gap rather than a formality: the module is a separately compiled `cdylib` with its own statics, so a task-local set host-side reads as absent inside it, and the host's sync entry points are where scope and credentials cross that boundary. The case syncs a `folder` source over a temp directory holding one file — no network, no credentials — and then asks the store, over the same bus, whether the chunks are there. That second call is the point: `records_ingested` is the run's own report, `chunks_synced` is the store's, and a member that answered a plausible `SyncRunOutcome` without writing anything satisfies the first and fails the second. Mutation-tested by removing the fixture file, which turns `records_ingested` to 0 and takes the case red. Three things the implementation found, each of which the test now pins: **The registry is not `admit_module`'s `memory_sources` config key.** That key is the serialized registry snapshot operations read; `run_source_sync` resolves ids through `get_source_in` against the host's `config.toml`. Registering in the wrong one answers `NotFound` for every source — openhuman#5820's "no memory source registered as src_…" strand, which `config.rs`'s own doc comment predicts and which this test hit first. **`RunSourceSync` writes no audit row.** It goes through `engine::run_source_pipeline`; the audit appends live in `sources::sync::sync_source`, the periodic path. Asserting an audit row here — the obvious reading of "assert its effect" — would have asserted something the member does not do. The absence is pinned instead, because the two entry points look interchangeable from the wire and a caller polling the audit log after a manual sync would wait forever. **`SourceSyncState` refuses on this engine**, with `Invalid` and a message naming the connector module as the owner. `Invalid` rather than `Unsupported` is the distinction worth keeping: the family is served, this member's subject is not. Closes #149 --- crates/tinymemory-module/tests/module_e2e.rs | 169 +++++++++++++++++++ 1 file changed, 169 insertions(+) diff --git a/crates/tinymemory-module/tests/module_e2e.rs b/crates/tinymemory-module/tests/module_e2e.rs index 524c21d..aa90bd0 100644 --- a/crates/tinymemory-module/tests/module_e2e.rs +++ b/crates/tinymemory-module/tests/module_e2e.rs @@ -77,6 +77,7 @@ const LOADER_CASES: &[&str] = &[ "every_declared_method_is_actually_routed", "stateful_optional_families_round_trip_over_the_bus", "query_and_maintenance_families_dispatch_typed_requests", + "a_source_sync_runs_over_the_bus_and_lands_in_the_store", ]; #[test] @@ -1756,3 +1757,171 @@ async fn bootstrap_connection_finds_its_provider_registry_inside_the_module() { ); } } + +/// A source sync runs through the loaded module and its result lands in the store. +/// +/// `every_declared_method_is_actually_routed` walks the module's declared +/// method list and checks each one dispatches. Six sync members are in that +/// list and, before this test, none was ever called: routing is not behaviour, +/// and the sync pipeline is what a connector-backed install spends its time +/// doing (#149). +/// +/// # Why this needs a loaded module at all +/// +/// The pipeline has ~397 tests across this workspace — `sources/sync_tests.rs`, +/// `sync/audit_tests.rs`, the `tinymemory-sync` post-processor suites — and +/// every one runs in-process. What none of them reaches is the same pipeline +/// **over the bus**, which is a real difference rather than a formality: the +/// module is a separately compiled `cdylib` with its own statics, so a +/// task-local set host-side reads as *absent* inside it, and the host's sync +/// entry points are where scope and credentials cross that boundary. +/// +/// # Hermetic by construction +/// +/// A `folder` source over a temp directory holding one file. No network, no +/// credentials, no fixture beyond `tempfile` — the same approach +/// `sources/sync_tests.rs` takes for its own folder cases. +/// +/// # Two things this got wrong first, both worth keeping +/// +/// The registry is **not** `admit_module`'s `memory_sources` config key: that +/// key is the serialized registry snapshot operations read, while +/// `run_source_sync` resolves ids through `get_source_in` against the host's +/// `config.toml`. Registering in the wrong one is openhuman#5820's "no memory +/// source registered as src_…" strand, and it is exactly what this test hit. +/// +/// And `RunSourceSync` does **not** write a sync-audit row. The audit appends +/// live in `sources::sync::sync_source`, the periodic path; this member goes +/// through `engine::run_source_pipeline`, which has none. Asserting an audit +/// row here — the obvious reading of "assert its effect" — would have been +/// asserting something the member does not do. What it does do is land chunks, +/// so that is what is checked, through a second bus call. +#[tokio::test] +#[ignore = "loads a real module; must be the only test in its process"] +async fn a_source_sync_runs_over_the_bus_and_lands_in_the_store() { + use tinymemory_api::provider::{SourceSyncState, SyncAuditEntry, SyncRunOutcome}; + + let workspace = tempfile::tempdir().expect("tempdir"); + + let source_dir = workspace.path().join("sync-fixture"); + std::fs::create_dir_all(&source_dir).expect("create source dir"); + std::fs::write( + source_dir.join("note.md"), + "# Sync fixture\n\nA folder source the module can read without a network.\n", + ) + .expect("write fixture file"); + + // `admit_module` sends no `config_path`, and the tempdir's parent holds no + // `config.toml`, so `provider::host_config_path` falls through to + // `workspace_dir/config.toml`. Written before admission, because the module + // builds its store during initialization. + std::fs::write( + workspace.path().join("config.toml"), + format!( + "[[memory_sources]]\n\ + id = \"src_sync_fixture\"\n\ + kind = \"folder\"\n\ + label = \"Sync fixture\"\n\ + enabled = true\n\ + path = \"{}\"\n", + source_dir.display() + ), + ) + .expect("write source registry"); + + let (client, _host, _task) = admit_module(workspace.path()).await; + let bus = proxy(&client); + + // ── the run ───────────────────────────────────────────────────────────── + let outcome: SyncRunOutcome = bus + .call("RunSourceSync", ("src_sync_fixture",)) + .await + .expect("RunSourceSync must reach the pipeline over the bus"); + assert_eq!( + outcome.records_ingested, 1, + "the folder holds exactly one file; got {outcome:?}" + ); + + // ── the effect, read back over the same bus ───────────────────────────── + // + // The count above is the run's own report. This is the store's, and the two + // being separate is the point: a member that returned a plausible + // `SyncRunOutcome` without writing anything would satisfy the assertion + // above and fail this one. + let statuses: Vec = bus + .call( + "SourceIngestStatus", + (vec![tinymemory_api::provider::SourceIngestQuery { + source_id: "src_sync_fixture".to_string(), + // The convention the engine's own readers write, trailing + // separator included — without it a source keyed `src_a:` also + // counts the chunks of `src_ab:`. + chunk_id_prefix: "mem_src:src_sync_fixture:".to_string(), + }],), + ) + .await + .expect("SourceIngestStatus"); + let status = statuses + .iter() + .find(|row| row.source_id == "src_sync_fixture") + .unwrap_or_else(|| panic!("no ingest status for the synced source; got {statuses:?}")); + assert!( + status.chunks_synced > 0, + "the sync reported {} records but the store holds no chunks under the \ + source prefix: {status:?}", + outcome.records_ingested + ); + + // ── the second member ─────────────────────────────────────────────────── + // + // `SourceSyncState` is driven and refuses, which is the answer worth + // pinning: this engine does not own composio connection state at all — + // "reading a composio connection's sync state is synced through the + // connector module, not this engine". A refusal that names where the answer + // lives is a better contract than a `None` that looks like "no state yet", + // and a caller polling for a cursor needs to be able to tell those apart. + // + // The class matters as much as the refusal. `Invalid` says the request was + // wrong; `Unsupported` would say the family is absent, and this driver does + // serve `SourceSync` — it just does not serve this member's subject. + let state: Result, _> = bus + .call("SourceSyncState", ("folder", "src_sync_fixture")) + .await; + let refusal = state.expect_err("this engine does not own composio connection state"); + assert!( + refusal.wire_name().ends_with("Invalid"), + "the refusal must be Invalid — the request is wrong, the family is not \ + missing — got {refusal:?}" + ); + + // ── the audit log, and what it does not contain ───────────────────────── + // + // Reachable over the bus, and empty: `RunSourceSync` goes through + // `engine::run_source_pipeline`, which appends no audit row. The rows come + // from `sources::sync::sync_source`, the periodic path. Pinned because the + // two entry points look interchangeable from the wire and are not — a + // caller that ran a manual sync and then read the audit log for it would + // wait forever. + let audit: Vec = bus + .call("SyncAuditLog", (Some(16usize),)) + .await + .expect("SyncAuditLog must be reachable"); + assert!( + !audit.iter().any(|row| row.source_id == "src_sync_fixture"), + "RunSourceSync is not an audited path; if it has become one, this test \ + should assert the row rather than its absence. Got: {audit:?}" + ); + + // ── the NotFound contract ─────────────────────────────────────────────── + // + // Documented as "deliberately distinct from a sync that ran and found + // nothing, because a caller retrying a deleted source should learn that + // rather than see an empty success". Nothing checked it. + let missing: Result = + bus.call("RunSourceSync", ("src_does_not_exist",)).await; + let error = missing.expect_err("an unregistered source id must not be an empty success"); + assert!( + error.wire_name().ends_with("NotFound"), + "an unregistered id must refuse as NotFound, got {error:?}" + ); +}