Fimeg/lettabot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #542 from letta-ai/fix/thread-only-per-thread-conversations

Browse Source
This commit is contained in:
Cameron
2026-03-09 18:47:14 -07:00
parent 69cd7e5225 be60a00057
commit 15b98758cc
10 changed files with 63 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/configuration.md
View File

@@ -461,6 +461,8 @@ channels:
Thread messages inherit parent channel config, so child threads under `EZRA_CHANNEL_ID` use the same group rules.
When `threadMode: thread-only` is set, each thread automatically gets its own isolated conversation (message history). This overrides `shared` and `per-channel` conversation modes so that messages from different threads are never interleaved. Agent memory (blocks) is still shared across all threads. (In `disabled` mode, all messages use the agent's built-in default conversation and thread isolation does not apply.)
### Finding Group IDs
Each channel uses different identifiers for groups:

1
docs/discord-setup.md
View File

@@ -209,6 +209,7 @@ Behavior summary:
- Top-level messages are ignored in `thread-only` mode.
- Top-level @mentions create a thread and are answered in that thread when `autoCreateThreadOnMention` is enabled.
- Thread messages inherit parent channel config. If `EZRA_CHANNEL_ID` is configured, replies in its child threads use that same config.
- Each thread gets its own isolated conversation (message history), overriding `shared` and `per-channel` conversation modes. This prevents crosstalk between threads. Agent memory (blocks) is still shared.
Required Discord permissions for auto-create:

4
src/channels/discord-adapter.test.ts
View File

@@ -185,7 +185,7 @@ describe('DiscordAdapter command gating', () => {
await client!.emit('messageCreate', message);
expect(onCommand).toHaveBeenCalledTimes(1);
expect(onCommand).toHaveBeenCalledWith('status', 'thread-1', undefined);
expect(onCommand).toHaveBeenCalledWith('status', 'thread-1', undefined, true);
expect(message.channel.send).toHaveBeenCalledWith('ok');
await adapter.stop();
});
@@ -222,7 +222,7 @@ describe('DiscordAdapter command gating', () => {
expect(message.startThread).toHaveBeenCalledTimes(1);
expect(onCommand).toHaveBeenCalledTimes(1);
expect(onCommand).toHaveBeenCalledWith('status', 'thread-created', undefined);
expect(onCommand).toHaveBeenCalledWith('status', 'thread-created', undefined, true);
expect(threadSend).toHaveBeenCalledWith('ok');
expect(message.channel.send).not.toHaveBeenCalled();
await adapter.stop();

16
src/channels/discord.ts
View File

@@ -115,7 +115,7 @@ export class DiscordAdapter implements ChannelAdapter {
private attachmentsMaxBytes?: number;
onMessage?: (msg: InboundMessage) => Promise<void>;
onCommand?: (command: string, chatId?: string, args?: string) => Promise<string | null>;
onCommand?: (command: string, chatId?: string, args?: string, forcePerChat?: boolean) => Promise<string | null>;
constructor(config: DiscordConfig) {
this.config = {
@@ -330,9 +330,11 @@ Ask the bot owner to approve with:
? (message.channel as { send: (content: string) => Promise<unknown> })
: null;
let commandForcePerChat = false;
if (isGroup && this.config.groups) {
const threadMode = resolveDiscordThreadMode(this.config.groups, keys);
if (threadMode === 'thread-only' && !isThreadMessage) {
commandForcePerChat = threadMode === 'thread-only';
if (commandForcePerChat && !isThreadMessage) {
const shouldCreateThread =
wasMentioned && resolveDiscordAutoCreateThreadOnMention(this.config.groups, keys);
if (!shouldCreateThread) {
@@ -365,7 +367,7 @@ Ask the bot owner to approve with:
}
if (this.onCommand && isManagedCommand) {
const result = await this.onCommand(command, commandChatId, cmdArgs);
const result = await this.onCommand(command, commandChatId, cmdArgs, commandForcePerChat || undefined);
if (result) {
if (!commandSendTarget) return;
await commandSendTarget.send(result);
@@ -381,6 +383,7 @@ Ask the bot owner to approve with:
let isListeningMode = false;
let effectiveChatId = message.channel.id;
let effectiveGroupName = groupName;
let isThreadOnly = false;
// Group gating: config-based allowlist + mode
if (isGroup && this.config.groups) {
@@ -414,7 +417,8 @@ Ask the bot owner to approve with:
}
const threadMode = resolveDiscordThreadMode(this.config.groups, keys);
if (threadMode === 'thread-only' && !isThreadMessage) {
isThreadOnly = threadMode === 'thread-only';
if (isThreadOnly && !isThreadMessage) {
const shouldCreateThread =
wasMentioned && resolveDiscordAutoCreateThreadOnMention(this.config.groups, keys);
if (!shouldCreateThread) {
@@ -444,6 +448,7 @@ Ask the bot owner to approve with:
serverId: message.guildId || undefined,
wasMentioned,
isListeningMode,
forcePerChat: isThreadOnly || undefined,
attachments,
formatterHints: this.getFormatterHints(),
});
@@ -616,6 +621,7 @@ Ask the bot owner to approve with:
}
let isListeningMode = false;
let reactionForcePerChat = false;
if (isGroup && this.config.groups) {
if (!isGroupAllowed(this.config.groups, keys)) {
log.info(`Reaction group ${channelId} not in allowlist, ignoring`);
@@ -636,6 +642,7 @@ Ask the bot owner to approve with:
if (threadMode === 'thread-only' && !isThreadMessage) {
return;
}
reactionForcePerChat = threadMode === 'thread-only';
const limits = resolveDailyLimits(this.config.groups, keys);
const counterScope = limits.matchedKey ?? channelId;
@@ -674,6 +681,7 @@ Ask the bot owner to approve with:
groupName,
serverId: message.guildId || undefined,
isListeningMode,
forcePerChat: reactionForcePerChat || undefined,
reaction: {
emoji,
messageId: message.id,

2
src/channels/types.ts
View File

@@ -33,7 +33,7 @@ export interface ChannelAdapter {
// Event handlers (set by bot core)
onMessage?: (msg: InboundMessage) => Promise<void>;
onCommand?: (command: string, chatId?: string, args?: string) => Promise<string | null>;
onCommand?: (command: string, chatId?: string, args?: string, forcePerChat?: boolean) => Promise<string | null>;
}
/**

29
src/core/bot.ts
View File

@@ -150,10 +150,11 @@ export function resolveConversationKey(
conversationMode: string | undefined,
conversationOverrides: Set<string>,
chatId?: string,
forcePerChat?: boolean,
): string {
if (conversationMode === 'disabled') return 'default';
const normalized = channel.toLowerCase();
if (conversationMode === 'per-chat' && chatId) return `${normalized}:${chatId}`;
if ((conversationMode === 'per-chat' || forcePerChat) && chatId) return `${normalized}:${chatId}`;
if (conversationMode === 'per-channel') return normalized;
if (conversationOverrides.has(normalized)) return normalized;
return 'shared';
@@ -562,8 +563,8 @@ export class LettaBot implements AgentSession {
* Returns 'shared' in shared mode (unless channel is in perChannel overrides).
* Returns channel id in per-channel mode or for override channels.
*/
private resolveConversationKey(channel: string, chatId?: string): string {
return resolveConversationKey(channel, this.config.conversationMode, this.conversationOverrides, chatId);
private resolveConversationKey(channel: string, chatId?: string, forcePerChat?: boolean): string {
return resolveConversationKey(channel, this.config.conversationMode, this.conversationOverrides, chatId, forcePerChat);
}
/**
@@ -604,7 +605,7 @@ export class LettaBot implements AgentSession {
registerChannel(adapter: ChannelAdapter): void {
adapter.onMessage = (msg) => this.handleMessage(msg, adapter);
adapter.onCommand = (cmd, chatId, args) => this.handleCommand(cmd, adapter.id, chatId, args);
adapter.onCommand = (cmd, chatId, args, forcePerChat) => this.handleCommand(cmd, adapter.id, chatId, args, forcePerChat);
// Wrap outbound methods when any redaction layer is active.
// Secrets are enabled by default unless explicitly disabled.
@@ -650,7 +651,7 @@ export class LettaBot implements AgentSession {
}
}
const convKey = this.resolveConversationKey(effective.channel, effective.chatId);
const convKey = this.resolveConversationKey(effective.channel, effective.chatId, effective.forcePerChat);
if (convKey !== 'shared') {
this.enqueueForKey(convKey, effective, adapter);
} else {
@@ -665,7 +666,7 @@ export class LettaBot implements AgentSession {
// Commands
// =========================================================================
private async handleCommand(command: string, channelId?: string, chatId?: string, args?: string): Promise<string | null> {
private async handleCommand(command: string, channelId?: string, chatId?: string, args?: string, forcePerChat?: boolean): Promise<string | null> {
log.info(`Received: /${command}${args ? ` ${args}` : ''}`);
switch (command) {
case 'status': {
@@ -693,7 +694,7 @@ export class LettaBot implements AgentSession {
// other channels/chats' conversations are never silently destroyed.
// resolveConversationKey returns 'shared' for non-override channels,
// the channel id for per-channel, or channel:chatId for per-chat.
const convKey = channelId ? this.resolveConversationKey(channelId, chatId) : 'shared';
const convKey = channelId ? this.resolveConversationKey(channelId, chatId, forcePerChat) : 'shared';
// In disabled mode the bot always uses the agent's built-in default
// conversation -- there's nothing to reset locally.
@@ -724,7 +725,7 @@ export class LettaBot implements AgentSession {
}
}
case 'cancel': {
const convKey = channelId ? this.resolveConversationKey(channelId, chatId) : 'shared';
const convKey = channelId ? this.resolveConversationKey(channelId, chatId, forcePerChat) : 'shared';
// Check if there's actually an active run for this conversation key
if (!this.processingKeys.has(convKey) && !this.processing) {
@@ -889,7 +890,7 @@ export class LettaBot implements AgentSession {
// queuing it for normal processing. This prevents a deadlock where
// the stream is paused waiting for user input while the processing
// flag blocks new messages from being handled.
const incomingConvKey = this.resolveConversationKey(msg.channel, msg.chatId);
const incomingConvKey = this.resolveConversationKey(msg.channel, msg.chatId, msg.forcePerChat);
const pendingResolver = this.pendingQuestionResolvers.get(incomingConvKey);
if (pendingResolver) {
log.info(`Intercepted message as AskUserQuestion answer from ${msg.userId} (key=${incomingConvKey})`);
@@ -909,7 +910,7 @@ export class LettaBot implements AgentSession {
return;
}
const convKey = this.resolveConversationKey(msg.channel, msg.chatId);
const convKey = this.resolveConversationKey(msg.channel, msg.chatId, msg.forcePerChat);
if (convKey !== 'shared') {
// Per-channel, per-chat, or override mode: messages on different keys can run in parallel.
this.enqueueForKey(convKey, msg, adapter);
@@ -994,7 +995,7 @@ export class LettaBot implements AgentSession {
// Wait for the user's next message (intercepted by handleMessage).
// Key by convKey so each chat resolves independently in per-chat mode.
const questionConvKey = this.resolveConversationKey(msg.channel, msg.chatId);
const questionConvKey = this.resolveConversationKey(msg.channel, msg.chatId, msg.forcePerChat);
const answer = await new Promise<string>((resolve) => {
this.pendingQuestionResolvers.set(questionConvKey, resolve);
});
@@ -1169,7 +1170,7 @@ export class LettaBot implements AgentSession {
// Run session
let session: Session | null = null;
try {
const convKey = this.resolveConversationKey(msg.channel, msg.chatId);
const convKey = this.resolveConversationKey(msg.channel, msg.chatId, msg.forcePerChat);
const seq = ++this.sendSequence;
const userText = msg.text || '';
log.info(`processMessage seq=${seq} key=${convKey} retried=${retried} user=${msg.userId} textLen=${userText.length}`);
@@ -1605,7 +1606,7 @@ export class LettaBot implements AgentSession {
// Only retry if we never sent anything to the user. hasResponse tracks
// the current buffer, but finalizeMessage() clears it on type changes.
// sentAnyMessage is the authoritative "did we deliver output" flag.
const retryConvKey = this.resolveConversationKey(msg.channel, msg.chatId);
const retryConvKey = this.resolveConversationKey(msg.channel, msg.chatId, msg.forcePerChat);
const retryConvIdFromStore = (retryConvKey === 'shared'
? this.store.conversationId
: this.store.getConversationId(retryConvKey)) ?? undefined;
@@ -1836,7 +1837,7 @@ export class LettaBot implements AgentSession {
log.error('Failed to send error message to channel:', sendError);
}
} finally {
const finalConvKey = this.resolveConversationKey(msg.channel, msg.chatId);
const finalConvKey = this.resolveConversationKey(msg.channel, msg.chatId, msg.forcePerChat);
// When session reuse is disabled, invalidate after every message to
// eliminate any possibility of stream state bleed between sequential
// sends. Costs ~5s subprocess init overhead per message.

25
src/core/conversation-key.test.ts
View File

@@ -85,10 +85,33 @@ describe('resolveConversationKey', () => {
expect(resolveConversationKey('telegram', 'disabled', new Set(), '12345')).toBe('default');
});
it('returns "default" in disabled mode regardless of overrides', () => {
it('returns \"default\" in disabled mode regardless of overrides', () => {
const overrides = new Set(['telegram']);
expect(resolveConversationKey('telegram', 'disabled', overrides)).toBe('default');
});
// --- forcePerChat ---
it('forcePerChat overrides shared mode to per-chat', () => {
expect(resolveConversationKey('discord', 'shared', new Set(), '99999', true)).toBe('discord:99999');
});
it('forcePerChat overrides per-channel mode to per-chat', () => {
expect(resolveConversationKey('discord', 'per-channel', new Set(), '99999', true)).toBe('discord:99999');
});
it('forcePerChat without chatId falls back normally', () => {
expect(resolveConversationKey('discord', 'shared', new Set(), undefined, true)).toBe('shared');
});
it('forcePerChat still respects disabled mode', () => {
expect(resolveConversationKey('discord', 'disabled', new Set(), '99999', true)).toBe('default');
});
it('forcePerChat=false does not change behavior', () => {
expect(resolveConversationKey('discord', 'shared', new Set(), '99999', false)).toBe('shared');
expect(resolveConversationKey('discord', 'per-channel', new Set(), '99999', false)).toBe('discord');
});
});
// ---------------------------------------------------------------------------

1
src/core/group-batcher.ts
View File

@@ -93,6 +93,7 @@ export class GroupBatcher {
wasMentioned: messages.some((m) => m.wasMentioned),
// Preserve listening-mode intent only if every message in the batch is non-mentioned listen mode.
isListeningMode: messages.every((m) => m.isListeningMode === true) ? true : undefined,
forcePerChat: last.forcePerChat,
isBatch: true,
batchedMessages: messages,
formatterHints: last.formatterHints,

7
src/core/session-manager.ts
View File

@@ -373,10 +373,11 @@ export class SessionManager {
return this.ensureSessionForKey(key, bootstrapRetried);
}
// LRU eviction: in per-chat mode, limit concurrent sessions to avoid
// unbounded subprocess growth.
// LRU eviction: limit concurrent sessions to avoid unbounded subprocess
// growth. Applies in per-chat mode and when forcePerChat (e.g., Discord
// thread-only) creates per-thread keys in other modes.
const maxSessions = this.config.maxSessions ?? 10;
if (this.config.conversationMode === 'per-chat' && this.sessions.size >= maxSessions) {
if (this.sessions.size >= maxSessions) {
let oldestKey: string | null = null;
let oldestTime = Infinity;
for (const [k, ts] of this.sessionLastUsed) {

1
src/core/types.ts
View File

@@ -84,6 +84,7 @@ export interface InboundMessage {
isBatch?: boolean; // Is this a batched group message?
batchedMessages?: InboundMessage[]; // Original individual messages (for batch formatting)
isListeningMode?: boolean; // Listening mode: agent processes for memory but response is suppressed
forcePerChat?: boolean; // Force per-chat conversation routing (e.g., Discord thread-only mode)
formatterHints?: FormatterHints; // Channel capabilities for directive rendering
}