diff --git a/Cargo.lock b/Cargo.lock
index 5caf806..0c5ca9c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1810,6 +1810,7 @@ dependencies = [
  "serde_json",
  "tinymemory-api",
  "tinymemory-conformance",
+ "tinymemory-sync",
  "tokio",
 ]
 
@@ -1867,6 +1868,7 @@ dependencies = [
  "tinycortex-api",
  "tinymemory",
  "tinymemory-api",
+ "tinymemory-sync",
  "tokio",
  "tracing",
  "url",
@@ -1891,6 +1893,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..5b33af5 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.
@@ -79,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/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/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/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
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<Value> {
+    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<String> {
+    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<String> {
+    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<String> {
+    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<String> {
+    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<Value> {
+    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<String> {
+    // 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<String> {
+    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<String> {
+    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<String> {
+    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<String> {
+    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<Value> = 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<Vec<String>> {
+    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<Vec<String>> {
+    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<String> = 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<Value> = 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<chrono::DateTime<chrono::Utc>> {
+    date_str
+        .parse::<chrono::DateTime<chrono::Utc>>()
+        .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<Tz: chrono::TimeZone>(
+    utc: chrono::DateTime<chrono::Utc>,
+    tz: &Tz,
+) -> Option<String>
+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<String> {
+    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<String, Value>, name: &str) -> Option<Value> {
+    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, Value>) -> 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<String, Value>) -> Vec<Value> {
+    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<String> {
+    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<Value> {
+    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<String> {
+    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<String> {
+    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<Value> {
+    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<String> {
+    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<String> {
+    // 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<Value> {
+    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<String> {
+    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<String> {
+    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<String> {
+    // 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::<Vec<_>>()
+                            .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<Value> = 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<Value> {
+    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<Value> {
+    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<Value> = 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<Value> {
+    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<Value> = 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<Value> {
+    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<String, Value> {
+    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");
+}
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<String> {
+    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<dyn MemoryProvider> = Arc::new(InMemoryProvider::new());
+    assert_eq!(arc.driver_id(), "reference");
+}