fix(channels): isolate thread subscriptions per agent#17049
fix(channels): isolate thread subscriptions per agent#17049Akash504-ai wants to merge 3 commits into
Conversation
🦋 Changeset detectedLatest commit: e9893b0 The changes in this PR will be included in the next version bump. This PR includes changesets to release 20 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
|
@Akash504-ai is attempting to deploy a commit to the Mastra Team on Vercel. A member of the Team first needs to authorize it. |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
WalkthroughScope external channel thread IDs by agent: MastraStateAdapter now stores an agentId, looks up and persists channel_externalThreadId as ChangesAgent-scoped channel subscription isolation
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Suggested reviewers
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 ESLint
ESLint install failed: lockfile failed supply-chain policy check. Run Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
packages/core/src/channels/agent-channels.ts (1)
1360-1371:⚠️ Potential issue | 🟠 Major | ⚡ Quick winAdd legacy thread-id fallback to avoid post-upgrade thread split.
Line 1368 now looks up only
agentId:externalThreadId. Existing installations with legacy unscoped metadata will miss matches, causing duplicate thread creation and loss of prior thread continuity until data is migrated.Suggested compatibility patch
const scopedExternalThreadId = `${this.agent.id}:${externalThreadId}`; const metadata = { channel_platform: platform, channel_externalThreadId: scopedExternalThreadId, channel_externalChannelId: channelId, }; -const { threads } = await memoryStore.listThreads({ +let { threads } = await memoryStore.listThreads({ filter: { metadata }, perPage: 1, }); + +if (threads.length === 0) { + const legacyMetadata = { + channel_platform: platform, + channel_externalThreadId: externalThreadId, + channel_externalChannelId: channelId, + }; + ({ threads } = await memoryStore.listThreads({ + filter: { metadata: legacyMetadata }, + perPage: 1, + })); + + if (threads[0]) { + await memoryStore.updateThread({ + id: threads[0].id, + title: threads[0].title ?? '', + metadata: { ...(threads[0].metadata ?? {}), channel_externalThreadId: scopedExternalThreadId }, + }); + } +}🤖 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 `@packages/core/src/channels/agent-channels.ts` around lines 1360 - 1371, The current lookup uses only the scopedExternalThreadId (`${this.agent.id}:${externalThreadId}`) which breaks older installations that stored unscoped thread IDs; update the logic around the call to memoryStore.listThreads (and the surrounding metadata construction) to fall back to legacy IDs: first search using metadata.channel_externalThreadId = scopedExternalThreadId, and if no threads are returned, retry using metadata.channel_externalThreadId = externalThreadId (unscoped), or alternatively build a filter that matches either value (OR) if the memoryStore.listThreads API supports it; ensure any downstream code that uses the found thread handles the result of the second query the same way.
🤖 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 `@packages/core/src/channels/__tests__/state-adapter.test.ts`:
- Around line 83-109: The test currently only saves a thread mapped to agent-a,
so agentB's isSubscribed can be false by absence; add a separate saved thread
for agent-b using memoryStore.saveThread with channel_externalThreadId
`agent-b:${externalThreadId}` (mirroring the agent-a saveThread block but with
id/title/resourceId/metadata namespaced to agent-b), then explicitly exercise
subscribe/isSubscribed for both adapters: call await
agentA.subscribe(externalThreadId) and assert
agentA.isSubscribed(externalThreadId) is true while
agentB.isSubscribed(externalThreadId) is false, and optionally call await
agentB.subscribe(externalThreadId) and assert
agentB.isSubscribed(externalThreadId) becomes true to confirm isolation; locate
changes around MastraStateAdapter usage, memoryStore.saveThread, subscribe, and
isSubscribed.
---
Outside diff comments:
In `@packages/core/src/channels/agent-channels.ts`:
- Around line 1360-1371: The current lookup uses only the scopedExternalThreadId
(`${this.agent.id}:${externalThreadId}`) which breaks older installations that
stored unscoped thread IDs; update the logic around the call to
memoryStore.listThreads (and the surrounding metadata construction) to fall back
to legacy IDs: first search using metadata.channel_externalThreadId =
scopedExternalThreadId, and if no threads are returned, retry using
metadata.channel_externalThreadId = externalThreadId (unscoped), or
alternatively build a filter that matches either value (OR) if the
memoryStore.listThreads API supports it; ensure any downstream code that uses
the found thread handles the result of the second query the same way.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 5aaac9a0-ea4c-4e2b-b0b8-94ebd35ff425
📒 Files selected for processing (4)
.changeset/agent-channel-subscription-isolation.mdpackages/core/src/channels/__tests__/state-adapter.test.tspackages/core/src/channels/agent-channels.tspackages/core/src/channels/state-adapter.ts
1 participant
Description
Fixes channel subscription state leaking across agents/bots sharing the same Mastra storage adapter.
Previously,
MastraStateAdapterlooked up subscriptions only bychannel_externalThreadId, which caused multiple agents using the same storage backend to share subscription state unintentionally. As a result, different bots could respond to the same incoming message if they shared the same external thread ID (such as a TelegramchatId).This PR scopes external thread IDs by
agentIdwhen storing and resolving thread subscriptions, ensuring subscriptions remain isolated per agent.Also adds regression tests covering multi-agent isolation behavior.
Related issue(s)
Fixes #17037
Type of change
Checklist
ELI5
Two different bots were accidentally sharing "who's listening" notes because they used the same storage. This change tags each bot's notes with its own name so bots no longer see each other's subscriptions.
What Changed
This PR fixes a bug where channel thread subscription state leaked across agents that share the same Mastra storage backend by scoping stored external thread IDs to the agent that owns them. Previously, MastraStateAdapter looked up threads only by channel_externalThreadId, causing subscriptions to be treated as global across agents sharing the same storage. The adapter now prefixes external thread IDs with the agent ID (format:
${agentId}:${externalThreadId}) when storing and resolving thread mappings so subscription state remains isolated per agent.Files Modified
packages/core/src/channels/state-adapter.ts
packages/core/src/channels/agent-channels.ts
packages/core/src/channels/tests/state-adapter.test.ts
.changeset/agent-channel-subscription-isolation.md
@mastra/coredescribing the fix.Behavior Change
Subscriptions are now agent-scoped: multiple agents sharing the same storage backend will no longer see or inherit each other's channel/thread subscription state.
Tests
Added and updated unit tests covering: