Conversation
📝 WalkthroughWalkthroughThis pull request introduces an optional Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
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 |
commit: |
![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 (2)
src/component/messages.ts (1)
377-429:⚠️ Potential issue | 🟠 Major
finalizeMessagedoesn't propagateonSaveMessages, silently skipping the callback during streaming finalization.When
finalizeMessageis called and the streamed content has accumulated (lines 397–414), it callsaddMessagesHandlerinternally withoutonSaveMessages. This path runs duringstreamTextcompletion when the pending message hasn't been filled in yet, meaning messages can be written to the thread through this path without triggering the configured callback. This contradicts the documented guarantee that the callback fires forstreamText.🐛 Proposed fix
export const finalizeMessage = mutation({ args: { messageId: v.id("messages"), + onSaveMessages: v.optional(v.string()), result: v.union( v.object({ status: v.literal("success") }), v.object({ status: v.literal("failed"), error: v.string() }), ), }, ... handler: async (ctx, { messageId, result }) => { ... if (!message.message?.content.length) { const messages = await getStreamingMessagesWithMetadata(ctx, message, result); if (messages.length > 0) { await addMessagesHandler(ctx, { messages, threadId: message.threadId, agentName: message.agentName, failPendingSteps: false, pendingMessageId: messageId, userId: message.userId, embeddings: undefined, + onSaveMessages: args.onSaveMessages, }); return; } }The
onSaveMessageshandle would need to be threaded from the client callers (component.messages.finalizeMessage) instartGeneration'sfailclosure and any other call sites.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/component/messages.ts` around lines 377 - 429, finalizeMessage currently calls addMessagesHandler when streaming produced messages (in the block that checks message.message?.content length) but does not pass the onSaveMessages callback, so saved messages via this streaming-finalization path skip the configured onSaveMessages hook; update finalizeMessage to accept and forward an onSaveMessages parameter to addMessagesHandler (threading the onSaveMessages argument through the finalizeMessage mutation signature and into the call to addMessagesHandler), and update all callers (e.g., the startGeneration failure closure and any other places invoking component.messages.finalizeMessage) to pass their onSaveMessages handler through so the callback is invoked for streamed completions as well.src/client/index.ts (1)
1008-1056:⚠️ Potential issue | 🟡 Minor
saveStepandsaveObjectdon't propagateonSaveMessages.Both methods call
ctx.runMutation(this.component.messages.addMessages, ...)directly without forwardingthis.options.onSaveMessages. If users callagent.saveStep(...)oragent.saveObject(...), the configured callback will silently not fire. The other explicit save methods (saveMessage,saveMessages,asSaveMessagesMutation) all correctly propagate it via thesaveMessageshelper path.If these methods are intentionally excluded from the callback contract, a brief comment to that effect would prevent confusion.
Also applies to: 1065-1107
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client/index.ts` around lines 1008 - 1056, The saveStep (and likewise saveObject) method currently calls ctx.runMutation(this.component.messages.addMessages, ...) directly and thus does not forward the configured onSaveMessages callback; update saveStep and saveObject to propagate this.options.onSaveMessages into the mutation invocation (or refactor to call the existing saveMessages helper/asSaveMessagesMutation path that already forwards onSaveMessages) so the user-provided callback fires when these APIs are used; reference the saveStep and saveObject methods and ensure the mutation payload includes the onSaveMessages handler from this.options or that the helper path is reused; if omission was intentional, add a short clarifying comment in those methods noting they do not invoke onSaveMessages.
🧹 Nitpick comments (2)
src/client/messages.ts (1)
221-252: StandalonesaveMessagesilently drops theonSaveMessagescallback.When
saveMessagecallssaveMessages(line 238), it doesn't forward anonSaveMessageshandler.SaveMessageArgsdoesn't include the field either. This means the callback is only triggered via theAgentclass path (Agent.saveMessage → this.saveMessages). Callers using the standalonesaveMessageexport directly cannot attach the callback, even though the type docs promise coverage forsaveMessage.💡 Proposed fix — add `onSaveMessages` to `SaveMessageArgs` and forward it
export type SaveMessageArgs = { threadId: string; userId?: string | null; promptMessageId?: string; metadata?: Omit<MessageWithMetadata, "message">; embedding?: { vector: number[]; model: string }; pendingMessageId?: string; + /** + * Optional callback mutation to invoke after the message is saved. + * Called within the same transaction as the message save. + */ + onSaveMessages?: SaveMessagesHandler; } & ( ... );And in
saveMessage:const { messages } = await saveMessages(ctx, component, { ... embeddings, + onSaveMessages: args.onSaveMessages, });🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client/messages.ts` around lines 221 - 252, The standalone saveMessage function currently drops the onSaveMessages callback when delegating to saveMessages; update the SaveMessageArgs type to include an optional onSaveMessages callback and pass args.onSaveMessages through in the call to saveMessages (i.e., include onSaveMessages: args.onSaveMessages in the options object passed to saveMessages) so callers using the exported saveMessage receive the same callback behavior as Agent.saveMessage → this.saveMessages.src/client/start.ts (1)
96-107:onSaveMessagesis declared redundantly in thestartGenerationoptions type.
onSaveMessages?: SaveMessagesHandleris already part of bothOptions(via the new field at line 678 oftypes.ts) andConfig(line 137 oftypes.ts), which are intersected into this parameter. The additional explicit declaration at line 106 is redundant.♻️ Proposed cleanup
{ threadId, ...opts }: Options & Config & { userId?: string | null; threadId?: string; languageModel?: LanguageModel; agentName: string; agentForToolCtx?: Agent; - onSaveMessages?: SaveMessagesHandler; },🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/client/start.ts` around lines 96 - 107, The parameter type for the start generation function currently redundantly re-declares onSaveMessages in the destructured options object; remove the explicit onSaveMessages?: SaveMessagesHandler from the parameter intersection so the function relies on the existing onSaveMessages field defined in Options and Config, keeping the rest of the destructured props (threadId, userId, languageModel, agentName, agentForToolCtx, etc.) unchanged and ensuring the signature uses the existing Options & Config intersection only.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/client/types.ts`:
- Around line 338-391: SaveMessagesCallbackArgs currently defines only threadId
and messages but the code invokes the handler with an extra userId field,
causing runtime validation failures; update SaveMessagesCallbackArgs to include
userId as an optional string (userId?: string) so the declared argument shape
matches what components pass, and ensure SaveMessagesHandler
(FunctionReference<...>) uses the updated SaveMessagesCallbackArgs type so
mutation validators and TypeScript stay in sync with the invocation that passes
{ userId, threadId, messages }.
---
Outside diff comments:
In `@src/client/index.ts`:
- Around line 1008-1056: The saveStep (and likewise saveObject) method currently
calls ctx.runMutation(this.component.messages.addMessages, ...) directly and
thus does not forward the configured onSaveMessages callback; update saveStep
and saveObject to propagate this.options.onSaveMessages into the mutation
invocation (or refactor to call the existing saveMessages
helper/asSaveMessagesMutation path that already forwards onSaveMessages) so the
user-provided callback fires when these APIs are used; reference the saveStep
and saveObject methods and ensure the mutation payload includes the
onSaveMessages handler from this.options or that the helper path is reused; if
omission was intentional, add a short clarifying comment in those methods noting
they do not invoke onSaveMessages.
In `@src/component/messages.ts`:
- Around line 377-429: finalizeMessage currently calls addMessagesHandler when
streaming produced messages (in the block that checks message.message?.content
length) but does not pass the onSaveMessages callback, so saved messages via
this streaming-finalization path skip the configured onSaveMessages hook; update
finalizeMessage to accept and forward an onSaveMessages parameter to
addMessagesHandler (threading the onSaveMessages argument through the
finalizeMessage mutation signature and into the call to addMessagesHandler), and
update all callers (e.g., the startGeneration failure closure and any other
places invoking component.messages.finalizeMessage) to pass their onSaveMessages
handler through so the callback is invoked for streamed completions as well.
---
Duplicate comments:
In `@src/component/messages.ts`:
- Around line 310-325: The onSaveMessages invocation passes userId via
ctx.runMutation (see onSaveMessages and the call in the messages save path), but
the exported type SaveMessagesCallbackArgs lacks userId; update the
SaveMessagesCallbackArgs type (in src/client/types.ts) to include userId?:
string so runtime validation accepts the extra field and the TypeScript type
matches the JSDoc/example; ensure the optional modifier matches the example
(userId optional) so existing callbacks remain compatible.
---
Nitpick comments:
In `@src/client/messages.ts`:
- Around line 221-252: The standalone saveMessage function currently drops the
onSaveMessages callback when delegating to saveMessages; update the
SaveMessageArgs type to include an optional onSaveMessages callback and pass
args.onSaveMessages through in the call to saveMessages (i.e., include
onSaveMessages: args.onSaveMessages in the options object passed to
saveMessages) so callers using the exported saveMessage receive the same
callback behavior as Agent.saveMessage → this.saveMessages.
In `@src/client/start.ts`:
- Around line 96-107: The parameter type for the start generation function
currently redundantly re-declares onSaveMessages in the destructured options
object; remove the explicit onSaveMessages?: SaveMessagesHandler from the
parameter intersection so the function relies on the existing onSaveMessages
field defined in Options and Config, keeping the rest of the destructured props
(threadId, userId, languageModel, agentName, agentForToolCtx, etc.) unchanged
and ensuring the signature uses the existing Options & Config intersection only.
| export type SaveMessagesCallbackArgs = { | ||
| /** | ||
| * The thread the messages were saved to. | ||
| */ | ||
| threadId: string; | ||
| /** | ||
| * The messages that were saved. | ||
| */ | ||
| messages: MessageDoc[]; | ||
| }; | ||
|
|
||
| /** | ||
| * A reference to a mutation function that will be called whenever messages are | ||
| * saved to a thread. This callback is invoked **within the same transaction** | ||
| * as the message save, making it transactional. | ||
| * | ||
| * This includes messages saved via generateText, streamText, generateObject, | ||
| * streamObject, saveMessage, and saveMessages. | ||
| * | ||
| * Use this to trigger side effects when messages are saved, such as updating | ||
| * counters, creating notifications, or syncing with external systems. | ||
| * | ||
| * @example | ||
| * ```ts | ||
| * // In your convex/myModule.ts: | ||
| * export const onNewMessages = internalMutation({ | ||
| * args: { | ||
| * userId: v.optional(v.string()), | ||
| * threadId: v.string(), | ||
| * messages: v.array(vMessageDoc), | ||
| * }, | ||
| * handler: async (ctx, args) => { | ||
| * // This runs in the same transaction as the message save | ||
| * await ctx.db.insert("messageEvents", { | ||
| * threadId: args.threadId, | ||
| * messageCount: args.messages.length, | ||
| * timestamp: Date.now(), | ||
| * }); | ||
| * }, | ||
| * }); | ||
| * | ||
| * // In your agent configuration: | ||
| * const agent = new Agent(components.agent, { | ||
| * name: "myAgent", | ||
| * languageModel: openai.chat("gpt-4o-mini"), | ||
| * onSaveMessages: internal.myModule.onNewMessages, | ||
| * }); | ||
| * ``` | ||
| */ | ||
| export type SaveMessagesHandler = FunctionReference< | ||
| "mutation", | ||
| "internal" | "public", | ||
| SaveMessagesCallbackArgs | ||
| >; |
There was a problem hiding this comment.
SaveMessagesCallbackArgs is missing userId, causing a type-runtime mismatch.
The SaveMessagesCallbackArgs type only declares threadId and messages, but src/component/messages.ts invokes the callback with { userId, threadId, messages: savedMessages } (lines 318–322). Convex strictly validates mutation arguments against their declared validator, so any callback mutation that doesn't declare userId will fail at runtime when the component passes it.
The embedded JSDoc example already shows the correct shape (userId: v.optional(v.string())), but users relying on the exported type to type-check their mutation will get a TypeScript error if they add userId to their validator since it isn't in the type.
🐛 Proposed fix
export type SaveMessagesCallbackArgs = {
/**
* The thread the messages were saved to.
*/
threadId: string;
/**
* The messages that were saved.
*/
messages: MessageDoc[];
+ /**
+ * The user associated with the thread, if any.
+ */
+ userId?: string;
};📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| export type SaveMessagesCallbackArgs = { | |
| /** | |
| * The thread the messages were saved to. | |
| */ | |
| threadId: string; | |
| /** | |
| * The messages that were saved. | |
| */ | |
| messages: MessageDoc[]; | |
| }; | |
| /** | |
| * A reference to a mutation function that will be called whenever messages are | |
| * saved to a thread. This callback is invoked **within the same transaction** | |
| * as the message save, making it transactional. | |
| * | |
| * This includes messages saved via generateText, streamText, generateObject, | |
| * streamObject, saveMessage, and saveMessages. | |
| * | |
| * Use this to trigger side effects when messages are saved, such as updating | |
| * counters, creating notifications, or syncing with external systems. | |
| * | |
| * @example | |
| * ```ts | |
| * // In your convex/myModule.ts: | |
| * export const onNewMessages = internalMutation({ | |
| * args: { | |
| * userId: v.optional(v.string()), | |
| * threadId: v.string(), | |
| * messages: v.array(vMessageDoc), | |
| * }, | |
| * handler: async (ctx, args) => { | |
| * // This runs in the same transaction as the message save | |
| * await ctx.db.insert("messageEvents", { | |
| * threadId: args.threadId, | |
| * messageCount: args.messages.length, | |
| * timestamp: Date.now(), | |
| * }); | |
| * }, | |
| * }); | |
| * | |
| * // In your agent configuration: | |
| * const agent = new Agent(components.agent, { | |
| * name: "myAgent", | |
| * languageModel: openai.chat("gpt-4o-mini"), | |
| * onSaveMessages: internal.myModule.onNewMessages, | |
| * }); | |
| * ``` | |
| */ | |
| export type SaveMessagesHandler = FunctionReference< | |
| "mutation", | |
| "internal" | "public", | |
| SaveMessagesCallbackArgs | |
| >; | |
| export type SaveMessagesCallbackArgs = { | |
| /** | |
| * The thread the messages were saved to. | |
| */ | |
| threadId: string; | |
| /** | |
| * The messages that were saved. | |
| */ | |
| messages: MessageDoc[]; | |
| /** | |
| * The user associated with the thread, if any. | |
| */ | |
| userId?: string; | |
| }; | |
| /** | |
| * A reference to a mutation function that will be called whenever messages are | |
| * saved to a thread. This callback is invoked **within the same transaction** | |
| * as the message save, making it transactional. | |
| * | |
| * This includes messages saved via generateText, streamText, generateObject, | |
| * streamObject, saveMessage, and saveMessages. | |
| * | |
| * Use this to trigger side effects when messages are saved, such as updating | |
| * counters, creating notifications, or syncing with external systems. | |
| * | |
| * `@example` | |
| * |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/client/types.ts` around lines 338 - 391, SaveMessagesCallbackArgs
currently defines only threadId and messages but the code invokes the handler
with an extra userId field, causing runtime validation failures; update
SaveMessagesCallbackArgs to include userId as an optional string (userId?:
string) so the declared argument shape matches what components pass, and ensure
SaveMessagesHandler (FunctionReference<...>) uses the updated
SaveMessagesCallbackArgs type so mutation validators and TypeScript stay in sync
with the invocation that passes { userId, threadId, messages }.
1 participant
This PR adds a new
onSaveMessagescallback feature that allows developers to execute custom logic whenever messages are saved to a thread. The callback is invoked within the same transaction as the message save, making it transactional.Key changes:
SaveMessagesHandlerandSaveMessagesCallbackArgstypesonSaveMessagescallbackExample usage:
The callback receives the thread ID and the saved messages, allowing for side effects like updating counters, creating notifications, or syncing with external systems.
Summary by CodeRabbit
New Features
Chores