Repository navigation
feat(composio/slack): port Gmail-style ingest pipeline, retire memory/slack_ingestion - #1354
Merged
Merged
Conversation
…/slack_ingestion
- New `composio/providers/slack/{ingest,post_process,types,rpc,schemas}.rs`
mirror Gmail's directory shape: per-message ingest, raw-archive write,
`set_chunk_raw_refs` linkage, slug-dispatched post-process.
- Drop the 6h UTC bucketer entirely — L0 seal cascade (PR tinyhumansai#1348) handles
batching now. Provider feeds messages straight to
`ingest_page_into_memory_tree` per channel.
- Source id stays `slack:{connection_id}` (workspace-wide); channel
context lives on `SlackMessage` and in `ChatBatch.channel_label`.
- `slack_memory.*` JSON-RPC namespace moves from `memory/slack_ingestion/`
into `composio/providers/slack/`; `core/all.rs` registration updated.
- `memory/slack_ingestion/` deleted (7 files, ~960 LOC).
Validated: cargo check clean; `composio::providers::slack` 43 tests pass;
`memory::tree::` 527 tests pass; cargo fmt clean.
…merge PR tinyhumansai#1351 added a `kind` field to `RawItem` and a `raw_rel_path` helper that routes per-kind subdirs (`emails/`, `chats/`, …). Update Slack's ingest to: - Tag `RawItem` with `RawKind::Chat` (raw archive now lands under `raw/<source_slug>/chats/<ts_ms>_<ts_raw>.md`). - Use `raw_rel_path` instead of building paths by hand — drops the local `sanitize_uid_for_path` helper since `write_raw_items` does the sanitization centrally now. - Drop the `slugify_source_id` import (no longer needed). Validated: cargo check clean; 43 slack tests pass; 534 memory-tree tests pass; cargo fmt clean.
Contributor
|
ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughThis pull request migrates Slack ingestion from 6‑hour UTC buckets to per‑channel page ingestion, consolidates Slack logic under composio::providers::slack, adds post-processing that normalizes provider responses, introduces canonical Slack types, updates RPC/schemas, and removes legacy bucketing code. ChangesSlack Ingestion Refactoring
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Comment |
Member
Author
|
@CodeRabbit review |
Contributor
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
src/openhuman/composio/providers/slack/ingest.rs (1)
163-183: 💤 Low valueConsider hoisting the tags allocation outside the loop.
The
tagsvector is recreated identically for every message iteration. Moving it before the loop would avoid repeated allocations:async fn ingest_per_message( config: &Config, source_id: &str, owner: &str, page_messages: &[SlackMessage], ) -> usize { let mut total_chunks = 0usize; + let tags: Vec<String> = DEFAULT_TAGS.iter().map(|s| (*s).to_string()).collect(); for m in page_messages { // ... - let tags = DEFAULT_TAGS.iter().map(|s| (*s).to_string()).collect(); + let tags_clone = tags.clone(); - match ingest_chat(config, source_id, owner, tags, batch).await { + match ingest_chat(config, source_id, owner, tags_clone, batch).await {This is minor given the API rate-limited throughput.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/composio/providers/slack/ingest.rs` around lines 163 - 183, The tags vector is allocated anew for each iteration inside the for m in page_messages loop; hoist its creation outside the loop by moving let tags = DEFAULT_TAGS.iter().map(|s| (*s).to_string().collect() before the for so the same tags Vec<String> is reused for every message. Update code to remove the inner let tags = ... and reference the outer tags inside the loop; keep the original mapping logic and the same variable name to minimize changes (symbols: DEFAULT_TAGS, tags, page_messages, ChatBatch).src/openhuman/composio/providers/slack/provider.rs (1)
753-755: 💤 Low valueConsider avoiding the clone in backfill path.
The current code clones each message when converting from
Vec<&SlackMessage>toVec<SlackMessage>. While acceptable for a one-shot backfill operation, you could restructure to pass references ifingest_page_into_memory_treeaccepted&[&SlackMessage]or similar.This is minor given the backfill context.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/composio/providers/slack/provider.rs` around lines 753 - 755, The backfill currently clones each SlackMessage into a new Vec<SlackMessage> before calling ingest_page_into_memory_tree; change ingest_page_into_memory_tree to accept a slice of references (e.g., &[&SlackMessage]) so you can pass msgs_for_channel.as_slice() (or construct a Vec<&SlackMessage> by mapping references) instead of cloning, and update all call sites of ingest_page_into_memory_tree and its signature accordingly (refer to ingest_page_into_memory_tree, msgs_for_channel, buckets, page, connection_id, and ctx.config to locate the changes).src/openhuman/composio/providers/slack/rpc.rs (1)
45-108: ⚡ Quick winAdd debug/trace instrumentation around the new Slack RPC flows.
These handlers only log failures plus the terminal
single_log, so connection filtering, provider discovery, and state-load skips are hard to reconstruct when a sync behaves unexpectedly. Please add entry/exit and key branch diagnostics with stable[composio:slack][rpc]-style prefixes and correlation fields.As per coding guidelines,
Use log / tracing at debug or trace level on RPC entry/exit, error paths, state transitions, and any branch that is hard to infer from tests; include stable prefixes (e.g. [domain], [rpc]) and correlation fields.Also applies to: 136-183
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/composio/providers/slack/rpc.rs` around lines 45 - 108, Add debug/trace instrumentation to sync_trigger_rpc: log an entry message at start with prefix "[composio:slack][rpc]" and include correlation fields (e.g., req.connection_id and total connections), debug the provider discovery result (provider existence), the Composio client creation, the list_connections result (count), the candidate filtering steps (before/after filtering and when req.connection_id narrows results), each skipped/no-op branch (no candidates found), and an exit message with outcomes summary; use log::debug or log::trace for these messages and keep existing error/warn handling for provider.sync errors (log when provider.sync is invoked and when it returns Err), referencing the symbols sync_trigger_rpc, provider, build_composio_client/client, connections, candidates, req.connection_id, outcomes, ProviderContext, and provider.sync so the logs can reconstruct the flow.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/openhuman/composio/providers/slack/sync.rs`:
- Around line 202-205: parse_ts currently truncates the Slack ts fractional part
and ChannelCursors persists only unix-seconds, causing duplicate replays; change
the cursor handling so you persist the full Slack ts string (or another lossless
representation) instead of seconds, stop converting to whole seconds in
parse_ts, and use the exact stored ts as the resume/oldest value in the sync
logic (references: parse_ts, ChannelCursors, and the code paths that set/consume
oldest around the sync fetch logic and the other parse_ts usage at the second
location noted).
- Around line 168-176: The code currently allows an empty channel_id and maps it
to a blank channel, causing malformed SlackMessage records; change the logic
around channel_id (the raw.get("channel_id") -> as_str() -> to_string()
sequence) to treat missing or empty channel_id as an invalid hit and skip/return
early instead of creating a SlackMessage: after extracting channel_id, if it's
None or channel_id.trim().is_empty() then bail out (e.g., return None or
continue) so you never consult channel_map or construct a SlackMessage for empty
channels; update any calling code that expects an Option to handle the skipped
case accordingly.
---
Nitpick comments:
In `@src/openhuman/composio/providers/slack/ingest.rs`:
- Around line 163-183: The tags vector is allocated anew for each iteration
inside the for m in page_messages loop; hoist its creation outside the loop by
moving let tags = DEFAULT_TAGS.iter().map(|s| (*s).to_string().collect() before
the for so the same tags Vec<String> is reused for every message. Update code to
remove the inner let tags = ... and reference the outer tags inside the loop;
keep the original mapping logic and the same variable name to minimize changes
(symbols: DEFAULT_TAGS, tags, page_messages, ChatBatch).
In `@src/openhuman/composio/providers/slack/provider.rs`:
- Around line 753-755: The backfill currently clones each SlackMessage into a
new Vec<SlackMessage> before calling ingest_page_into_memory_tree; change
ingest_page_into_memory_tree to accept a slice of references (e.g.,
&[&SlackMessage]) so you can pass msgs_for_channel.as_slice() (or construct a
Vec<&SlackMessage> by mapping references) instead of cloning, and update all
call sites of ingest_page_into_memory_tree and its signature accordingly (refer
to ingest_page_into_memory_tree, msgs_for_channel, buckets, page, connection_id,
and ctx.config to locate the changes).
In `@src/openhuman/composio/providers/slack/rpc.rs`:
- Around line 45-108: Add debug/trace instrumentation to sync_trigger_rpc: log
an entry message at start with prefix "[composio:slack][rpc]" and include
correlation fields (e.g., req.connection_id and total connections), debug the
provider discovery result (provider existence), the Composio client creation,
the list_connections result (count), the candidate filtering steps (before/after
filtering and when req.connection_id narrows results), each skipped/no-op branch
(no candidates found), and an exit message with outcomes summary; use log::debug
or log::trace for these messages and keep existing error/warn handling for
provider.sync errors (log when provider.sync is invoked and when it returns
Err), referencing the symbols sync_trigger_rpc, provider,
build_composio_client/client, connections, candidates, req.connection_id,
outcomes, ProviderContext, and provider.sync so the logs can reconstruct the
flow.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3135c403-6fed-4c57-ad51-68bba8d32164
📒 Files selected for processing (16)
src/core/all.rssrc/openhuman/composio/providers/slack/ingest.rssrc/openhuman/composio/providers/slack/mod.rssrc/openhuman/composio/providers/slack/post_process.rssrc/openhuman/composio/providers/slack/post_process_tests.rssrc/openhuman/composio/providers/slack/provider.rssrc/openhuman/composio/providers/slack/rpc.rssrc/openhuman/composio/providers/slack/schemas.rssrc/openhuman/composio/providers/slack/sync.rssrc/openhuman/composio/providers/slack/types.rssrc/openhuman/memory/mod.rssrc/openhuman/memory/slack_ingestion/README.mdsrc/openhuman/memory/slack_ingestion/bucketer.rssrc/openhuman/memory/slack_ingestion/mod.rssrc/openhuman/memory/slack_ingestion/ops.rssrc/openhuman/memory/slack_ingestion/types.rs
💤 Files with no reviewable changes (6)
- src/openhuman/memory/slack_ingestion/README.md
- src/openhuman/memory/slack_ingestion/mod.rs
- src/openhuman/memory/slack_ingestion/types.rs
- src/openhuman/memory/slack_ingestion/bucketer.rs
- src/openhuman/memory/slack_ingestion/ops.rs
- src/openhuman/memory/mod.rs
…channel_id Two CodeRabbit findings on PR tinyhumansai#1354: 1. **Cursor precision** (Major). `ChannelCursors` was `BTreeMap<String, i64>` storing whole unix seconds, and `process_channel` advanced the cursor from `m.timestamp.timestamp()`. Slack `ts` is `<seconds>.<micro>` per- workspace sequence — channels with multiple messages in the same wall- clock second would replay that whole second on the next incremental fetch (`oldest = "<sec>.000000"` with `inclusive=false`). Switch the map value to the **raw Slack `ts` string** so full microsecond precision round-trips. Cursor lookup uses the stored string verbatim as `oldest`. New `parse_ts_components(ts) -> (i64, u64)` helper picks the latest message by `(seconds, micros)` tuple instead of relying on lexicographic-equals-numeric for the raw string. Migration: persisted state from before this commit fails to deserialize the new shape; `decode_cursors` already soft-falls to an empty map, so the next sync re-walks the backfill window once. No data loss — chunk IDs are content-hashed. 2. **Empty `channel_id` in search hits** (Minor). `parse_search_match` accepted an empty channel id and emitted a `SlackMessage` with `channel_id=""`, which the new ingest path would then funnel into a single empty-channel bucket. Added a `?` filter so malformed matches are dropped at parse time. Tests: +2 (`extract_search_messages_drops_match_with_missing_channel_id`, `parse_ts_components_preserves_microseconds`). 45 slack tests pass.
AusAgentSmith
pushed a commit
to AusAgentSmith/openhuman
that referenced
this pull request
May 23, 2026
senamakel
added a commit
to nocstah/openhuman
that referenced
this pull request
Sep 11, 2026
…/slack_ingestion (tinyhumansai#1354)\n
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
composio/providers/slack/{ingest,post_process,types,rpc,schemas}.rsmirror Gmail's directory layout (per-message ingest, raw-archive write,set_chunk_raw_refslinkage, slug-dispatched post-process).memory/slack_ingestion/module entirely (7 files, ~960 LOC removed). Theslack_memory.*JSON-RPC namespace moves tocomposio/providers/slack/;core/all.rsregistration updated accordingly.ingest_page_into_memory_treeper channel.RawKind::Chat, archive lands underraw/<source_slug>/chats/<ts_ms>_<ts_raw>.md.Problem
Slack memory ingestion lived in its own
memory/slack_ingestion/module that didn't share shape with Gmail'scomposio/providers/gmail/{ingest,post_process}.rs. The split meant:sync.rs::extract_messagesrather than a dedicated slug-dispatched step like Gmail'spost_process.rs.Solution
Composio is the source path for both providers (the webview path in
channels/providers/slack.rsis agent-channel-only; never fed memory). So this is a layout/shape parity job, not an integration-source change.New under
composio/providers/slack/:types.rs—SlackMessage(now carrieschannel_id,channel_name,is_private,author_id,permalinkso canonicalisation has full context without a separateSlackChannelplumbed through),SlackChannel. OldBucketstruct dropped.ingest.rs—pub async fn ingest_page_into_memory_tree(config, owner, connection_id, &[SlackMessage]) -> Result<usize>. Per-message ingest (oneingest_chatcall per Slack message, single-messageChatBatch), raw-archive write viawrite_raw_itemswithRawKind::Chat, thenset_chunk_raw_refslinking each chunk to its raw entry. Helpers:bucket_by_channel,channel_label,slack_message_to_chat_message,write_raw_archive.post_process.rs— slug-dispatched data reshaper forSLACK_FETCH_CONVERSATION_HISTORY,SLACK_LIST_CONVERSATIONS,SLACK_SEARCH_MESSAGES. Pure data; user-id resolution stays in the ingest layer where the per-syncSlackUserscache lives (matches Gmail's "post_process is data-only" pattern).post_process_tests.rs— 9 cases covering all three slugs, nested vs top-level envelopes, edge cases.rpc.rs+schemas.rs— moved verbatim frommemory/slack_ingestion/; namespaceslack_memorypreserved; exports renamed toall_slack_memory_*.Updated:
provider.rs—post_process_action_resultimpl added;process_channelandrun_backfill_via_searchnow callpost_process→extract_messages→ingest_page_into_memory_tree(no buckets).sync.rs— slimmed to thin enrichers over the post-processed shape.memory/mod.rs,core/all.rs—slack_memorycontrollers now register fromcomposio::providers::slack.Deleted:
src/openhuman/memory/slack_ingestion/(entire directory).Submission Checklist
docs/TESTING-STRATEGY.mdsrc/openhuman/composio/providers/slack/...; ≥ 80% diff coverage is enforced viacargo-llvm-covand lcov merge in CI.docs/TESTING-STRATEGY.md)Impact
slack_memory.sync_trigger/sync_statusJSON-RPC methods unchanged in shape; only registration site moved. Source idslack:{connection_id}preserved (workspace-wide). ExistingRawRefpaths from prior runs still resolve (per feat(memory): kind-aware raw layout, _source.md mirror, source/<slug> tag #1351's parity contract).Related
RawKind::Chat).thread_ts(Gmail has the same gap today).AI Authored PR Metadata (required for Codex/Linear PRs)
Linear Issue
Commit & Branch
Validation Run
cargo test --lib composio::providers::slack→ 43 passed;cargo test --lib memory::tree::→ 534 passed.cargo fmt --checkclean;cargo checkclean (only pre-existing warnings).pnpm typecheck— no app-side changes in this PR.Validation Blocked
command:N/Aerror:N/Aimpact:N/ABehavior Changes
memory/slack_ingestion/module retired; bucketer dropped in favor of L0 seal cascade.raw/<source_slug>/chats/; leaf summaries form within a sync tick instead of waiting for 6h windows to close.Parity Contract
slack:{connection_id}source_id unchanged;slack_memory.*RPC shape unchanged; chunk-id content-hashing makes the layout change a no-op UPSERT for already-ingested data; existingRawRefpaths still resolve (feat(memory): kind-aware raw layout, _source.md mirror, source/<slug> tag #1351 parity).SlackUserssoft-fallback to raw ids preserved;execute_with_retryrate-limit retry preserved;extract_with_retry's daily budget accounting preserved.Duplicate / Superseded PR Handling
Summary by CodeRabbit
New Features
Refactor
Tests