From da888f793c1474ed3e171b1cdd6b006c00b05c1f Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 18 Aug 2026 12:50:08 +0530 Subject: [PATCH 1/4] Extract the Composio normalisers into an engine-neutral crate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #18 §B3: "Payload normalisers are pure `Value -> Value` transforms with no engine dependency. Move them back into a `tinymemory-sync` crate — so a non-TinyCortex engine gets Composio sync for free." They were not in this workspace at all. They lived inside the TinyCortex engine, and `tinymemory-core` reached in through `tinycortex::memory::sync::composio::providers::normalize::*` to use them. A host binding a different memory engine therefore could not have Composio sync, despite none of this code caring which engine is bound. That is the coupling §B3 names, and it ran through the engine rather than around it. `tinymemory-sync` is fifteen files and 2,598 lines, depending on `serde_json`, two logging facades, and `chrono`. It links no engine, no storage, no async runtime — and no contract either. Two things found while moving, both stated rather than smoothed over. The crate is not quite the pure function of its input that §B3 describes. `format_email_local_time` renders in `chrono::Local`, so it reads the host's timezone, and `notion::now_ms` reads the clock. Both are deliberate upstream — the agent presents local times without doing UTC arithmetic, and Notion payloads carry no ingestion timestamp — and the raw UTC field is preserved alongside, so sorting and deduplication stay UTC-based. Documented at the crate root and at each function rather than left for someone whose output moves when they change `TZ`. The two logging facades are also inherited: `gmail_post_process` traces through `tracing`, `slack_post_process` through `log`. Preserved rather than unified, because §B3 is a move and swapping a facade changes where a host's log lines surface — a behaviour change hiding inside a relocation. The move is otherwise verbatim, with four exceptions, all forced by this workspace's lint configuration being stricter than the engine's gate reached: two `unwrap`s removed by checking presence immutably before fetching mutably, one `if let ... else { return None }` rewritten as `?`, and one `unwrap` in `ensure_object` turned into a scoped `expect` with the invariant spelled out — the case `AGENTS.md` explicitly permits. Doc links pointing at engine-internal paths are unlinked to prose, since this crate deliberately cannot see them. Acceptance, measured: `cargo tree -p tinymemory-sync` links zero of `tinycortex`, `rusqlite`, `tinymemory-core`, `tinymemory-api`. Core no longer names the engine's normalisers anywhere. The engine keeps its copy until tinyhumansai/tinycortex removes it; that side is a companion change, and the module is dead code there — its only remaining references are four doc links. Refs #18 (§B3) --- Cargo.lock | 11 + Cargo.toml | 5 +- core/Cargo.toml | 5 + .../sync/composio/providers/clickup/mod.rs | 2 +- .../src/sync/composio/providers/github/mod.rs | 2 +- core/src/sync/composio/providers/gmail/mod.rs | 2 +- core/src/sync/composio/providers/helpers.rs | 4 +- .../src/sync/composio/providers/linear/mod.rs | 2 +- core/src/sync/composio/providers/mod.rs | 2 +- .../src/sync/composio/providers/notion/mod.rs | 2 +- core/src/sync/composio/providers/slack/mod.rs | 2 +- crates/tinymemory-module/Cargo.lock | 11 + sync/Cargo.toml | 45 ++ sync/src/clickup.rs | 133 +++++ sync/src/clickup_tests.rs | 101 ++++ sync/src/github.rs | 130 +++++ sync/src/github_tests.rs | 123 +++++ sync/src/gmail_post_process.rs | 503 ++++++++++++++++++ sync/src/gmail_post_process_tests.rs | 359 +++++++++++++ sync/src/helpers.rs | 50 ++ sync/src/helpers_tests.rs | 40 ++ sync/src/lib.rs | 38 ++ sync/src/linear.rs | 157 ++++++ sync/src/linear_tests.rs | 189 +++++++ sync/src/notion.rs | 120 +++++ sync/src/notion_tests.rs | 142 +++++ sync/src/slack_post_process.rs | 323 +++++++++++ sync/src/slack_post_process_tests.rs | 262 +++++++++ 28 files changed, 2754 insertions(+), 11 deletions(-) create mode 100644 sync/Cargo.toml create mode 100644 sync/src/clickup.rs create mode 100644 sync/src/clickup_tests.rs create mode 100644 sync/src/github.rs create mode 100644 sync/src/github_tests.rs create mode 100644 sync/src/gmail_post_process.rs create mode 100644 sync/src/gmail_post_process_tests.rs create mode 100644 sync/src/helpers.rs create mode 100644 sync/src/helpers_tests.rs create mode 100644 sync/src/lib.rs create mode 100644 sync/src/linear.rs create mode 100644 sync/src/linear_tests.rs create mode 100644 sync/src/notion.rs create mode 100644 sync/src/notion_tests.rs create mode 100644 sync/src/slack_post_process.rs create mode 100644 sync/src/slack_post_process_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 5caf806..ce9b175 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1867,6 +1867,7 @@ dependencies = [ "tinycortex-api", "tinymemory", "tinymemory-api", + "tinymemory-sync", "tokio", "tracing", "url", @@ -1891,6 +1892,16 @@ dependencies = [ "tokio", ] +[[package]] +name = "tinymemory-sync" +version = "0.1.0" +dependencies = [ + "chrono", + "log", + "serde_json", + "tracing", +] + [[package]] name = "tinymemory-tinycortex" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 2d18feb..fa90487 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] -members = [".", "api", "core", "adapters/tinycortex", "adapters/remote", "conformance"] -default-members = [".", "api", "core", "adapters/tinycortex", "adapters/remote", "conformance"] +# `sync` is the engine-neutral Composio normalisers (issue #18 §B3). +members = [".", "api", "core", "sync", "adapters/tinycortex", "adapters/remote", "conformance"] +default-members = [".", "api", "core", "sync", "adapters/tinycortex", "adapters/remote", "conformance"] # `vendor/` holds engine submodules (tinycortex, tinybus, tinyagents), each of # which is its own workspace with its own lockfile. Same exclusion # `vendor/tinycortex` uses for its own nested vendor directory. diff --git a/core/Cargo.toml b/core/Cargo.toml index 628cafd..d876807 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -16,6 +16,11 @@ readme = "../README.md" # The contract. `tinymemory-core` implements and consumes it; the host seam # traits (config, event sink, embeddings, chat) live in `tinymemory_api::host`. tinymemory-api = { path = "../api" } +# Composio payload normalisers, extracted out of the engine (issue #18 §B3). +# They were reached through `tinycortex` until now, which meant a host binding a +# different engine could not have Composio sync despite none of this code +# caring which engine is bound. +tinymemory-sync = { path = "../sync" } tinymemory = { path = ".." } # The default embedded engine. `store/`, `tree/` and `sync/` drive it directly; diff --git a/core/src/sync/composio/providers/clickup/mod.rs b/core/src/sync/composio/providers/clickup/mod.rs index 807d7fe..f38484a 100644 --- a/core/src/sync/composio/providers/clickup/mod.rs +++ b/core/src/sync/composio/providers/clickup/mod.rs @@ -17,7 +17,7 @@ // The payload normalisers moved to tinycortex (they are pure Value // transforms, i.e. driver-side). Aliased under the old module name so // every `normalization::extract_*` call site below stays unchanged. -use crate::engine::backend::sync::composio::providers::normalize::clickup as normalization; +use tinymemory_sync::clickup as normalization; mod provider; #[cfg(test)] mod tests; diff --git a/core/src/sync/composio/providers/github/mod.rs b/core/src/sync/composio/providers/github/mod.rs index 58c3cd8..90633f3 100644 --- a/core/src/sync/composio/providers/github/mod.rs +++ b/core/src/sync/composio/providers/github/mod.rs @@ -16,7 +16,7 @@ // The payload normalisers moved to tinycortex (they are pure Value // transforms, i.e. driver-side). Aliased under the old module name so // every `normalization::extract_*` call site below stays unchanged. -use crate::engine::backend::sync::composio::providers::normalize::github as normalization; +use tinymemory_sync::github as normalization; mod provider; #[cfg(test)] mod tests; diff --git a/core/src/sync/composio/providers/gmail/mod.rs b/core/src/sync/composio/providers/gmail/mod.rs index f8c99a0..df067ec 100644 --- a/core/src/sync/composio/providers/gmail/mod.rs +++ b/core/src/sync/composio/providers/gmail/mod.rs @@ -1,7 +1,7 @@ // The Gmail post-processor moved to tinycortex (a pure Value transform, i.e. // driver-side). Aliased under the old module name so the single call site in // `provider.rs` stays unchanged. -use crate::engine::backend::sync::composio::providers::normalize::gmail_post_process as post_process; +use tinymemory_sync::gmail_post_process as post_process; mod provider; #[cfg(test)] mod tests; diff --git a/core/src/sync/composio/providers/helpers.rs b/core/src/sync/composio/providers/helpers.rs index d2aca25..e0358e2 100644 --- a/core/src/sync/composio/providers/helpers.rs +++ b/core/src/sync/composio/providers/helpers.rs @@ -1,11 +1,11 @@ //! Shared helpers for Composio provider implementations. //! //! `pick_str` used to live here. It is a provider payload normaliser, so it -//! moved to `crate::engine::backend::sync::composio::providers::normalize::helpers` +//! moved to `tinymemory_sync::helpers` //! and is re-exported from this module's parent. The helpers that remain are //! request-building rather than normalisation, and stay host-side. -use crate::engine::backend::sync::composio::providers::normalize::helpers::pick_str; +use tinymemory_sync::helpers::pick_str; /// Shallow-merge an `extra` JSON object into a (mutable) action-args /// object. Only object-typed extras are merged; non-object `extra` diff --git a/core/src/sync/composio/providers/linear/mod.rs b/core/src/sync/composio/providers/linear/mod.rs index af81bca..7ccce87 100644 --- a/core/src/sync/composio/providers/linear/mod.rs +++ b/core/src/sync/composio/providers/linear/mod.rs @@ -6,7 +6,7 @@ // The payload normalisers moved to tinycortex (they are pure Value // transforms, i.e. driver-side). Aliased under the old module name so // every `normalization::extract_*` call site below stays unchanged. -use crate::engine::backend::sync::composio::providers::normalize::linear as normalization; +use tinymemory_sync::linear as normalization; mod provider; #[cfg(test)] mod tests; diff --git a/core/src/sync/composio/providers/mod.rs b/core/src/sync/composio/providers/mod.rs index 7bb43ac..9637d30 100644 --- a/core/src/sync/composio/providers/mod.rs +++ b/core/src/sync/composio/providers/mod.rs @@ -281,11 +281,11 @@ pub(crate) use helpers::{first_array_str, merge_extra}; // re-exported here so the ~40 in-tree call sites keep resolving unchanged. // Note this is deliberately NOT `providers::common::pick_str`, which coerces // numbers to strings — see the doc comments on both definitions. -pub(crate) use crate::engine::backend::sync::composio::providers::normalize::helpers::pick_str; pub use registry::{ all_providers, get_provider, init_default_providers, register_provider, ProviderArc, }; pub use scope_lookup::{curated_scope_for, toolkit_has_scope}; +pub(crate) use tinymemory_sync::helpers::pick_str; pub use tool_scope::{classify_unknown, find_curated, toolkit_from_slug, CuratedTool, ToolScope}; pub use traits::{resolve_sync_interval_secs, sync_interval_env_var, ComposioProvider}; pub use types::{ diff --git a/core/src/sync/composio/providers/notion/mod.rs b/core/src/sync/composio/providers/notion/mod.rs index f781129..138fe60 100644 --- a/core/src/sync/composio/providers/notion/mod.rs +++ b/core/src/sync/composio/providers/notion/mod.rs @@ -1,7 +1,7 @@ // The payload normalisers moved to tinycortex (they are pure Value // transforms, i.e. driver-side). Aliased under the old module name so // every `normalization::extract_*` call site below stays unchanged. -use crate::engine::backend::sync::composio::providers::normalize::notion as normalization; +use tinymemory_sync::notion as normalization; mod provider; #[cfg(test)] mod tests; diff --git a/core/src/sync/composio/providers/slack/mod.rs b/core/src/sync/composio/providers/slack/mod.rs index 8d1e84e..39b745a 100644 --- a/core/src/sync/composio/providers/slack/mod.rs +++ b/core/src/sync/composio/providers/slack/mod.rs @@ -10,7 +10,7 @@ // driver-side). Re-exported under the old module name — `pub`, not a plain // `use`, because `tests/raw_coverage/memory_threads_raw_coverage_e2e.rs` // imports this path directly. -pub use crate::engine::backend::sync::composio::providers::normalize::slack_post_process as post_process; +pub use tinymemory_sync::slack_post_process as post_process; pub mod types; mod provider; diff --git a/crates/tinymemory-module/Cargo.lock b/crates/tinymemory-module/Cargo.lock index 6b52126..e9d5751 100644 --- a/crates/tinymemory-module/Cargo.lock +++ b/crates/tinymemory-module/Cargo.lock @@ -1992,6 +1992,7 @@ dependencies = [ "tinycortex-api", "tinymemory", "tinymemory-api", + "tinymemory-sync", "tokio", "tracing", "url", @@ -2022,6 +2023,16 @@ dependencies = [ "uuid", ] +[[package]] +name = "tinymemory-sync" +version = "0.1.0" +dependencies = [ + "chrono", + "log", + "serde_json", + "tracing", +] + [[package]] name = "tinymemory-tinycortex" version = "0.1.0" diff --git a/sync/Cargo.toml b/sync/Cargo.toml new file mode 100644 index 0000000..223e667 --- /dev/null +++ b/sync/Cargo.toml @@ -0,0 +1,45 @@ +[package] +name = "tinymemory-sync" +publish = false +version = "0.1.0" +edition = "2021" +rust-version = "1.96" +license = "MIT" +repository = "https://github.com/tinyhumansai/tinymemory" +description = "Engine-neutral Composio payload normalisers for TinyMemory" + +# The whole dependency list, and it is the point of the crate. These are pure +# `Value -> Value` transforms: no engine, no storage, no network, no async +# runtime. A dependency added here should have to argue for itself against that +# sentence (issue #18 §B3). +[dependencies] +serde_json = "1" +# Two logging facades, neither an implementation, both carried over from the +# engine layout this crate was extracted from: `gmail_post_process` traces +# through `tracing`, `slack_post_process` through `log`. Preserved rather than +# unified, because §B3 is a *move* and swapping a facade would change where a +# host's log lines surface — a behaviour change hiding inside a relocation. +# Worth reconciling in its own change. +tracing = "0.1" +log = "0.4" +# RFC 2822/3339 date handling for Gmail `Date:` headers. +# +# `clock` is on, and it is the one place this crate is not a pure function of +# its input: `format_email_local_time` renders in `chrono::Local`, so it reads +# the host's timezone. That is deliberate upstream — the agent presents local +# times without doing UTC arithmetic, and the raw UTC field is preserved +# alongside — but it means "pure `Value -> Value`" is true of every normaliser +# here except that one. Better said out loud than discovered by someone whose +# output moved when they changed TZ. +chrono = { version = "0.4", features = ["clock"] } + +[lints.rust] +unsafe_code = "forbid" +missing_docs = "warn" +unreachable_pub = "warn" + +[lints.clippy] +all = { level = "warn", priority = -1 } +unwrap_used = "warn" +expect_used = "warn" +panic = "warn" diff --git a/sync/src/clickup.rs b/sync/src/clickup.rs new file mode 100644 index 0000000..ae340b6 --- /dev/null +++ b/sync/src/clickup.rs @@ -0,0 +1,133 @@ +//! ClickUp host normalization helpers — result extraction, task-title extraction, +//! and time utilities. +//! +//! ClickUp's REST API (and therefore Composio's wrapping of it) returns +//! task lists in a small handful of shapes depending on which endpoint +//! is called. The functions here walk the union of common shapes so the +//! provider doesn't have to branch per Composio envelope variant. + +use serde_json::Value; + +use super::helpers::pick_str; + +/// Walk the Composio response envelope for ClickUp task list results. +/// +/// ClickUp's "filtered team tasks" endpoint returns `{ "tasks": [...] }` +/// at the top level; Composio re-wraps the upstream payload under +/// `data` or `data.data` depending on the action. We probe each shape +/// in order and return the first array we find. +pub fn extract_tasks(data: &Value) -> Vec { + let candidates = [ + data.pointer("/data/tasks"), + data.pointer("/tasks"), + data.pointer("/data/data/tasks"), + data.pointer("/data/results"), + data.pointer("/results"), + data.pointer("/data/items"), + data.pointer("/items"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(arr) = cand.as_array() { + return arr.clone(); + } + } + Vec::new() +} + +/// Extract a human-readable title from a ClickUp task object. +/// +/// ClickUp tasks store the name at `name` (or `data.name` after Composio +/// envelope wrapping). When the name is missing we fall back to the +/// task ID so chunks remain identifiable. +pub fn extract_task_name(task: &Value) -> Option { + pick_str(task, &["name", "data.name", "title", "data.title"]) +} + +/// Extract a stable cursor timestamp (milliseconds since epoch as a +/// string) from a ClickUp task object. +/// +/// The ClickUp API returns `date_updated` as a stringified epoch ms +/// (e.g. `"1733412345678"`); we keep it as a string so lexicographic +/// comparison against the stored cursor remains valid as long as the +/// length doesn't change (it won't until year 33658). +pub fn extract_task_updated(task: &Value) -> Option { + pick_str( + task, + &[ + "date_updated", + "data.date_updated", + "updated_at", + "data.updated_at", + "dateUpdated", + "data.dateUpdated", + ], + ) +} + +/// Current wall-clock time in milliseconds since the UNIX epoch. +pub fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +/// Extract the authorized user's numeric ID from the +/// `CLICKUP_GET_AUTHORIZED_USER` response. +/// +/// Composio wraps the upstream `{"user": {"id": …}}` shape; this walker +/// is defensive against both raw and wrapped payloads. Returns the ID +/// as a string because `CLICKUP_GET_FILTERED_TEAM_TASKS` accepts the +/// `assignees` filter as a string array. +pub fn extract_user_id(data: &Value) -> Option { + let candidates = [ + data.pointer("/user/id"), + data.pointer("/data/user/id"), + data.pointer("/id"), + data.pointer("/data/id"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(n) = cand.as_u64() { + return Some(n.to_string()); + } + if let Some(n) = cand.as_i64() { + return Some(n.to_string()); + } + if let Some(s) = cand.as_str() { + let trimmed = s.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + None +} + +/// Extract a list of workspace (team) IDs from the +/// `CLICKUP_GET_AUTHORIZED_TEAMS_WORKSPACES` response. +/// +/// ClickUp returns `{"teams": [{"id": "...", "name": "..."}, …]}`. We +/// keep the IDs as strings — `CLICKUP_GET_FILTERED_TEAM_TASKS` requires +/// a `team_id` (string) argument. +pub fn extract_workspace_ids(data: &Value) -> Vec { + let candidates = [ + data.pointer("/teams"), + data.pointer("/data/teams"), + data.pointer("/workspaces"), + data.pointer("/data/workspaces"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(arr) = cand.as_array() { + return arr + .iter() + .filter_map(|t| pick_str(t, &["id", "team_id", "workspace_id"])) + .collect(); + } + } + Vec::new() +} + +#[cfg(test)] +#[path = "clickup_tests.rs"] +mod tests; diff --git a/sync/src/clickup_tests.rs b/sync/src/clickup_tests.rs new file mode 100644 index 0000000..99b4667 --- /dev/null +++ b/sync/src/clickup_tests.rs @@ -0,0 +1,101 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +// +// A failing assertion in a test *is* a panic. The crate-wide lints exist to +// keep the library from panicking, not the tests. + +use super::*; +use serde_json::json; + +#[test] +fn extract_tasks_from_data_tasks() { + let data = json!({ "data": { "tasks": [{"id": "t1"}] } }); + assert_eq!(extract_tasks(&data).len(), 1); +} + +#[test] +fn extract_tasks_from_top_level_tasks() { + let data = json!({ "tasks": [{"id": "a"}, {"id": "b"}] }); + assert_eq!(extract_tasks(&data).len(), 2); +} + +#[test] +fn extract_tasks_empty_when_missing() { + let data = json!({ "foo": "bar" }); + assert!(extract_tasks(&data).is_empty()); +} + +#[test] +fn extract_task_name_from_top_level() { + let task = json!({ "id": "t1", "name": "Build feature X" }); + assert_eq!(extract_task_name(&task), Some("Build feature X".into())); +} + +#[test] +fn extract_task_name_falls_back_to_data_name() { + let task = json!({ "data": { "name": "Wrapped" } }); + assert_eq!(extract_task_name(&task), Some("Wrapped".into())); +} + +#[test] +fn extract_task_name_none_when_missing() { + let task = json!({ "id": "t1" }); + assert!(extract_task_name(&task).is_none()); +} + +#[test] +fn extract_task_updated_handles_string_form() { + let task = json!({ "date_updated": "1733412345678" }); + assert_eq!( + extract_task_updated(&task), + Some("1733412345678".to_string()) + ); +} + +#[test] +fn extract_task_updated_handles_nested_data() { + let task = json!({ "data": { "dateUpdated": "1700000000000" } }); + assert_eq!( + extract_task_updated(&task), + Some("1700000000000".to_string()) + ); +} + +#[test] +fn extract_user_id_handles_numeric_id() { + let data = json!({ "user": { "id": 12345 } }); + assert_eq!(extract_user_id(&data), Some("12345".to_string())); +} + +#[test] +fn extract_user_id_handles_wrapped_payload() { + let data = json!({ "data": { "user": { "id": "777" } } }); + assert_eq!(extract_user_id(&data), Some("777".to_string())); +} + +#[test] +fn extract_user_id_none_when_missing() { + let data = json!({ "foo": "bar" }); + assert!(extract_user_id(&data).is_none()); +} + +#[test] +fn extract_workspace_ids_from_teams_array() { + let data = json!({ + "teams": [ + { "id": "ws1", "name": "Personal" }, + { "id": "ws2", "name": "Acme" }, + ] + }); + assert_eq!(extract_workspace_ids(&data), vec!["ws1", "ws2"]); +} + +#[test] +fn extract_workspace_ids_empty_when_no_teams() { + let data = json!({ "foo": "bar" }); + assert!(extract_workspace_ids(&data).is_empty()); +} + +#[test] +fn now_ms_returns_nonzero() { + assert!(now_ms() > 0); +} diff --git a/sync/src/github.rs b/sync/src/github.rs new file mode 100644 index 0000000..393e93e --- /dev/null +++ b/sync/src/github.rs @@ -0,0 +1,130 @@ +//! GitHub host normalization helpers — result extraction, identity helpers, and time utilities. +//! +//! GitHub's REST API (proxied through Composio) returns search results and +//! authenticated-user payloads in a small number of shapes. The functions here +//! walk the union of common Composio envelope variants so the provider stays +//! clean and branch-free. + +use serde_json::Value; + +use super::helpers::pick_str; + +/// Walk the Composio response envelope for GitHub search issue results. +/// +/// `GITHUB_SEARCH_ISSUES_AND_PULL_REQUESTS` wraps GitHub's `GET /search/issues` response, which +/// returns `{"total_count": N, "items": [...]}`. Composio may re-wrap this under +/// `data` or `data.data`; we probe each shape in order. +pub fn extract_issues(data: &Value) -> Vec { + let candidates = [ + data.pointer("/data/items"), + data.pointer("/items"), + data.pointer("/data/data/items"), + data.pointer("/data/results"), + data.pointer("/results"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(arr) = cand.as_array() { + return arr.clone(); + } + } + Vec::new() +} + +/// Extract a stable, globally unique identifier for a GitHub issue or PR. +/// +/// GitHub's internal `id` field is a large integer unique across all issues +/// and PRs on github.com. We convert it to a string for use as a sync key. +/// Falls back to composing from `html_url` path if `id` is absent. +pub fn extract_issue_id(issue: &Value) -> Option { + // Primary: numeric internal GitHub ID. + if let Some(id) = issue.get("id").or_else(|| issue.pointer("/data/id")) { + if let Some(n) = id.as_u64() { + return Some(n.to_string()); + } + if let Some(s) = id.as_str() { + let trimmed = s.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + // Fallback: parse owner/repo/number from html_url path segments. + // URL shape: https://github.com/{owner}/{repo}/issues/{number} + if let Some(url) = pick_str(issue, &["html_url", "data.html_url", "url", "data.url"]) { + if let Some(slug) = github_url_to_slug(&url) { + return Some(slug); + } + } + None +} + +/// Build a human-readable document title for a GitHub issue/PR. +/// +/// Format: `GitHub: {owner}/{repo}#{number}: {title}`. +/// Falls back to just the title or a placeholder when fields are missing. +pub fn extract_issue_title(issue: &Value) -> Option { + let title = pick_str(issue, &["title", "data.title"])?; + + // Best-effort: extract owner/repo#N from html_url for the prefix. + let prefix = pick_str(issue, &["html_url", "data.html_url"]) + .and_then(|url| github_url_to_slug(&url)) + .unwrap_or_default(); + + if prefix.is_empty() { + Some(title) + } else { + Some(format!("GitHub: {prefix}: {title}")) + } +} + +/// Parse `https://github.com/{owner}/{repo}/issues/{number}` (or `/pull/`) +/// into `"{owner}/{repo}#{number}"`. Returns `None` for unrecognised shapes. +fn github_url_to_slug(url: &str) -> Option { + let segs: Vec<&str> = url.trim_end_matches('/').split('/').collect(); + // Minimum: ["https:", "", "github.com", owner, repo, "issues", number] + if segs.len() >= 7 { + let number = segs[segs.len() - 1]; + let _kind = segs[segs.len() - 2]; // "issues" or "pull" — ignored + let repo = segs[segs.len() - 3]; + let owner = segs[segs.len() - 4]; + if !owner.is_empty() && !repo.is_empty() && !number.is_empty() { + return Some(format!("{owner}/{repo}#{number}")); + } + } + None +} + +/// Extract the `updated_at` ISO 8601 timestamp from a GitHub issue. +/// +/// GitHub returns `updated_at` as `"2024-05-21T15:30:00Z"`. ISO 8601 strings +/// sort lexicographically, so we use them directly as the sync cursor. +pub fn extract_issue_updated_at(issue: &Value) -> Option { + pick_str( + issue, + &[ + "updated_at", + "data.updated_at", + "updatedAt", + "data.updatedAt", + ], + ) +} + +/// Extract the authenticated user's login handle from a +/// `GITHUB_GET_THE_AUTHENTICATED_USER` response. +pub fn extract_user_login(data: &Value) -> Option { + pick_str(data, &["login", "data.login"]) +} + +/// Current wall-clock time in milliseconds since the UNIX epoch. +pub fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +#[cfg(test)] +#[path = "github_tests.rs"] +mod tests; diff --git a/sync/src/github_tests.rs b/sync/src/github_tests.rs new file mode 100644 index 0000000..17aa138 --- /dev/null +++ b/sync/src/github_tests.rs @@ -0,0 +1,123 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +// +// A failing assertion in a test *is* a panic. The crate-wide lints exist to +// keep the library from panicking, not the tests. + +use super::*; +use serde_json::json; + +#[test] +fn extract_issues_from_data_items() { + let data = json!({ "data": { "items": [{"id": 1}] } }); + assert_eq!(extract_issues(&data).len(), 1); +} + +#[test] +fn extract_issues_from_top_level_items() { + let data = json!({ "items": [{"id": 1}, {"id": 2}] }); + assert_eq!(extract_issues(&data).len(), 2); +} + +#[test] +fn extract_issues_empty_when_missing() { + let data = json!({ "foo": "bar" }); + assert!(extract_issues(&data).is_empty()); +} + +#[test] +fn extract_issue_id_from_numeric_field() { + let issue = json!({ "id": 123456789u64, "title": "Fix bug" }); + assert_eq!(extract_issue_id(&issue), Some("123456789".to_string())); +} + +#[test] +fn extract_issue_id_from_wrapped_data() { + let issue = json!({ "data": { "id": 99u64 } }); + assert_eq!(extract_issue_id(&issue), Some("99".to_string())); +} + +#[test] +fn extract_issue_id_falls_back_to_html_url() { + let issue = json!({ + "html_url": "https://github.com/owner/repo/issues/42" + }); + assert_eq!(extract_issue_id(&issue), Some("owner/repo#42".to_string())); +} + +#[test] +fn extract_issue_id_none_when_missing() { + let issue = json!({ "title": "No ID here" }); + assert!(extract_issue_id(&issue).is_none()); +} + +#[test] +fn extract_issue_title_builds_prefixed_title() { + let issue = json!({ + "id": 1u64, + "title": "Fix race condition", + "html_url": "https://github.com/acme/core/issues/99" + }); + assert_eq!( + extract_issue_title(&issue), + Some("GitHub: acme/core#99: Fix race condition".to_string()) + ); +} + +#[test] +fn extract_issue_title_returns_raw_title_when_no_url() { + let issue = json!({ "title": "Bare title" }); + assert_eq!(extract_issue_title(&issue), Some("Bare title".to_string())); +} + +#[test] +fn extract_issue_title_none_when_missing() { + let issue = json!({ "id": 1u64 }); + assert!(extract_issue_title(&issue).is_none()); +} + +#[test] +fn extract_issue_updated_at_from_top_level() { + let issue = json!({ "updated_at": "2024-05-21T15:30:00Z" }); + assert_eq!( + extract_issue_updated_at(&issue), + Some("2024-05-21T15:30:00Z".to_string()) + ); +} + +#[test] +fn extract_issue_updated_at_from_data_wrapper() { + let issue = json!({ "data": { "updated_at": "2023-01-01T00:00:00Z" } }); + assert_eq!( + extract_issue_updated_at(&issue), + Some("2023-01-01T00:00:00Z".to_string()) + ); +} + +#[test] +fn extract_issue_updated_at_none_when_missing() { + let issue = json!({ "id": 1u64 }); + assert!(extract_issue_updated_at(&issue).is_none()); +} + +#[test] +fn extract_user_login_from_top_level() { + let data = json!({ "login": "octocat" }); + assert_eq!(extract_user_login(&data), Some("octocat".to_string())); +} + +#[test] +fn extract_user_login_from_data_wrapper() { + let data = json!({ "data": { "login": "monalisa" } }); + assert_eq!(extract_user_login(&data), Some("monalisa".to_string())); +} + +#[test] +fn extract_user_login_none_when_missing() { + let data = json!({ "id": 1u64 }); + assert!(extract_user_login(&data).is_none()); +} + +#[test] +fn now_ms_returns_nonzero() { + assert!(now_ms() > 0); +} diff --git a/sync/src/gmail_post_process.rs b/sync/src/gmail_post_process.rs new file mode 100644 index 0000000..327479a --- /dev/null +++ b/sync/src/gmail_post_process.rs @@ -0,0 +1,503 @@ +//! Gmail-specific post-processing of Composio action responses. +//! +//! The upstream `GMAIL_FETCH_EMAILS` payload is extremely verbose +//! (full MIME tree under `payload.parts[]`, 50+ `Received:` headers, +//! display-layer noise the model never uses). This module rewrites +//! it into a slim envelope per message: +//! +//! ```json +//! { +//! "messages": [ +//! { +//! "id": "…", +//! "threadId": "…", +//! "subject": "…", +//! "from": "…", +//! "to": "…", +//! "date": "…", +//! "labels": ["INBOX", "UNREAD"], +//! "markdown": "…body…", +//! "attachments": [ { "filename": "...", "mimeType": "..." } ] +//! } +//! ], +//! "nextPageToken": "…", +//! "resultSizeEstimate": 201 +//! } +//! ``` +//! +//! ## Body source +//! +//! Composio's backend ships a +//! `markdownFormatted` field on the response envelope — one string +//! per tool call, pre-rendered with HTML stripped, URLs shortened, +//! footers removed, whitespace normalised. We split it per message +//! along `\n---\n` boundaries (with `## ` heading fallbacks) and +//! pin each slice to the corresponding entry in `messages[]` via +//! [`apply_response_level_markdown`]. The reshape's +//! `extract_markdown_body` then prefers that pinned field over +//! falling back to the upstream `messageText`. +//! +//! No in-house HTML→markdown conversion lives here anymore — the +//! backend does the cleaning. If `markdownFormatted` is absent for +//! a given response we fall through to whatever plain text the +//! upstream provided in `messageText`. +//! +//! Callers that need the raw Composio shape can pass `raw_html: +//! true` (or `rawHtml: true`) in the action arguments — this +//! short-circuits the reshape entirely. +//! +//! Only `GMAIL_FETCH_EMAILS` is reshaped today; other Gmail action +//! responses are passed through unchanged. When we add envelopes for +//! more slugs they should live in this file, branched from +//! [`post_process`]. + +use serde_json::{json, Map, Value}; + +/// Entry point called from `GmailProvider::post_process_action_result`. +/// +/// Dispatches on the Composio action slug. Unknown Gmail slugs fall +/// through to a no-op. +pub fn post_process(slug: &str, arguments: Option<&Value>, data: &mut Value) { + if is_raw_html_flag_set(arguments) { + tracing::debug!( + slug, + "[composio:gmail][post-process] raw_html flag set, passing through" + ); + return; + } + if slug == "GMAIL_FETCH_EMAILS" { + reshape_fetch_emails(data) + } +} + +/// Stash per-message slices of the response-level `markdownFormatted` +/// onto the corresponding entries inside `data.messages[]`. +/// +/// The Composio backend (tinyhumansai/backend#683) ships ONE +/// `markdownFormatted` string per tool call covering all messages — +/// already URL-shortened, footer-stripped, and whitespace-normalised. +/// To get per-email files in the raw archive we split that string +/// along section boundaries (`## ` headings or `---` rules) and pin +/// each slice to the message at the same index. `extract_markdown_body` +/// then prefers `msg.markdownFormatted` over re-decoding the MIME +/// tree. +/// +/// **Must be called BEFORE [`post_process`]** because `post_process` +/// reshapes `data` into the slim envelope; once `messages[]` carries +/// our slim shape the upstream message ordering is already locked in +/// but we may have lost original ordering signals if any. +/// +/// No-op when the slice count doesn't match `messages.len()` — we +/// can't safely align segments to messages without an exact match, +/// so we let `extract_markdown_body` fall through to its MIME path. +pub fn apply_response_level_markdown(data: &mut Value, top_md: &str) { + let trimmed = top_md.trim(); + if trimmed.is_empty() { + return; + } + // Presence is checked immutably first, then fetched mutably. The original + // form re-fetched with `unwrap()` after a mutable probe, which is sound but + // relies on the reader to see why; this crate forbids `unwrap`, and the + // immutable probe expresses the same reasoning to the compiler. + let container = if data.get("messages").is_some() { + data + } else if data.get("data").and_then(Value::as_object).is_some() { + match data.get_mut("data") { + Some(inner) => inner, + None => return, + } + } else { + tracing::debug!( + "[composio:gmail][post-process] apply_response_level_markdown: \ + no messages container in response — skipping" + ); + return; + }; + let Some(messages) = container.get_mut("messages").and_then(|v| v.as_array_mut()) else { + return; + }; + let count = messages.len(); + if count == 0 { + return; + } + // Clone hints out of the messages array so the slice borrows + // don't conflict with the upcoming `messages.iter_mut()` mutation. + let hints: Vec = messages.clone(); + let Some(slices) = split_response_markdown_per_message_with_hint(trimmed, count, Some(&hints)) + else { + tracing::debug!( + messages = count, + md_len = trimmed.len(), + "[composio:gmail][post-process] could not split response-level markdownFormatted \ + into {count} slices — falling back to per-message MIME decode" + ); + return; + }; + for (msg, slice) in messages.iter_mut().zip(slices) { + if let Some(obj) = msg.as_object_mut() { + obj.insert("markdownFormatted".to_string(), Value::String(slice)); + } + } + tracing::debug!( + messages = count, + "[composio:gmail][post-process] stashed per-message markdownFormatted slices" + ); +} + +/// Split a top-level `markdownFormatted` string into per-message +/// segments. Returns `Some(slices)` only when the split yields +/// exactly `expected_count` entries — otherwise the format isn't one +/// of the patterns we know about and we let the caller fall back. +/// +/// Primary boundary is the `\n---\n` horizontal rule the backend +/// emits between messages (confirmed against real +/// `GMAIL_FETCH_EMAILS` output). H2/H3 headings are kept as +/// fallbacks for older renderings. The preamble (`# Inbox (N +/// messages)`-style intro, if present) is dropped — we accept +/// either `expected` parts (no preamble) or `expected + 1` +/// (preamble + N messages). +/// +/// `messages_hint` is the slim message array from the same response +/// — when present we use the per-message `subject` field to verify +/// each segment really does belong to the message at the same index. +/// Mismatches force a fallback so we never write a wrong-message body +/// to the raw archive. +pub fn split_response_markdown_per_message(md: &str, expected_count: usize) -> Option> { + split_response_markdown_per_message_with_hint(md, expected_count, None) +} + +/// Split a response-level markdown blob into one slice per message. +/// +/// `hint` carries the message ids in response order, which is what makes the +/// split reliable: the blob's own section headings are backend-rendered and +/// have changed shape between versions, so matching on them alone silently +/// mis-attributed bodies. +pub fn split_response_markdown_per_message_with_hint( + md: &str, + expected_count: usize, + messages_hint: Option<&[Value]>, +) -> Option> { + if expected_count == 0 { + return None; + } + if expected_count == 1 { + return Some(vec![md.to_string()]); + } + + // Boundary patterns to try, in priority order. `\n---\n` is the + // confirmed marker; the heading variants stay as belt-and-braces + // for older / variant backend renderings. + let candidates: &[(&str, &str)] = &[ + ("\n---\n", "---\n"), + ("\n\n## ", "## "), + ("\n\n### ", "### "), + ("\n\n# ", "# "), + ("\n***\n", "***\n"), + ]; + + for (sep, prefix) in candidates { + let parts: Vec<&str> = md.split(sep).collect(); + let (drop_preamble, prepend_first) = if parts.len() == expected_count { + (false, false) // no preamble; first segment had no prefix + } else if parts.len() == expected_count + 1 { + (true, true) // preamble dropped; every kept segment had a prefix + } else { + continue; + }; + let segments: Vec = parts + .into_iter() + .skip(if drop_preamble { 1 } else { 0 }) + .enumerate() + .map(|(i, s)| { + if i == 0 && !prepend_first { + s.to_string() + } else { + format!("{prefix}{s}") + } + }) + .collect(); + + // Validate alignment against the JSON message array: every + // segment whose corresponding message has a non-empty subject + // must mention that subject somewhere in its body. If a single + // pair fails, we treat the split as unreliable and try the + // next pattern. Empty / null subjects skip validation (e.g. + // notification mails where the subject is ""). + if let Some(hints) = messages_hint { + if !validate_segments_against_hints(&segments, hints) { + tracing::debug!( + expected = expected_count, + sep = sep, + "[composio:gmail][post-process] split candidate failed subject check" + ); + continue; + } + } + return Some(segments); + } + None +} + +/// True if every (segment, message) pair where the message has a +/// non-empty subject contains that subject somewhere in the segment +/// (case-insensitive substring match — a defensive heuristic, not a +/// strict equality check, since the backend may format subjects +/// inside markdown links or with surrounding decoration). +fn validate_segments_against_hints(segments: &[String], hints: &[Value]) -> bool { + if segments.len() != hints.len() { + return false; + } + for (seg, hint) in segments.iter().zip(hints.iter()) { + let subject = hint + .get("subject") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim(); + if subject.is_empty() { + continue; + } + if !seg + .to_ascii_lowercase() + .contains(&subject.to_ascii_lowercase()) + { + return false; + } + } + true +} + +/// Returns true when the caller explicitly set `raw_html: true` (or the +/// camelCase `rawHtml: true`) in the `arguments` object. +fn is_raw_html_flag_set(arguments: Option<&Value>) -> bool { + let Some(obj) = arguments.and_then(|v| v.as_object()) else { + return false; + }; + obj.get("raw_html") + .or_else(|| obj.get("rawHtml")) + .and_then(|v| v.as_bool()) + .unwrap_or(false) +} + +/// Rewrite a `GMAIL_FETCH_EMAILS` `data` object in place into the slim +/// envelope documented at the module level. +/// +/// The Composio response can be shaped either as `{ messages, nextPageToken, ... }` +/// directly, or wrapped one level deeper under `{ data: { messages: … } }` +/// depending on backend version; we handle both. +fn reshape_fetch_emails(data: &mut Value) { + // Unwrap an optional `data:` envelope so downstream logic only has + // to deal with one shape. + let container = if data.get("messages").is_some() { + data + } else if data.get("data").and_then(Value::as_object).is_some() { + match data.get_mut("data") { + Some(inner) => inner, + None => return, + } + } else { + return; + }; + + let Some(obj) = container.as_object_mut() else { + return; + }; + + let raw_messages = obj + .remove("messages") + .and_then(|v| match v { + Value::Array(arr) => Some(arr), + _ => None, + }) + .unwrap_or_default(); + let next_page_token = obj.remove("nextPageToken").unwrap_or(Value::Null); + let result_size_estimate = obj.remove("resultSizeEstimate").unwrap_or(Value::Null); + + let messages: Vec = raw_messages.into_iter().map(reshape_message).collect(); + + let mut envelope = Map::new(); + envelope.insert("messages".into(), Value::Array(messages)); + if !next_page_token.is_null() { + envelope.insert("nextPageToken".into(), next_page_token); + } + if !result_size_estimate.is_null() { + envelope.insert("resultSizeEstimate".into(), result_size_estimate); + } + + *container = Value::Object(envelope); +} + +/// Parse an RFC 3339 or RFC 2822 date string into a UTC `DateTime`. +pub fn parse_email_date(date_str: &str) -> Option> { + date_str + .parse::>() + .or_else(|_| { + chrono::DateTime::parse_from_rfc2822(date_str).map(|d| d.with_timezone(&chrono::Utc)) + }) + .ok() +} + +const EMAIL_LOCAL_TIME_FMT: &str = "%Y-%m-%d %I:%M %p %:z"; + +/// Format a UTC `DateTime` in the given timezone. Returns `None` when the +/// formatted result is identical to the UTC rendering (no-op for UTC hosts). +pub fn format_at_tz( + utc: chrono::DateTime, + tz: &Tz, +) -> Option +where + Tz::Offset: std::fmt::Display, +{ + let local_dt = utc.with_timezone(tz); + let formatted = local_dt.format(EMAIL_LOCAL_TIME_FMT).to_string(); + + let utc_formatted = utc.format(EMAIL_LOCAL_TIME_FMT).to_string(); + if formatted == utc_formatted { + return None; + } + Some(formatted) +} + +/// Convert a UTC email timestamp string to a human-readable local-time string. +/// +/// Accepts RFC 3339 (`"2026-05-31T10:33:00Z"`) or RFC 2822 +/// (`"Sat, 31 May 2026 10:33:00 +0000"`) input. Returns a formatted string +/// in the host's local timezone, e.g. `"2026-05-31 05:33 AM -05:00"`, +/// so the agent can present local times without UTC arithmetic. +/// +/// The raw `date` field is always preserved alongside this field so +/// internal sorting, deduplication, and debugging remain UTC-based. +/// +/// Returns `None` when the input cannot be parsed or the output format +/// would be identical to the UTC input (no-op for UTC hosts). +pub fn format_email_local_time(date_str: &str) -> Option { + let utc = parse_email_date(date_str)?; + format_at_tz(utc, &chrono::Local) +} + +/// Map one raw Composio message object to its slim counterpart. +/// +/// Body source picked by [`extract_markdown_body`]: +/// 1. The per-message `markdownFormatted` slice pinned by +/// [`apply_response_level_markdown`] (preferred — backend-rendered). +/// 2. The upstream `messageText` plaintext (fallback). +/// 3. Empty string. +fn reshape_message(raw: Value) -> Value { + let Value::Object(obj) = raw else { + return raw; + }; + + let id = obj.get("messageId").cloned().unwrap_or(Value::Null); + let thread_id = obj.get("threadId").cloned().unwrap_or(Value::Null); + let subject = obj.get("subject").cloned().unwrap_or(Value::Null); + let sender = obj.get("sender").cloned().unwrap_or(Value::Null); + let to = obj.get("to").cloned().unwrap_or(Value::Null); + let date = obj + .get("messageTimestamp") + .cloned() + .or_else(|| pick_header(&obj, "Date")) + .unwrap_or(Value::Null); + let labels = obj + .get("labelIds") + .cloned() + .unwrap_or_else(|| Value::Array(Vec::new())); + let list_unsubscribe = pick_header(&obj, "List-Unsubscribe").unwrap_or(Value::Null); + + let markdown = extract_markdown_body(&obj); + let attachments = extract_attachments(&obj); + + // Compute a local-time representation of the UTC `date` so the agent + // presents times in the user's timezone rather than quoting raw UTC. + let date_local = date.as_str().and_then(format_email_local_time); + + let mut out = Map::new(); + out.insert("id".into(), id); + out.insert("threadId".into(), thread_id); + out.insert("subject".into(), subject); + out.insert("from".into(), sender); + out.insert("to".into(), to); + out.insert("date".into(), date); + if let Some(local) = date_local { + out.insert("date_local".into(), Value::String(local)); + } + out.insert("labels".into(), labels); + if !list_unsubscribe.is_null() { + out.insert("list_unsubscribe".into(), list_unsubscribe); + } + out.insert("markdown".into(), Value::String(markdown)); + if !attachments.is_empty() { + out.insert("attachments".into(), Value::Array(attachments)); + } + Value::Object(out) +} + +/// Find a header value by (case-insensitive) name in the Composio +/// `payload.headers[]` array. Returns `Some(Value::String)` on hit. +fn pick_header(msg: &Map, name: &str) -> Option { + let headers = msg.get("payload")?.get("headers")?.as_array()?; + for h in headers { + let hn = h.get("name").and_then(|v| v.as_str()).unwrap_or(""); + if hn.eq_ignore_ascii_case(name) { + if let Some(v) = h.get("value").and_then(|v| v.as_str()) { + return Some(Value::String(v.to_string())); + } + } + } + None +} + +/// Pick a body for the slim envelope. +/// +/// We trust the Composio backend's pre-rendered `markdownFormatted` +/// (set per-message by [`apply_response_level_markdown`] from the +/// response-level field). When that's absent we fall back to the +/// upstream's plain-text `messageText` verbatim — no in-house +/// HTML→markdown decoding lives here anymore. The backend already +/// strips HTML, shortens URLs, and normalises whitespace; running +/// our own pipeline on top duplicated work and corrupted some +/// renderings. +fn extract_markdown_body(msg: &Map) -> String { + if let Some(formatted) = msg + .get("markdownFormatted") + .or_else(|| msg.get("markdown_formatted")) + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + { + return formatted.to_string(); + } + if let Some(text) = msg + .get("messageText") + .and_then(|v| v.as_str()) + .map(str::trim) + .filter(|s| !s.is_empty()) + { + return text.to_string(); + } + String::new() +} + +/// Pull a minimal attachments descriptor from the Composio +/// `attachmentList` array. +fn extract_attachments(msg: &Map) -> Vec { + if let Some(list) = msg.get("attachmentList").and_then(|v| v.as_array()) { + return list + .iter() + .filter_map(|a| { + let filename = a.get("filename").and_then(|v| v.as_str())?; + if filename.is_empty() { + return None; + } + let mime = a + .get("mimeType") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + Some(json!({ "filename": filename, "mimeType": mime })) + }) + .collect(); + } + Vec::new() +} + +#[cfg(test)] +#[path = "gmail_post_process_tests.rs"] +mod tests; diff --git a/sync/src/gmail_post_process_tests.rs b/sync/src/gmail_post_process_tests.rs new file mode 100644 index 0000000..69eda63 --- /dev/null +++ b/sync/src/gmail_post_process_tests.rs @@ -0,0 +1,359 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +// +// A failing assertion in a test *is* a panic. The crate-wide lints exist to +// keep the library from panicking, not the tests. + +use super::*; +use serde_json::json; + +fn fixture_with_backend_markdown() -> Value { + json!({ + "messages": [ + { + "messageId": "m1", + "threadId": "t1", + "subject": "Hello", + "sender": "a@x.com", + "to": "b@y.com", + "messageTimestamp": "2026-04-17T12:00:00Z", + "labelIds": ["INBOX", "UNREAD"], + // Pre-rendered slice (set by `apply_response_level_markdown` + // in production; inline here for the reshape test). + "markdownFormatted": "# Hello\n\nbody copy", + "messageText": "fallback should not be used", + "display_url": "ignore-me", + "preview": { "body": "Hi plain", "subject": "Hello" }, + "attachmentList": [ + { "filename": "report.pdf", "mimeType": "application/pdf", "size": 12345 }, + { "filename": "", "mimeType": "text/html" } + ], + "payload": {} + } + ], + "nextPageToken": "tok-1", + "resultSizeEstimate": 42 + }) +} + +#[test] +fn reshape_emits_slim_envelope() { + let mut v = fixture_with_backend_markdown(); + post_process("GMAIL_FETCH_EMAILS", None, &mut v); + + assert_eq!(v["nextPageToken"], "tok-1"); + assert_eq!(v["resultSizeEstimate"], 42); + + let msgs = v["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + let m = &msgs[0]; + + assert_eq!(m["id"], "m1"); + assert_eq!(m["threadId"], "t1"); + assert_eq!(m["subject"], "Hello"); + assert_eq!(m["from"], "a@x.com"); + assert_eq!(m["to"], "b@y.com"); + assert_eq!(m["date"], "2026-04-17T12:00:00Z"); + assert_eq!(m["labels"], json!(["INBOX", "UNREAD"])); + + let md = m["markdown"].as_str().unwrap(); + assert_eq!(md, "# Hello\n\nbody copy"); + + // Noise fields removed. + assert!(m.get("display_url").is_none()); + assert!(m.get("preview").is_none()); + assert!(m.get("payload").is_none()); + assert!(m.get("messageText").is_none()); + + // Attachments: empty filename entry is filtered. + let atts = m["attachments"].as_array().unwrap(); + assert_eq!(atts.len(), 1); + assert_eq!(atts[0]["filename"], "report.pdf"); + assert_eq!(atts[0]["mimeType"], "application/pdf"); +} + +#[test] +fn raw_html_flag_passes_through_unchanged() { + let mut v = fixture_with_backend_markdown(); + let original = v.clone(); + let args = json!({ "raw_html": true }); + post_process("GMAIL_FETCH_EMAILS", Some(&args), &mut v); + assert_eq!( + v, original, + "raw_html=true must preserve the Composio shape" + ); +} + +#[test] +fn camel_case_raw_html_also_recognized() { + let mut v = fixture_with_backend_markdown(); + let original = v.clone(); + let args = json!({ "rawHtml": true }); + post_process("GMAIL_FETCH_EMAILS", Some(&args), &mut v); + assert_eq!(v, original); +} + +#[test] +fn falls_back_to_message_text_when_no_backend_markdown() { + let mut v = json!({ + "messages": [{ + "messageId": "m1", + "threadId": "t1", + "subject": "s", + "sender": "a@x.com", + "to": "b@y.com", + "messageTimestamp": "2026-04-17", + "labelIds": [], + "messageText": " plain body text ", + "payload": {} + }], + "nextPageToken": null + }); + post_process("GMAIL_FETCH_EMAILS", None, &mut v); + let md = v["messages"][0]["markdown"].as_str().unwrap(); + assert_eq!(md, "plain body text"); + assert!(v.get("nextPageToken").is_none(), "null tokens dropped"); +} + +#[test] +fn unwraps_data_envelope() { + let mut v = json!({ + "data": { + "messages": [{ + "messageId": "m1", + "threadId": "t1", + "subject": "s", + "sender": "a@x.com", + "to": "b@y.com", + "messageTimestamp": "2026-04-17", + "labelIds": [], + "messageText": "body", + "payload": {} + }] + } + }); + post_process("GMAIL_FETCH_EMAILS", None, &mut v); + // Reshape writes into `data` in place. + let msgs = v["data"]["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["markdown"], "body"); +} + +#[test] +fn non_fetch_slug_is_noop() { + let mut v = json!({ "messages": [{ "messageId": "m1", "messageText": "x" }] }); + let original = v.clone(); + post_process("GMAIL_SEND_EMAIL", None, &mut v); + assert_eq!(v, original); +} + +#[test] +fn prefers_backend_markdown_formatted_when_present() { + // Composio backend (tinyhumansai/backend#683 +) ships + // `markdownFormatted` already URL-shortened + footer-stripped + // per message (after `apply_response_level_markdown` slices the + // response-level field). When present, our post-processor must + // use it verbatim instead of falling back to `messageText`. + let mut v = json!({ + "messages": [{ + "messageId": "m1", + "threadId": "t1", + "subject": "s", + "sender": "a@x.com", + "to": "b@y.com", + "messageTimestamp": "2026-04-17", + "labelIds": [], + "markdownFormatted": "# Already nice\n\nShort URL: https://gh.io/abc", + "messageText": "fallback should not be used", + "payload": {} + }] + }); + post_process("GMAIL_FETCH_EMAILS", None, &mut v); + let md = v["messages"][0]["markdown"].as_str().unwrap(); + assert_eq!(md, "# Already nice\n\nShort URL: https://gh.io/abc"); +} + +#[test] +fn empty_markdown_formatted_falls_through_to_message_text() { + let mut v = json!({ + "messages": [{ + "messageId": "m1", + "threadId": "t1", + "subject": "s", + "sender": "a@x.com", + "to": "b@y.com", + "messageTimestamp": "2026-04-17", + "labelIds": [], + "markdownFormatted": " \n \n", + "messageText": "real body", + "payload": {} + }] + }); + post_process("GMAIL_FETCH_EMAILS", None, &mut v); + let md = v["messages"][0]["markdown"].as_str().unwrap(); + assert!(md.contains("real body")); +} + +// ── split_response_markdown_per_message ───────────────────────────────── + +#[test] +fn split_response_markdown_uses_horizontal_rule_marker() { + // The confirmed backend marker is `\n---\n`. Three messages → + // expect three slices when there's no preamble. + let md = "## Alice's update\n\nbody A with https://gh.io/abc\n---\n## Bob's reply\n\nbody B\n---\n## Carol\n\nbody C"; + let slices = super::split_response_markdown_per_message(md, 3).unwrap(); + assert_eq!(slices.len(), 3); + assert!(slices[0].contains("Alice's update")); + assert!(slices[1].contains("Bob's reply")); + assert!(slices[2].contains("Carol")); + // The `---\n` prefix is preserved on every-but-the-first segment + // so the section break survives the round-trip. + assert!(slices[1].starts_with("---\n")); + assert!(slices[2].starts_with("---\n")); +} + +#[test] +fn split_response_markdown_drops_preamble() { + // When a preamble like `# Inbox` precedes the first marker, we + // see N+1 parts after split — the preamble must be dropped. + let md = "# Inbox (2 messages)\n---\n## A\n\nbody A\n---\n## B\n\nbody B"; + let slices = super::split_response_markdown_per_message(md, 2).unwrap(); + assert_eq!(slices.len(), 2); + assert!(slices[0].contains("body A")); + assert!(slices[1].contains("body B")); + // Both segments should carry the prefix when preamble was dropped. + assert!(slices[0].starts_with("---\n")); + assert!(slices[1].starts_with("---\n")); +} + +#[test] +fn split_response_markdown_falls_back_to_h2_marker() { + // No `---` rules — backend used h2 headings as boundaries. + let md = "## Alice\n\nbody A\n\n## Bob\n\nbody B"; + let slices = super::split_response_markdown_per_message(md, 2).unwrap(); + assert_eq!(slices.len(), 2); + assert!(slices[0].contains("body A")); + assert!(slices[1].contains("body B")); +} + +#[test] +fn split_response_markdown_returns_none_on_count_mismatch() { + let md = "## only one section here"; + assert!(super::split_response_markdown_per_message(md, 3).is_none()); +} + +#[test] +fn split_response_markdown_single_message_returns_whole_input() { + let md = "## solo\n\nthe whole body"; + let slices = super::split_response_markdown_per_message(md, 1).unwrap(); + assert_eq!(slices, vec![md.to_string()]); +} + +#[test] +fn split_with_hint_rejects_when_subjects_dont_match() { + let md = "## Foo\nbody1\n---\n## Bar\nbody2"; + let hints = vec![ + json!({"subject": "Completely different subject A"}), + json!({"subject": "Completely different subject B"}), + ]; + let out = super::split_response_markdown_per_message_with_hint(md, 2, Some(&hints)); + assert!(out.is_none(), "subject mismatch must force fallback"); +} + +#[test] +fn split_with_hint_accepts_when_subjects_match() { + let md = "## Welcome to Gmail\nbody1\n---\n## Your invoice\nbody2"; + let hints = vec![ + json!({"subject": "Welcome to Gmail"}), + json!({"subject": "Your invoice"}), + ]; + let slices = super::split_response_markdown_per_message_with_hint(md, 2, Some(&hints)).unwrap(); + assert_eq!(slices.len(), 2); + assert!(slices[0].contains("Welcome to Gmail")); + assert!(slices[1].contains("Your invoice")); +} + +#[test] +fn split_with_hint_skips_messages_with_blank_subject() { + let md = "## A\nbody1\n---\n## B\nbody2"; + let hints = vec![json!({"subject": "A"}), json!({"subject": ""})]; + let slices = super::split_response_markdown_per_message_with_hint(md, 2, Some(&hints)).unwrap(); + assert_eq!(slices.len(), 2); +} + +// ── format_email_local_time ────────────────────────────────────────────────── + +#[test] +fn format_email_local_time_returns_none_for_unparseable_date() { + assert!(super::format_email_local_time("not-a-date").is_none()); + assert!(super::format_email_local_time("").is_none()); +} + +#[test] +fn format_email_local_time_preserves_utc_raw_date_in_reshape() { + let mut v = json!({ + "messages": [{ + "messageId": "m1", + "threadId": "t1", + "subject": "Test", + "sender": "a@example.com", + "to": "b@example.com", + "messageTimestamp": "2026-05-31T10:33:00Z", + "labelIds": [], + "messageText": "body", + "payload": {} + }] + }); + post_process("GMAIL_FETCH_EMAILS", None, &mut v); + let msg = &v["messages"][0]; + assert_eq!(msg["date"], "2026-05-31T10:33:00Z"); +} + +#[test] +fn parse_email_date_accepts_rfc3339_and_rfc2822() { + assert!(super::parse_email_date("2026-05-31T10:33:00Z").is_some()); + assert!(super::parse_email_date("Sun, 31 May 2026 10:33:00 +0000").is_some()); + assert!(super::parse_email_date("not-a-date").is_none()); +} + +#[test] +fn format_at_tz_deterministic_with_fixed_offset() { + use chrono::FixedOffset; + + let utc = super::parse_email_date("2026-05-31T10:33:00Z").unwrap(); + + let est = FixedOffset::west_opt(5 * 3600).unwrap(); + let result = super::format_at_tz(utc, &est).unwrap(); + assert_eq!(result, "2026-05-31 05:33 AM -05:00"); + + let ist = FixedOffset::east_opt(5 * 3600 + 1800).unwrap(); + let result = super::format_at_tz(utc, &ist).unwrap(); + assert_eq!(result, "2026-05-31 04:03 PM +05:30"); +} + +#[test] +fn format_at_tz_returns_none_for_utc() { + let utc = super::parse_email_date("2026-05-31T10:33:00Z").unwrap(); + let utc_tz = chrono::FixedOffset::east_opt(0).unwrap(); + assert!(super::format_at_tz(utc, &utc_tz).is_none()); +} + +#[test] +fn apply_response_level_markdown_stashes_per_message_field() { + let mut data = json!({ + "messages": [ + {"messageId": "m1", "subject": "Hello"}, + {"messageId": "m2", "subject": "World"}, + ] + }); + let top_md = "## Hello\nbody A — link https://gh.io/abc\n---\n## World\nbody B"; + super::apply_response_level_markdown(&mut data, top_md); + let m1 = data["messages"][0]["markdownFormatted"].as_str().unwrap(); + let m2 = data["messages"][1]["markdownFormatted"].as_str().unwrap(); + assert!(m1.contains("Hello")); + assert!( + m1.contains("https://gh.io/abc"), + "shortened URL must survive" + ); + assert!(m2.contains("World")); + assert!(!m1.contains("World"), "no cross-message bleed"); +} diff --git a/sync/src/helpers.rs b/sync/src/helpers.rs new file mode 100644 index 0000000..101239e --- /dev/null +++ b/sync/src/helpers.rs @@ -0,0 +1,50 @@ +//! Shared helpers for the provider normalisers in this module. + +/// Walk a JSON object using a list of dotted-path candidates and return the +/// first non-empty **string** match. +/// +/// # This is deliberately NOT `super::super::common::pick_str` +/// +/// The crate carries two `pick_str` functions with the same name and +/// genuinely different behaviour. Do not "deduplicate" them: +/// +/// | | this one (`normalize::helpers`) | `common::pick_str` | +/// |---|---|---| +/// | traversal | `Value::get` per `.`-separated segment — objects only | `Value::pointer` — also indexes into arrays | +/// | non-string leaf | rejected, returns `None` | `Number` is coerced via `to_string()` | +/// +/// The number case is the one that bites. A payload whose `id` is `42` +/// rather than `"42"` yields `None` here and `Some("42")` there, which +/// silently changes what a normaliser emits as a document id. The callers of +/// this function were written against the reject-non-strings behaviour and +/// have a test pinning it (`pick_str_rejects_non_string_values` below, and +/// the host-side mirror of it). +pub fn pick_str(value: &serde_json::Value, paths: &[&str]) -> Option { + for path in paths { + let mut cur = value; + let mut ok = true; + for segment in path.split('.') { + match cur.get(segment) { + Some(next) => cur = next, + None => { + ok = false; + break; + } + } + } + if !ok { + continue; + } + if let Some(s) = cur.as_str() { + let trimmed = s.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + None +} + +#[cfg(test)] +#[path = "helpers_tests.rs"] +mod tests; diff --git a/sync/src/helpers_tests.rs b/sync/src/helpers_tests.rs new file mode 100644 index 0000000..b6410b8 --- /dev/null +++ b/sync/src/helpers_tests.rs @@ -0,0 +1,40 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +// +// A failing assertion in a test *is* a panic. The crate-wide lints exist to +// keep the library from panicking, not the tests. + +use super::*; +use serde_json::json; + +#[test] +fn pick_str_finds_first_non_empty_match() { + let v = json!({"data": {"user": {"name": "Ada", "email": "ada@example.com"}}}); + assert_eq!( + pick_str(&v, &["data.user.name", "data.user.email"]), + Some("Ada".into()) + ); + assert_eq!( + pick_str(&v, &["data.missing", "data.user.email"]), + Some("ada@example.com".into()) + ); + assert_eq!(pick_str(&v, &["nope.nope"]), None); +} + +#[test] +fn pick_str_respects_path_order() { + let v = json!({"a": "first", "b": "second"}); + assert_eq!(pick_str(&v, &["a", "b"]), Some("first".into())); + assert_eq!(pick_str(&v, &["b", "a"]), Some("second".into())); +} + +/// The drift guard for the divergence documented on [`pick_str`]. If this +/// ever starts returning `Some("42")`, someone has re-pointed the +/// normalisers at `common::pick_str` and changed their output. +#[test] +fn pick_str_rejects_non_string_values() { + let v = json!({"count": 42, "flag": true, "empty": "", "whitespace": " "}); + assert_eq!(pick_str(&v, &["count"]), None); + assert_eq!(pick_str(&v, &["flag"]), None); + assert_eq!(pick_str(&v, &["empty"]), None); + assert_eq!(pick_str(&v, &["whitespace"]), None); +} diff --git a/sync/src/lib.rs b/sync/src/lib.rs new file mode 100644 index 0000000..4eb3ba6 --- /dev/null +++ b/sync/src/lib.rs @@ -0,0 +1,38 @@ +//! Composio provider payload normalisers, engine-neutral by construction. +//! +//! Issue #18 §B3: "Payload normalisers … are pure `Value → Value` transforms +//! with no engine dependency. Move them back into a `tinymemory-sync` crate … +//! so a non-TinyCortex engine gets Composio sync for free." +//! +//! They lived inside the TinyCortex engine, and `tinymemory-core` reached into +//! it to use them — which meant a host binding a *different* memory engine +//! could not have Composio sync at all, despite none of this code caring which +//! engine is bound. Nothing here reads a database, opens a socket, or names an +//! engine type; the dependency list is `serde_json`, two logging facades, and +//! `chrono`. +//! +//! One caveat on "pure", because it is load-bearing and easy to miss. +//! [`gmail_post_process::format_email_local_time`] renders in `chrono::Local`, +//! so it reads the host's timezone — every other normaliser here is a function +//! of its input alone. The raw UTC field is preserved alongside it, so sorting +//! and deduplication stay UTC-based; what varies by host is only the +//! presentation string. +//! +//! These are pure `serde_json::Value` → `Value` transforms: given a raw +//! Composio action response, pull out the fields that make up a task, an +//! issue, a page or a message. They hold no credentials, touch no network, +//! and make no scheduling decisions — provider-specific normalisation is +//! driver-side by definition (see the host's `docs/specs/kernel.md` §4). +//! + +pub mod clickup; +pub mod github; +pub mod helpers; +pub mod linear; +pub mod notion; + +// The `_post_process` suffix is kept from the engine layout it came from, where +// `slack.rs` and `github.rs` one directory up already held those names. Renaming +// on the way out would have made this a rename *and* a move in one diff. +pub mod gmail_post_process; +pub mod slack_post_process; diff --git a/sync/src/linear.rs b/sync/src/linear.rs new file mode 100644 index 0000000..843f98a --- /dev/null +++ b/sync/src/linear.rs @@ -0,0 +1,157 @@ +//! Linear host normalization helpers — result extraction, issue-title extraction, +//! viewer identity, cursor extraction, and time utilities. +//! +//! Linear's GraphQL API (and therefore Composio's wrapping of it) returns +//! connection-style lists (`{ nodes: [...], pageInfo: {...} }`) at the top +//! level or nested under `data`. The functions here walk the union of +//! common shapes so the provider does not have to branch per Composio +//! envelope variant. + +use serde_json::Value; + +use super::helpers::pick_str; + +/// Walk the Composio response envelope for Linear issue list results. +/// +/// Linear's list endpoints return `{ nodes: [...] }` or +/// `{ issues: { nodes: [...] } }` shapes; Composio may re-wrap the +/// upstream payload under `data` or `data.data`. We probe each shape +/// in order and return the first array we find. +pub fn extract_issues(data: &Value) -> Vec { + let candidates = [ + data.pointer("/data/nodes"), + data.pointer("/nodes"), + data.pointer("/data/issues/nodes"), + data.pointer("/issues/nodes"), + data.pointer("/data/data/nodes"), + data.pointer("/data/data/issues/nodes"), + data.pointer("/data/results"), + data.pointer("/results"), + data.pointer("/data/items"), + data.pointer("/items"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(arr) = cand.as_array() { + return arr.clone(); + } + } + Vec::new() +} + +/// Extract a human-readable title from a Linear issue object. +/// +/// Linear issues store the name at `title` (or `data.title` after +/// Composio envelope wrapping). Falls back to `name` / `identifier` +/// so the chunk remains identifiable even for unusual response shapes. +pub fn extract_issue_title(issue: &Value) -> Option { + pick_str( + issue, + &[ + "title", + "data.title", + "name", + "data.name", + "identifier", + "data.identifier", + ], + ) +} + +/// Extract a stable cursor timestamp from a Linear issue object. +/// +/// Linear uses ISO-8601 strings for timestamps (`updatedAt`). We keep +/// the value as a string so lexicographic comparison against the stored +/// cursor is valid. +pub fn extract_issue_updated(issue: &Value) -> Option { + pick_str( + issue, + &[ + "updatedAt", + "data.updatedAt", + "updated_at", + "data.updated_at", + ], + ) +} + +/// Extract the viewer (authenticated user) object from a +/// `LINEAR_LIST_LINEAR_USERS { isMe: true }` response. +/// +/// Linear's GraphQL viewer endpoint returns `{ nodes: [{ id, email, … }] }`. +/// Composio may wrap this under `data` or `data.data`. We probe each +/// shape and return the first element of the nodes array, falling back +/// to the payload itself if it looks like a direct user object (has +/// `id` or `email`). +pub fn extract_viewer(data: &Value) -> Option { + let array_candidates = [ + data.pointer("/data/nodes"), + data.pointer("/nodes"), + data.pointer("/data/data/nodes"), + data.pointer("/data/users/nodes"), + ]; + for cand in array_candidates.into_iter().flatten() { + if let Some(arr) = cand.as_array() { + if let Some(first) = arr.first() { + return Some(first.clone()); + } + } + } + // Fallback: if the payload itself looks like a user object, return it. + if data.get("id").is_some() || data.get("email").is_some() { + return Some(data.clone()); + } + None +} + +/// Extract the viewer's ID string from a `LINEAR_LIST_LINEAR_USERS` +/// response. Returns `None` if the payload does not contain a +/// recognizable user ID. +pub fn extract_viewer_id(data: &Value) -> Option { + let viewer = extract_viewer(data)?; + pick_str(&viewer, &["id", "data.id"]) +} + +/// Extract a pagination cursor from a Linear connection `pageInfo` block. +/// +/// Returns `Some(endCursor)` only when `hasNextPage` is `true`; +/// `None` when the last page has been reached or when the envelope does +/// not carry `pageInfo` at all. +pub fn extract_pagination_cursor(data: &Value) -> Option { + // Mirrors the `extract_issues` envelope shapes, so every shape that can + // carry a node list can also carry its `pageInfo` cursor. + let page_info_candidates = [ + data.pointer("/data/pageInfo"), + data.pointer("/pageInfo"), + data.pointer("/data/data/pageInfo"), + data.pointer("/data/issues/pageInfo"), + data.pointer("/data/data/issues/pageInfo"), + ]; + for cand in page_info_candidates.into_iter().flatten() { + let has_next = cand + .get("hasNextPage") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + if has_next { + if let Some(cursor) = cand.get("endCursor").and_then(|v| v.as_str()) { + let trimmed = cursor.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + } + None +} + +/// Current wall-clock time in milliseconds since the UNIX epoch. +pub fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +#[cfg(test)] +#[path = "linear_tests.rs"] +mod tests; diff --git a/sync/src/linear_tests.rs b/sync/src/linear_tests.rs new file mode 100644 index 0000000..78b5bf1 --- /dev/null +++ b/sync/src/linear_tests.rs @@ -0,0 +1,189 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +// +// A failing assertion in a test *is* a panic. The crate-wide lints exist to +// keep the library from panicking, not the tests. + +use super::*; +use serde_json::json; + +// ── extract_issues ─────────────────────────────────────────────── + +#[test] +fn extract_issues_from_data_nodes() { + let data = json!({ "data": { "nodes": [{"id": "i1"}, {"id": "i2"}] } }); + assert_eq!(extract_issues(&data).len(), 2); +} + +#[test] +fn extract_issues_from_top_level_nodes() { + let data = json!({ "nodes": [{"id": "i3"}] }); + assert_eq!(extract_issues(&data).len(), 1); +} + +#[test] +fn extract_issues_from_data_issues_nodes() { + let data = + json!({ "data": { "issues": { "nodes": [{"id": "i4"}, {"id": "i5"}, {"id": "i6"}] } } }); + assert_eq!(extract_issues(&data).len(), 3); +} + +#[test] +fn extract_issues_from_top_level_issues_nodes() { + let data = json!({ "issues": { "nodes": [{"id": "i7"}] } }); + assert_eq!(extract_issues(&data).len(), 1); +} + +#[test] +fn extract_issues_from_doubly_nested_issues_nodes() { + let data = + json!({ "data": { "data": { "issues": { "nodes": [{"id": "i8"}, {"id": "i9"}] } } } }); + assert_eq!(extract_issues(&data).len(), 2); +} + +#[test] +fn extract_issues_from_results() { + let data = json!({ "results": [{"id": "i7"}] }); + assert_eq!(extract_issues(&data).len(), 1); +} + +#[test] +fn extract_issues_empty_when_missing() { + let data = json!({ "foo": "bar" }); + assert!(extract_issues(&data).is_empty()); +} + +// ── extract_issue_title ────────────────────────────────────────── + +#[test] +fn extract_issue_title_from_title_field() { + let issue = json!({ "id": "i1", "title": "Fix the login bug" }); + assert_eq!( + extract_issue_title(&issue), + Some("Fix the login bug".into()) + ); +} + +#[test] +fn extract_issue_title_falls_back_to_wrapped_data() { + let issue = json!({ "data": { "title": "Wrapped issue" } }); + assert_eq!(extract_issue_title(&issue), Some("Wrapped issue".into())); +} + +#[test] +fn extract_issue_title_falls_back_to_identifier() { + let issue = json!({ "identifier": "ENG-42" }); + assert_eq!(extract_issue_title(&issue), Some("ENG-42".into())); +} + +// ── extract_issue_updated ──────────────────────────────────────── + +#[test] +fn extract_issue_updated_from_updated_at() { + let issue = json!({ "updatedAt": "2026-03-01T12:00:00.000Z" }); + assert_eq!( + extract_issue_updated(&issue), + Some("2026-03-01T12:00:00.000Z".to_string()) + ); +} + +#[test] +fn extract_issue_updated_falls_back_to_snake_case() { + let issue = json!({ "data": { "updated_at": "2026-01-15T08:30:00.000Z" } }); + assert_eq!( + extract_issue_updated(&issue), + Some("2026-01-15T08:30:00.000Z".to_string()) + ); +} + +// ── extract_viewer ─────────────────────────────────────────────── + +#[test] +fn extract_viewer_from_data_nodes() { + let data = json!({ "data": { "nodes": [{ "id": "usr_1", "email": "a@b.com" }] } }); + let v = extract_viewer(&data).expect("should find viewer"); + assert_eq!(v["id"], "usr_1"); +} + +#[test] +fn extract_viewer_from_top_level_nodes() { + let data = json!({ "nodes": [{ "id": "usr_2" }] }); + let v = extract_viewer(&data).expect("should find viewer"); + assert_eq!(v["id"], "usr_2"); +} + +#[test] +fn extract_viewer_fallback_direct_object() { + let data = json!({ "id": "usr_direct", "name": "Direct User" }); + let v = extract_viewer(&data).expect("should return direct object"); + assert_eq!(v["id"], "usr_direct"); +} + +#[test] +fn extract_viewer_returns_none_when_absent() { + let data = json!({ "foo": "bar" }); + assert!(extract_viewer(&data).is_none()); +} + +// ── extract_pagination_cursor ──────────────────────────────────── + +#[test] +fn extract_pagination_cursor_returns_cursor_when_has_next_page() { + let data = json!({ + "data": { + "pageInfo": { + "hasNextPage": true, + "endCursor": "cursor_abc" + } + } + }); + assert_eq!( + extract_pagination_cursor(&data), + Some("cursor_abc".to_string()) + ); +} + +#[test] +fn extract_pagination_cursor_returns_none_when_last_page() { + let data = json!({ + "pageInfo": { + "hasNextPage": false, + "endCursor": "cursor_xyz" + } + }); + assert!(extract_pagination_cursor(&data).is_none()); +} + +#[test] +fn extract_pagination_cursor_from_doubly_nested_issues() { + // The same `data.data.issues` shape `extract_issues` reads must also + // expose its pageInfo cursor, or a doubly-nested payload never pages. + let data = json!({ + "data": { + "data": { + "issues": { + "pageInfo": { + "hasNextPage": true, + "endCursor": "cursor_issue_2" + } + } + } + } + }); + assert_eq!( + extract_pagination_cursor(&data), + Some("cursor_issue_2".to_string()) + ); +} + +#[test] +fn extract_pagination_cursor_returns_none_when_absent() { + let data = json!({ "nodes": [{"id": "i1"}] }); + assert!(extract_pagination_cursor(&data).is_none()); +} + +// ── now_ms ─────────────────────────────────────────────────────── + +#[test] +fn now_ms_returns_nonzero() { + assert!(now_ms() > 0); +} diff --git a/sync/src/notion.rs b/sync/src/notion.rs new file mode 100644 index 0000000..f13c17f --- /dev/null +++ b/sync/src/notion.rs @@ -0,0 +1,120 @@ +//! Notion host normalization helpers — result extraction, pagination cursor, +//! page title extraction, and time utilities. + +use serde_json::Value; + +use super::helpers::pick_str; + +/// Walk the Composio response envelope for Notion page results. +pub fn extract_results(data: &Value) -> Vec { + let candidates = [ + data.pointer("/data/results"), + data.pointer("/results"), + data.pointer("/data/data/results"), + data.pointer("/data/items"), + data.pointer("/items"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(arr) = cand.as_array() { + return arr.clone(); + } + } + Vec::new() +} + +/// Extract the rendered page body markdown from a `NOTION_GET_PAGE_MARKDOWN` +/// response. Composio wraps action output in varying envelope shapes, so we +/// try the common locations tolerantly and return the first non-empty string. +/// Returns `None` if no markdown field is found (caller falls back to the +/// metadata-only body and logs the raw shape for diagnosis). +pub fn extract_page_markdown(data: &Value) -> Option { + const PATHS: &[&str] = &[ + "/markdown", + "/data/markdown", + "/data/response_data/markdown", + "/response_data/markdown", + "/data/content", + "/content", + "/data/markdown_content", + "/markdown_content", + "/text", + "/data/text", + ]; + for p in PATHS { + if let Some(s) = data.pointer(p).and_then(Value::as_str) { + if !s.trim().is_empty() { + return Some(s.to_string()); + } + } + } + None +} + +/// Extract the Notion pagination cursor (for `start_cursor` on the +/// next request). +pub fn extract_notion_cursor(data: &Value) -> Option { + let candidates = [ + data.pointer("/data/next_cursor"), + data.pointer("/next_cursor"), + data.pointer("/data/data/next_cursor"), + ]; + for cand in candidates.into_iter().flatten() { + if let Some(s) = cand.as_str() { + let trimmed = s.trim(); + if !trimmed.is_empty() { + return Some(trimmed.to_string()); + } + } + } + None +} + +/// Try to extract a human-readable title from a Notion page object. +/// +/// Notion pages store the title in `properties.title` or +/// `properties.Name.title[0].plain_text`. We try several shapes. +pub fn extract_page_title(page: &Value) -> Option { + // Try the common `properties.title.title[0].plain_text` shape. + let props = page + .get("properties") + .or_else(|| page.get("data")?.get("properties")); + if let Some(props) = props { + // Walk all properties looking for a "title" type field. + if let Some(obj) = props.as_object() { + for (_key, val) in obj { + if val.get("type").and_then(Value::as_str) == Some("title") { + if let Some(arr) = val.get("title").and_then(Value::as_array) { + let text: String = arr + .iter() + .filter_map(|t| t.get("plain_text").and_then(Value::as_str)) + .collect::>() + .join(""); + if !text.is_empty() { + return Some(text); + } + } + } + } + } + } + + // Fallback: top-level "title" field (some Composio shapes). + pick_str(page, &["title", "data.title", "name", "data.name"]) +} + +/// Milliseconds since the Unix epoch. +/// +/// The one clock read in this crate. Notion payloads carry no ingestion +/// timestamp, so the normaliser stamps one; everything else here is a function +/// of its input alone. +pub fn now_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +#[cfg(test)] +#[path = "notion_tests.rs"] +mod tests; diff --git a/sync/src/notion_tests.rs b/sync/src/notion_tests.rs new file mode 100644 index 0000000..84b46e1 --- /dev/null +++ b/sync/src/notion_tests.rs @@ -0,0 +1,142 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +// +// A failing assertion in a test *is* a panic. The crate-wide lints exist to +// keep the library from panicking, not the tests. + +use super::*; +use serde_json::json; + +#[test] +fn extract_results_from_data_results() { + let data = json!({"data": {"results": [{"id": "page1"}]}}); + let results = extract_results(&data); + assert_eq!(results.len(), 1); +} + +#[test] +fn extract_page_markdown_reads_top_level_field() { + // Matches the live GET_PAGE_MARKDOWN envelope observed empirically: + // {id, markdown, object, request_id, truncated, unknown_block_ids}. + let data = json!({ + "id": "p1", + "markdown": "# Heading\n\nbody text", + "object": "page", + "truncated": false, + }); + assert_eq!( + extract_page_markdown(&data).as_deref(), + Some("# Heading\n\nbody text") + ); +} + +#[test] +fn extract_page_markdown_reads_nested_envelope() { + let data = json!({ "data": { "markdown": "nested body" } }); + assert_eq!(extract_page_markdown(&data).as_deref(), Some("nested body")); +} + +#[test] +fn extract_page_markdown_none_for_empty_or_missing() { + // Empty markdown (a DB row with no body blocks) → None → metadata-only. + assert_eq!(extract_page_markdown(&json!({ "markdown": "" })), None); + assert_eq!(extract_page_markdown(&json!({ "markdown": " " })), None); + // No markdown field at all → None. + assert_eq!(extract_page_markdown(&json!({ "id": "p1" })), None); +} + +#[test] +fn extract_results_from_top_level() { + let data = json!({"results": [{"id": "a"}, {"id": "b"}]}); + let results = extract_results(&data); + assert_eq!(results.len(), 2); +} + +#[test] +fn extract_results_from_data_items() { + let data = json!({"data": {"items": [{"id": "x"}]}}); + let results = extract_results(&data); + assert_eq!(results.len(), 1); +} + +#[test] +fn extract_results_empty_when_no_match() { + let data = json!({"foo": "bar"}); + assert!(extract_results(&data).is_empty()); +} + +#[test] +fn extract_notion_cursor_from_data() { + let data = json!({"data": {"next_cursor": "cur123"}}); + assert_eq!(extract_notion_cursor(&data), Some("cur123".into())); +} + +#[test] +fn extract_notion_cursor_from_top_level() { + let data = json!({"next_cursor": "abc"}); + assert_eq!(extract_notion_cursor(&data), Some("abc".into())); +} + +#[test] +fn extract_notion_cursor_none_when_empty() { + let data = json!({"data": {"next_cursor": " "}}); + assert_eq!(extract_notion_cursor(&data), None); +} + +#[test] +fn extract_notion_cursor_none_when_missing() { + assert_eq!(extract_notion_cursor(&json!({})), None); +} + +#[test] +fn extract_page_title_from_properties_title_type() { + let page = json!({ + "properties": { + "Name": { + "type": "title", + "title": [{"plain_text": "Hello"}, {"plain_text": " World"}] + } + } + }); + assert_eq!(extract_page_title(&page), Some("Hello World".into())); +} + +#[test] +fn extract_page_title_from_nested_data_properties() { + let page = json!({ + "data": { + "properties": { + "Title": { + "type": "title", + "title": [{"plain_text": "My Page"}] + } + } + } + }); + assert_eq!(extract_page_title(&page), Some("My Page".into())); +} + +#[test] +fn extract_page_title_fallback_to_top_level_title() { + let page = json!({"title": "Fallback Title"}); + assert_eq!(extract_page_title(&page), Some("Fallback Title".into())); +} + +#[test] +fn extract_page_title_none_when_empty() { + let page = json!({"properties": {"Name": {"type": "title", "title": []}}}); + // Empty title array means no text + assert!( + extract_page_title(&page).is_none() || extract_page_title(&page) == Some(String::new()) + ); +} + +#[test] +fn extract_page_title_none_when_no_title_field() { + let page = json!({"id": "123"}); + assert!(extract_page_title(&page).is_none()); +} + +#[test] +fn now_ms_returns_nonzero() { + assert!(now_ms() > 0); +} diff --git a/sync/src/slack_post_process.rs b/sync/src/slack_post_process.rs new file mode 100644 index 0000000..28ce30b --- /dev/null +++ b/sync/src/slack_post_process.rs @@ -0,0 +1,323 @@ +//! Slack-specific post-processing of Composio action responses. +//! +//! Composio's Slack responses are verbose API envelopes. This module +//! rewrites each supported action's response into a slim, stable shape +//! that the ingest pipeline and enrichers can consume without walking +//! Composio's unstable nested envelopes. +//! +//! ## Supported slugs +//! +//! - `SLACK_FETCH_CONVERSATION_HISTORY` — reshapes into top-level +//! `messages[]` with `{ ts, user, text, thread_ts, channel_id }`. +//! Empty-text messages are dropped. `channel_id` is absent here (it's +//! in the request, not the response); the caller injects it via the +//! enricher in the host's `SlackSyncPipeline`. +//! +//! - `SLACK_LIST_CONVERSATIONS` — reshapes into top-level `channels[]` +//! with `{ id, name, is_private }` per channel. Entries with an empty +//! id are dropped. +//! +//! - `SLACK_SEARCH_MESSAGES` — reshapes `messages.matches[]` (possibly +//! nested) into top-level `messages[]` with `{ ts, user, text, +//! thread_ts, channel_id }`. `channel_id` is pulled from each match's +//! `channel.id` field. `paging.pages` is preserved at top-level for +//! caller pagination. +//! +//! ## Design note: user-id resolution is NOT here +//! +//! `SlackUsers` is a per-sync cache built from a separate API call — +//! not a function of any individual response. Resolving user ids +//! happens in the host's `SlackSyncPipeline` (the enricher layer), keeping +//! this module purely data-shape–oriented. +//! This matches Gmail's pattern of "post_process is data-only". +//! +//! Unknown slugs are silently no-ops so new Composio actions don't +//! break the provider. + +use serde_json::{Map, Value}; + +/// Entry point called from `SlackProvider::post_process_action_result`. +/// +/// Dispatches on the Composio action slug and rewrites `data` in place. +/// Unknown slugs are silently ignored. +pub fn post_process(slug: &str, _arguments: Option<&Value>, data: &mut Value) { + log::debug!("[composio:slack][post-process] slug={slug}"); + match slug { + "SLACK_FETCH_CONVERSATION_HISTORY" => reshape_fetch_history(data), + "SLACK_LIST_CONVERSATIONS" => reshape_list_conversations(data), + "SLACK_SEARCH_MESSAGES" => reshape_search_messages(data), + _ => { + log::debug!("[composio:slack][post-process] unknown slug={slug}, passing through"); + } + } +} + +// ─── SLACK_FETCH_CONVERSATION_HISTORY ────────────────────────────────────── + +/// Rewrite a `SLACK_FETCH_CONVERSATION_HISTORY` response in place. +/// +/// Walks possible nested envelopes (`/data/messages`, `/messages`, +/// `/data/data/messages`) to find the raw messages array, drops messages +/// with empty `text`, and emits a slim `{ ts, user, text, thread_ts }` +/// shape under a top-level `messages[]` key. The consumed nested array is +/// removed from the payload so the raw verbose rows don't linger alongside +/// the slim copy. The caller injects `channel_id` via +/// [`super::sync::extract_messages`]. +fn reshape_fetch_history(data: &mut Value) { + let arr = take_array( + data, + &["/data/messages", "/messages", "/data/data/messages"], + 0, + ); + let slim: Vec = arr.into_iter().filter_map(slim_history_message).collect(); + let obj = ensure_object(data); + obj.insert("messages".to_string(), Value::Array(slim)); + log::debug!("[composio:slack][post-process] SLACK_FETCH_CONVERSATION_HISTORY reshaped"); +} + +fn slim_history_message(raw: Value) -> Option { + let text = raw + .get("text") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim(); + if text.is_empty() { + return None; + } + let mut out = Map::new(); + // `ts` is required: without it a caller can neither cursor nor archive. + out.insert("ts".into(), raw.get("ts")?.clone()); + if let Some(user) = raw.get("user").or_else(|| raw.get("bot_id")) { + out.insert("user".into(), user.clone()); + } + out.insert("text".into(), Value::String(text.to_string())); + if let Some(thread_ts) = raw.get("thread_ts") { + out.insert("thread_ts".into(), thread_ts.clone()); + } + if let Some(permalink) = raw.get("permalink") { + out.insert("permalink".into(), permalink.clone()); + } + Some(Value::Object(out)) +} + +/// Find the first array at any of `candidates`, remove that field (plus +/// `envelope_depth` ancestor object envelopes) from `data`, and return the +/// array. Removing the consumed nested payload keeps the reshaped output from +/// carrying duplicate raw rows. +fn take_array(data: &mut Value, candidates: &[&str], envelope_depth: usize) -> Vec { + for path in candidates { + let arr = match data.pointer(path).and_then(|v| v.as_array().cloned()) { + Some(a) => a, + None => continue, + }; + let mut remove_path = path.to_string(); + for _ in 0..envelope_depth { + remove_path = match remove_path.rsplit_once('/') { + Some((parent, _)) => parent.to_string(), + None => break, + }; + } + remove_nested(data, &remove_path); + return arr; + } + Vec::new() +} + +/// Remove the field at `path` from `data`, pruning any ancestor object that +/// the removal left empty so a consumed `data` envelope disappears entirely +/// instead of lingering as `{}`. +fn remove_nested(data: &mut Value, path: &str) { + let segments: Vec<&str> = path + .trim_start_matches('/') + .split('/') + .filter(|s| !s.is_empty()) + .collect(); + if segments.is_empty() { + return; + } + + // Remove the leaf field. + let mut current = &mut *data; + for seg in &segments[..segments.len() - 1] { + current = match current.get_mut(*seg) { + Some(next) => next, + None => return, + }; + } + if let Value::Object(map) = current { + map.remove(segments[segments.len() - 1]); + } + + // Prune empty object ancestors, deepest first. + for depth in (0..segments.len().saturating_sub(1)).rev() { + // Re-walk to the object at `segments[..=depth]`. + let mut ancestor = &mut *data; + for seg in &segments[..=depth] { + ancestor = match ancestor.get_mut(*seg) { + Some(next) => next, + None => return, + }; + } + if !matches!(ancestor, Value::Object(m) if m.is_empty()) { + break; + } + // Remove it from its parent (`segments[..depth]`). For `depth == 0` + // the parent is the top-level object, so an emptied `data` envelope + // key disappears entirely. + let mut parent = &mut *data; + for seg in &segments[..depth] { + parent = match parent.get_mut(*seg) { + Some(next) => next, + None => return, + }; + } + if let Value::Object(map) = parent { + map.remove(segments[depth]); + } + } +} + +// ─── SLACK_LIST_CONVERSATIONS ─────────────────────────────────────────────── + +/// Rewrite a `SLACK_LIST_CONVERSATIONS` response in place. +/// +/// Reshapes into a top-level `channels[]` with `{ id, name, is_private }` +/// per channel; entries with an empty id are dropped. +fn reshape_list_conversations(data: &mut Value) { + let arr = take_array( + data, + &[ + "/data/channels", + "/channels", + "/data/data/channels", + "/data/conversations", + "/conversations", + ], + 0, + ); + + let slim: Vec = arr.into_iter().filter_map(slim_channel).collect(); + let obj = ensure_object(data); + obj.insert("channels".to_string(), Value::Array(slim)); + log::debug!("[composio:slack][post-process] SLACK_LIST_CONVERSATIONS reshaped"); +} + +fn slim_channel(raw: Value) -> Option { + let id = raw.get("id").and_then(|v| v.as_str()).unwrap_or("").trim(); + if id.is_empty() { + return None; + } + let name = raw + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or(id) + .trim(); + let is_private = raw + .get("is_private") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + Some(Value::Object({ + let mut m = Map::new(); + m.insert("id".into(), Value::String(id.to_string())); + m.insert("name".into(), Value::String(name.to_string())); + m.insert("is_private".into(), Value::Bool(is_private)); + m + })) +} + +// ─── SLACK_SEARCH_MESSAGES ────────────────────────────────────────────────── + +/// Rewrite a `SLACK_SEARCH_MESSAGES` response in place. +/// +/// Reshapes `messages.matches[]` (possibly nested under one or two +/// `data` envelopes) into top-level `messages[]`. `channel_id` is pulled +/// from each match's `channel.id` field. `paging.pages` is preserved at +/// top-level under `pages` for the caller to drive pagination. +fn reshape_search_messages(data: &mut Value) { + // Preserve paging info before mutating data (take_array below removes the + // envelope that carries it). + let pages = [ + data.pointer("/data/messages/paging/pages"), + data.pointer("/messages/paging/pages"), + data.pointer("/data/data/messages/paging/pages"), + ] + .into_iter() + .flatten() + .find_map(|v| v.as_u64()) + .unwrap_or(1); + + // Envelope depth 1 removes the `messages` object (matches + paging) that + // held the consumed rows, not just the `matches` array. + let arr = take_array( + data, + &[ + "/data/messages/matches", + "/messages/matches", + "/data/data/messages/matches", + ], + 1, + ); + + let slim: Vec = arr.into_iter().filter_map(slim_search_match).collect(); + let obj = ensure_object(data); + obj.insert("messages".to_string(), Value::Array(slim)); + obj.insert("pages".to_string(), Value::Number(pages.into())); + log::debug!("[composio:slack][post-process] SLACK_SEARCH_MESSAGES reshaped"); +} + +fn slim_search_match(raw: Value) -> Option { + let text = raw + .get("text") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim(); + if text.is_empty() { + return None; + } + let ts = raw.get("ts")?; + let channel_id = raw + .pointer("/channel/id") + .and_then(|v| v.as_str()) + .unwrap_or("") + .trim(); + + let mut out = Map::new(); + out.insert("ts".into(), ts.clone()); + if let Some(user) = raw.get("user").or_else(|| raw.get("bot_id")) { + out.insert("user".into(), user.clone()); + } + out.insert("text".into(), Value::String(text.to_string())); + if let Some(thread_ts) = raw.get("thread_ts") { + out.insert("thread_ts".into(), thread_ts.clone()); + } + if !channel_id.is_empty() { + out.insert("channel_id".into(), Value::String(channel_id.to_string())); + } + if let Some(permalink) = raw.get("permalink") { + out.insert("permalink".into(), permalink.clone()); + } + Some(Value::Object(out)) +} + +// ─── Helpers ──────────────────────────────────────────────────────────────── + +/// Ensure `data` is a JSON object, replacing it with an empty object if +/// not. Returns a mutable ref to the inner map. +// Scoped rather than blanket, for the case `AGENTS.md` names: "genuinely +// unreachable states — where `expect` must carry a message explaining the +// invariant." The line below assigns `Value::Object` whenever `data` is not +// one, so the read-back cannot fail; the compiler cannot see that across the +// assignment. The two `unwrap`s this crate inherited elsewhere were removed +// rather than allowed. +#[allow(clippy::expect_used)] +fn ensure_object(data: &mut Value) -> &mut Map { + if !data.is_object() { + *data = Value::Object(Map::new()); + } + data.as_object_mut() + .expect("assigned Value::Object immediately above when data was not one") +} + +#[cfg(test)] +#[path = "slack_post_process_tests.rs"] +mod tests; diff --git a/sync/src/slack_post_process_tests.rs b/sync/src/slack_post_process_tests.rs new file mode 100644 index 0000000..aeec63a --- /dev/null +++ b/sync/src/slack_post_process_tests.rs @@ -0,0 +1,262 @@ +#![allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)] +// +// A failing assertion in a test *is* a panic. The crate-wide lints exist to +// keep the library from panicking, not the tests. + +use super::*; +use serde_json::json; + +// ─── SLACK_FETCH_CONVERSATION_HISTORY ───────────────────────────────────── + +#[test] +fn history_reshapes_top_level_messages() { + let mut data = json!({ + "messages": [ + { "ts": "1714003200.000100", "user": "U1", "text": "hello" }, + { "ts": "1714003300.000200", "user": "U2", "text": "world", "thread_ts": "1714003200.0" }, + { "ts": "1714003400.000300", "user": "U3", "text": " " }, // dropped: empty text + ], + "response_metadata": { "next_cursor": "abc" } + }); + post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); + + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 2, "empty-text message must be dropped"); + assert_eq!(msgs[0]["ts"], "1714003200.000100"); + assert_eq!(msgs[0]["user"], "U1"); + assert_eq!(msgs[0]["text"], "hello"); + assert!(msgs[0].get("thread_ts").is_none()); + assert_eq!(msgs[1]["thread_ts"], "1714003200.0"); +} + +#[test] +fn history_reshapes_nested_data_envelope() { + let mut data = json!({ + "data": { + "messages": [ + { "ts": "1714003200.0", "user": "U1", "text": "hi" } + ] + } + }); + post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["text"], "hi"); +} + +#[test] +fn history_reshapes_doubly_nested_envelope() { + let mut data = json!({ + "data": { + "data": { + "messages": [ + { "ts": "1714003200.0", "user": "U1", "text": "deep" } + ] + } + } + }); + post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["text"], "deep"); +} + +#[test] +fn history_drops_message_without_ts() { + let mut data = json!({ + "messages": [ + { "user": "U1", "text": "no timestamp" }, + { "ts": "1714003200.0", "user": "U2", "text": "has ts" }, + ] + }); + post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["text"], "has ts"); +} + +#[test] +fn history_removes_nested_envelope_after_reshape() { + let mut data = json!({ + "data": { + "messages": [ + { "ts": "1714003200.0", "user": "U1", "text": "hi" } + ] + } + }); + post_process("SLACK_FETCH_CONVERSATION_HISTORY", None, &mut data); + + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["text"], "hi"); + assert!( + data.pointer("/data").is_none(), + "consumed `data.messages` envelope must be removed, got: {data}" + ); +} + +// ─── SLACK_LIST_CONVERSATIONS ───────────────────────────────────────────── + +#[test] +fn list_conversations_reshapes_channels() { + let mut data = json!({ + "data": { + "channels": [ + { "id": "C1", "name": "eng", "is_private": false, "extra": "noise" }, + { "id": "G1", "name": "ops", "is_private": true }, + { "id": "", "name": "empty-id" }, // dropped + ] + } + }); + post_process("SLACK_LIST_CONVERSATIONS", None, &mut data); + let channels = data["channels"].as_array().unwrap(); + assert_eq!(channels.len(), 2, "empty-id entry must be dropped"); + assert_eq!(channels[0]["id"], "C1"); + assert_eq!(channels[0]["name"], "eng"); + assert_eq!(channels[0]["is_private"], false); + assert!( + channels[0].get("extra").is_none(), + "noise fields must be removed" + ); + assert_eq!(channels[1]["id"], "G1"); + assert_eq!(channels[1]["is_private"], true); +} + +#[test] +fn list_conversations_falls_back_to_conversations_key() { + let mut data = json!({ + "conversations": [ + { "id": "C2", "name": "dev", "is_private": false } + ] + }); + post_process("SLACK_LIST_CONVERSATIONS", None, &mut data); + let channels = data["channels"].as_array().unwrap(); + assert_eq!(channels.len(), 1); + assert_eq!(channels[0]["id"], "C2"); + assert!( + data.pointer("/conversations").is_none(), + "consumed `conversations` field must be removed" + ); +} + +// ─── SLACK_SEARCH_MESSAGES ──────────────────────────────────────────────── + +#[test] +fn search_messages_reshapes_matches() { + let mut data = json!({ + "messages": { + "matches": [ + { + "ts": "1714003200.0", + "user": "U1", + "text": "hello from search", + "channel": { "id": "C1" } + }, + { + "ts": "1714003300.0", + "user": "U2", + "text": " ", // dropped: whitespace only + "channel": { "id": "C1" } + }, + ], + "paging": { "pages": 3 } + } + }); + post_process("SLACK_SEARCH_MESSAGES", None, &mut data); + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1, "empty-text match must be dropped"); + assert_eq!(msgs[0]["ts"], "1714003200.0"); + assert_eq!(msgs[0]["text"], "hello from search"); + assert_eq!(msgs[0]["channel_id"], "C1"); + assert_eq!(data["pages"], 3, "paging.pages must be preserved"); +} + +#[test] +fn search_messages_nested_data_envelope() { + let mut data = json!({ + "data": { + "messages": { + "matches": [ + { "ts": "1714003200.0", "user": "U1", "text": "nested", "channel": { "id": "C2" } } + ], + "paging": { "pages": 1 } + } + } + }); + post_process("SLACK_SEARCH_MESSAGES", None, &mut data); + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["channel_id"], "C2"); + assert_eq!(data["pages"], 1_u64); +} + +#[test] +fn search_messages_no_matches_emits_empty_array() { + let mut data = json!({ "messages": { "matches": [] } }); + post_process("SLACK_SEARCH_MESSAGES", None, &mut data); + let msgs = data["messages"].as_array().unwrap(); + assert!(msgs.is_empty()); +} + +#[test] +fn search_messages_removes_nested_envelope_after_reshape() { + let mut data = json!({ + "data": { + "messages": { + "matches": [ + { "ts": "1714003200.0", "user": "U1", "text": "nested", "channel": { "id": "C2" } } + ], + "paging": { "pages": 1 } + } + } + }); + post_process("SLACK_SEARCH_MESSAGES", None, &mut data); + + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["channel_id"], "C2"); + assert_eq!(data["pages"], 1_u64); + assert!( + data.pointer("/data").is_none(), + "consumed `data.messages` envelope must be removed, got: {data}" + ); +} + +#[test] +fn search_messages_doubly_nested_paging_preserved() { + let mut data = json!({ + "data": { + "data": { + "messages": { + "matches": [ + { "ts": "1714003200.0", "user": "U1", "text": "deep", "channel": { "id": "C3" } } + ], + "paging": { "pages": 4 } + } + } + } + }); + post_process("SLACK_SEARCH_MESSAGES", None, &mut data); + + let msgs = data["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["text"], "deep"); + assert_eq!( + data["pages"], 4_u64, + "doubly-nested paging must be preserved" + ); + assert!( + data.pointer("/data").is_none(), + "consumed `data.data.messages` envelope must be removed, got: {data}" + ); +} + +// ─── Unknown slug ───────────────────────────────────────────────────────── + +#[test] +fn unknown_slug_is_noop() { + let mut data = json!({ "foo": "bar" }); + let original = data.clone(); + post_process("SLACK_SEND_MESSAGE", None, &mut data); + assert_eq!(data, original, "unknown slug must not mutate data"); +} From 626144ba20a29706b6fc25febb6960195347572c Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 18 Aug 2026 13:01:03 +0530 Subject: [PATCH 2/4] Run Composio normalisation and storage on a driver that is not TinyCortex MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Issue #18 §B3's stated purpose — "so a non-TinyCortex engine gets Composio sync for free" — asserted rather than assumed, and the first half of §B5's acceptance test. The extraction in the previous commit is only worth something if the result is reachable without an engine. This drives a raw Composio Gmail payload through `tinymemory_sync::gmail_post_process` and stores each normalised message into a bound provider, in a file whose dependencies are the facade, the conformance reference driver, and the sync crate. It names no engine, and before §B3 it could not have compiled: the normalisers lived inside TinyCortex, and `tinymemory-core` reached in to use them. Run against two drivers rather than one. "A non-TinyCortex engine" is a claim about drivers in general, not about whichever one happened to be convenient, so the identical path runs against the reference driver and against the null driver — whose retention semantics are the opposite, and whose empty read is asserted for that reason. Provenance is asserted too. The payload came off somebody's inbox, so it is stored `ExternalSync`, and the read-back checks the driver did not launder it to `Internal`. That is the one failure the taint argument exists to prevent, and it is worth pinning on the path where external content actually arrives. What this does not prove is written into the module docs rather than left to be inferred: §B5's full acceptance drives a live Composio API through the sync pipeline, and that pipeline still sits behind engine-owned state (§B1, §B2). What is testable today is that the transform and the storage tier have no engine between them, which is the part §B3 was responsible for. The first draft of this test failed, which was useful: the fixture used the reshaped field names rather than the upstream ones, so the normaliser emitted nothing and the assertion caught an empty result rather than a wrong one. The fixture now uses the upstream shape (`messageId`, `sender`, `messageTimestamp`) and the assertions the reshaped one (`id`), which is the transform under test. Refs #18 (§B3, toward §B5 and acceptance criterion 6) --- Cargo.lock | 1 + Cargo.toml | 3 + tests/sync_on_a_foreign_driver.rs | 160 ++++++++++++++++++++++++++++++ 3 files changed, 164 insertions(+) create mode 100644 tests/sync_on_a_foreign_driver.rs diff --git a/Cargo.lock b/Cargo.lock index ce9b175..0c5ca9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1810,6 +1810,7 @@ dependencies = [ "serde_json", "tinymemory-api", "tinymemory-conformance", + "tinymemory-sync", "tokio", ] diff --git a/Cargo.toml b/Cargo.toml index fa90487..5b33af5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -80,6 +80,9 @@ tinymemory-api = { path = "api", features = ["test-support"] } # integration tests. A dev-dependency only: the facade must not carry a test # harness into a consumer's dependency graph. tinymemory-conformance = { path = "conformance" } +# The extracted Composio normalisers, for the integration test that runs them +# against a driver that is not TinyCortex (issue #18 §B3). +tinymemory-sync = { path = "sync" } [features] default = [] diff --git a/tests/sync_on_a_foreign_driver.rs b/tests/sync_on_a_foreign_driver.rs new file mode 100644 index 0000000..db77ccb --- /dev/null +++ b/tests/sync_on_a_foreign_driver.rs @@ -0,0 +1,160 @@ +//! Composio payloads normalised and stored through a driver that is not TinyCortex. +//! +//! Issue #18 §B3's stated purpose — "so a non-TinyCortex engine gets Composio +//! sync for free" — and the first half of §B5's acceptance test, "Composio +//! Gmail sync completes end to end against a driver that is not TinyCortex". +//! +//! # What this proves, and what it does not +//! +//! It proves the *coupling* is gone. Before §B3 these normalisers lived inside +//! the TinyCortex engine, and `tinymemory-core` reached in through +//! `tinycortex::memory::sync::composio::providers::normalize::*` to use them — +//! so a host binding a different engine could not run them at all. This file +//! links `tinymemory-sync` and a provider, and never names an engine. +//! +//! It does **not** prove the full §B5 acceptance test. That drives a live +//! Composio API through the sync pipeline; the pipeline itself still lives +//! behind engine-owned state (§B1, §B2), which is why §B5 stays open. What is +//! testable today is that the transform and the storage tier have no engine +//! between them, which is the part §B3 was responsible for. + +#![allow(clippy::expect_used, clippy::panic)] + +use std::sync::Arc; + +use serde_json::json; +use tinymemory::api::null::NullMemoryProvider; +use tinymemory::api::provider::{MemoryCore, MemoryProvider}; +use tinymemory::api::types::{MemoryCategory, MemoryTaint, GLOBAL_NAMESPACE}; +use tinymemory_conformance::InMemoryProvider; + +/// A raw Composio Gmail fetch response, in the shape the normaliser expects. +/// +/// Field names are the upstream ones (`messageId`, `sender`, `messageTimestamp`) +/// rather than the reshaped ones; turning the first into the second is the +/// transform under test. +fn raw_gmail_response() -> serde_json::Value { + json!({ + "messages": [ + { + "messageId": "msg-1", + "threadId": "t1", + "subject": "Lunch?", + "sender": "someone@example.com", + "to": "me@example.com", + "messageTimestamp": "2026-04-17T12:00:00Z", + "labelIds": ["INBOX"], + "messageText": "the cat sat on the mat", + "payload": {} + } + ], + "nextPageToken": "tok-1" + }) +} + +/// Stores every normalised message into `provider`, returning the keys written. +/// +/// The whole point of the exercise: this function is generic over the driver +/// and names no engine. +async fn ingest_into(provider: &dyn MemoryProvider, raw: serde_json::Value) -> Vec { + let mut data = raw; + tinymemory_sync::gmail_post_process::post_process("GMAIL_FETCH_EMAILS", None, &mut data); + + let messages = data + .get("messages") + .and_then(serde_json::Value::as_array) + .cloned() + .unwrap_or_default(); + + let mut written = Vec::new(); + for message in messages { + let key = message + .get("id") + .and_then(serde_json::Value::as_str) + .unwrap_or("unknown") + .to_owned(); + let content = serde_json::to_string(&message).expect("a normalised message serialises"); + provider + .store( + GLOBAL_NAMESPACE, + &key, + &content, + MemoryCategory::Core, + None, + // External by provenance: this came off somebody's inbox. A + // driver that laundered it to `Internal` is the failure the + // taint argument exists to prevent. + MemoryTaint::ExternalSync, + ) + .await + .expect("store into the bound driver"); + written.push(key); + } + written +} + +#[tokio::test] +async fn a_gmail_payload_normalises_and_stores_without_an_engine() { + let provider = InMemoryProvider::new(); + let written = ingest_into(&provider, raw_gmail_response()).await; + + assert_eq!( + written, + vec!["msg-1".to_string()], + "one message was written" + ); + + let stored = provider + .get(GLOBAL_NAMESPACE, "msg-1") + .await + .expect("read back") + .expect("the message was just stored"); + + assert!( + stored.content.contains("the cat sat on the mat"), + "the normalised body did not survive the round trip: {}", + stored.content + ); + assert_eq!( + stored.taint, + MemoryTaint::ExternalSync, + "provenance was laundered on the way in" + ); +} + +#[tokio::test] +async fn the_same_payload_runs_against_a_second_unrelated_driver() { + // The claim is "a non-TinyCortex engine", not "this one particular + // non-TinyCortex engine". Running the identical path against a driver with + // completely different retention semantics is what makes that general. + let provider = NullMemoryProvider::new(); + let written = ingest_into(&provider, raw_gmail_response()).await; + + assert_eq!(written, vec!["msg-1".to_string()]); + assert!( + provider + .get(GLOBAL_NAMESPACE, "msg-1") + .await + .expect("read back") + .is_none(), + "the null driver retains nothing, so the read must be empty — if this \ + returned a record the driver is not the one we think it is" + ); +} + +#[tokio::test] +async fn the_normaliser_is_reachable_without_naming_an_engine() { + // The structural assertion behind §B3. This file's dependencies are the + // facade, the conformance reference driver, and `tinymemory-sync`. If the + // normalisers still lived in the engine this would not compile, which is + // the whole test — the body below just keeps it from being vacuous. + let mut data = json!({ "messages": [] }); + tinymemory_sync::slack_post_process::post_process("SLACK_LIST_CONVERSATIONS", None, &mut data); + assert!( + data.is_object(), + "the slack normaliser should leave an object in place" + ); + + let arc: Arc = Arc::new(InMemoryProvider::new()); + assert_eq!(arc.driver_id(), "reference"); +} From df6f0b20d4c903634af4726d6760c80a3b5e5718 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 18 Aug 2026 13:10:22 +0530 Subject: [PATCH 3/4] Close two gaps found reviewing this issue's own changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both are places where a guard I added was weaker than the claim it was meant to support. **The awkward-content assertion had too low a floor.** §E1's run against TinyCortex found that the engine refuses empty content, which the contract explicitly permits — `MemoryCore::store` documents `Invalid` "for caller input the driver rejects" — so the assertion was corrected to allow a refusal. The guard against that becoming vacuous was `accepted > 0`, which is too weak: a driver accepting only `empty` and refusing unicode, large and newlines would have passed. The floor is now `unicode` specifically. Refusing `empty` is documented validation and refusing `large` is a defensible size limit, but refusing ordinary UTF-8 is a broken driver — and unicode is the case where mangling shows at all, since truncation and re-encoding are invisible on ASCII. All seven drivers still pass. **`tinymemory-sync` was unguarded.** That crate exists because it has no engine behind it (§B3): the Composio normalisers lived inside TinyCortex, and a host binding a different engine could not run them. Nothing enforced that after the extraction. A dependency added two crates away would put the coupling back silently — the build would stay green and the property would just stop being true. `dependency-budget.sh` now reports the crate (20 crates today) and fails if it reaches `tinycortex`, `rusqlite`, `libsqlite`, `tinymemory-core` or `tinymemory-api`. Verified in both directions: it passes now, and injecting `tinymemory-api = { path = "../api" }` makes it fire and name the offender. Refs #18 (§B3, §E1) --- conformance/src/suite/mod.rs | 19 ++++++++++++------- scripts/ci/dependency-budget.sh | 20 ++++++++++++++++++++ 2 files changed, 32 insertions(+), 7 deletions(-) diff --git a/conformance/src/suite/mod.rs b/conformance/src/suite/mod.rs index 1766887..64754dd 100644 --- a/conformance/src/suite/mod.rs +++ b/conformance/src/suite/mod.rs @@ -601,7 +601,7 @@ pub async fn assert_awkward_content_round_trips(provider: &dyn MemoryProvider) { ("large", "x".repeat(64 * 1024)), ("newlines", "a\nb\r\nc\0d".to_string()), ]; - let mut accepted = 0usize; + let mut accepted: Vec<&str> = Vec::new(); for (key, content) in &cases { // A driver may refuse a shape outright — `MemoryCore::store` documents // `Invalid` "for caller input the driver rejects", and the TinyCortex @@ -629,7 +629,7 @@ pub async fn assert_awkward_content_round_trips(provider: &dyn MemoryProvider) { { continue; } - accepted += 1; + accepted.push(key); if let Some(got) = provider .get(&ns, key) .await @@ -638,12 +638,17 @@ pub async fn assert_awkward_content_round_trips(provider: &dyn MemoryProvider) { assert_eq!(&got.content, content, "{who}: `{key}` content was mangled"); } } - // Without this a driver that refused every shape would pass having stored - // nothing, which is the vacuous reading of "may refuse". + // "May refuse" needs a floor, or a driver that refused everything would pass + // having stored nothing. The floor is `unicode` specifically rather than a + // count: refusing `empty` is documented validation, and refusing `large` is + // a defensible size limit, but refusing ordinary UTF-8 text is a broken + // driver — and unicode is the case where mangling actually shows, since + // truncation and re-encoding are invisible on ASCII. assert!( - accepted > 0, - "{who}: refused every content shape — unicode, empty, large and \ - newlines were all rejected, so this assertion proved nothing" + accepted.contains(&"unicode"), + "{who}: refused ordinary UTF-8 content — accepted {accepted:?}. A driver \ + may refuse a shape, but not this one; every assertion about content \ + surviving a round trip rests on it." ); let keys: Vec<&str> = cases.iter().map(|(k, _)| *k).collect(); cleanup(provider, &ns, &keys).await; diff --git a/scripts/ci/dependency-budget.sh b/scripts/ci/dependency-budget.sh index 80c4e54..430fb20 100755 --- a/scripts/ci/dependency-budget.sh +++ b/scripts/ci/dependency-budget.sh @@ -38,6 +38,26 @@ printf '%-52s %s\n' "tinymemory-api" "$(count -p tinymemory-api)" printf '%-52s %s\n' "tinymemory-tinycortex (default)" "$(count -p tinymemory-tinycortex --no-default-features)" printf '%-52s %s\n' "tinymemory-tinycortex --features memory-git" "$(count -p tinymemory-tinycortex --features memory-git)" printf '%-52s %s\n' "tinymemory-remote" "$(count -p tinymemory-remote)" +printf '%-52s %s\n' "tinymemory-sync" "$(count -p tinymemory-sync)" + +# The sync crate exists because it has no engine behind it (issue #18 §B3): +# the Composio normalisers lived inside TinyCortex, and a host binding a +# different engine could not run them. Nothing else enforces that, and a +# dependency added two crates away would reintroduce the coupling silently — +# the build would still be green, and the property would just quietly stop +# being true. +sync_engine="$( + cargo tree -p tinymemory-sync -e normal --prefix none 2>/dev/null \ + | grep -Ei '^(tinycortex|rusqlite|libsqlite|tinymemory-core|tinymemory-api)' || true +)" +if [ -n "$sync_engine" ]; then + echo "tinymemory-sync reached an engine, a store, or the contract:" >&2 + echo "$sync_engine" >&2 + echo >&2 + echo "That crate is the one piece of Composio sync a non-TinyCortex host can" >&2 + echo "use. A dependency on any of the above puts it back behind an engine." >&2 + exit 1 +fi echo if [ "$minimal" -gt "$MINIMAL_CEILING" ]; then From 7cba2091963bb24f75c32d6f955abe956248d669 Mon Sep 17 00:00:00 2001 From: Shanu Date: Tue, 18 Aug 2026 16:35:21 +0530 Subject: [PATCH 4/4] Re-point the tinycortex gitlink at the merged commit tinycortex#149 landed as a squash (8401346b), discarding the branch head this pin pointed at; 34cbb6c is diverged from tinycortex main rather than an ancestor of it. The merged commit is also the one that deletes the 33 duplicated files under api/src/, which is the state this stack depends on. --- vendor/tinycortex | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinycortex b/vendor/tinycortex index 34cbb6c..8401346 160000 --- a/vendor/tinycortex +++ b/vendor/tinycortex @@ -1 +1 @@ -Subproject commit 34cbb6cfa91ea74d62605bd57790782b0c748556 +Subproject commit 8401346b574cacb1dc0cf6b36bc608ff5ef9f6f5