Skip to content

Commit 95b14cd

Browse files
intojhanuragintojhanurag
authored
fix(core): preserve tagged system messages when processors return systemMessages (#16950)
## Description Returned `systemMessages` from processors used to call `replaceAllSystemMessages()`, which cleared both untagged system messages and tagged system messages owned by other processors. That meant a later processor returning a system message array could strip tags such as `observational-memory`, breaking later refresh/clear behavior and causing stale or duplicated memory context across steps. This PR updates `ProcessorRunner` so returned `systemMessages` replace only untagged system messages. Tagged system messages are preserved for the processors that own them, and returned entries that correspond to previously tagged messages are not duplicated as untagged copies. - Preserves tagged system messages when regular input processors return `systemMessages` - Preserves tagged system messages when `processInputStep` processors return `systemMessages` - Adds regression coverage for OM-like tagged system messages with channel context and for runner-level returned `systemMessages` behavior ## Context #17168 already fixed the channel-specific symptom by switching `ChatChannelProcessor` from returning `{ systemMessages: [...] }` to calling `messageList.addSystem(msg, this.id)`. Its PR description explicitly defers the general fix to this PR: > "the bug is general to any processor that returns `systemMessages`… The runner-side fix (general, broader) is being pursued separately in #16950." The bug class is still live for any other processor that returns a `systemMessages` array. The clearest example today is `BrowserContextProcessor` (`packages/core/src/browser/processor.ts:85`): ```ts const systemMessages = [...args.systemMessages, { role: 'system' as const, content: lines.join(' ') }]; return { messages: args.messages, systemMessages }; ``` An agent using Observational Memory together with a browser tool would hit the same tag-wipe behavior today. This PR closes that gap once at the runner level so every current and future processor returning `systemMessages` is safe. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Code refactoring - [ ] Performance improvement - [ ] Test update ## Checklist - [ ] I have made corresponding changes to the documentation (if applicable) - [x] I have added tests that prove my fix is effective or that my feature works - [ ] I have addressed all Coderabbit comments on this PR <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## ELI5 When one processor updates the system messages (notes/instructions), it was accidentally deleting all system messages from other processors too—including special labeled ones that shouldn't be touched. This fix makes sure that only regular, unlabeled messages get replaced, while labeled messages (like "observational-memory" tags) stay safe and protected. --- ## Overview This PR fixes a bug in `ProcessorRunner` where processors returning `systemMessages` were inadvertently clearing **tagged** system messages owned by other processors. Previously, `replaceAllSystemMessages()` would strip all system messages regardless of ownership tags, breaking functionality that relies on those tags (like observational memory refresh/clear behavior). The fix preserves tagged system messages while allowing processors to replace only untagged system messages. ## Changes ### Core Logic Updates (`packages/core/src/processors/runner.ts`) - Added internal helper functions to manage system message merging: - `getTaggedSystemMessages`: Extracts system messages that have tags by comparing all system messages against untagged ones - `applyReturnedSystemMessages`: Intelligently applies processor-returned system messages, preserving tagged ones while replacing untagged ones - Updated `runInputProcessors`: Processors returning `systemMessages` now call `applyReturnedSystemMessages()` instead of `replaceAllSystemMessages()`, preserving tags - Updated `runProcessInputStep`: Applies the same preservation logic when step processors return `systemMessages` ### Type Documentation (`packages/core/src/processors/index.ts`) - Clarified JSDoc comments for `processInput` and `processInputStep` return types to explicitly document that returned `systemMessages` replace only untagged messages while preserving tagged system messages owned by other processors ### Test Coverage - **`process-input-step.test.ts`** (+86 lines): Added two regression tests in the `systemMessages modification` suite: - Verifies that appending untagged system messages preserves existing tagged messages (e.g., `observational-memory` tags) - Validates ordering semantics when untagged messages are prepended: tagged messages remain in original positions while untagged messages appear in expected sequence - **`runner.test.ts`** (+11/-6 lines): Updated existing Issue `#9969` regression test to assert that after a processor returns modified `systemMessages`, only untagged messages are reflected in the prompt, while tagged system messages remain unchanged and retain original content ### Release Documentation - Added `.changeset` entry documenting the patch-level fix for `@mastra/core` ## Impact This change prevents tagged system messages (used by observational memory and similar features) from being inadvertently dropped when subsequent processors return updated system messages, ensuring consistent and predictable behavior across processor chains. <!-- review_stack_entry_start --> [![Review Change Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/mastra-ai/mastra/pull/16950?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack) <!-- review_stack_entry_end --> <!-- end of auto-generated comment: release notes by coderabbit.ai -->
1 parent 2c79486 commit 95b14cd

18 files changed

Lines changed: 422 additions & 111 deletions

File tree

‎.changeset/fair-chicken-like.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'@mastra/core': patch
3+
---
4+
5+
Fixed processor-returned `systemMessages` wiping tagged system messages owned by other processors (e.g. observational memory). Processor `args.systemMessages` now exposes only the untagged system message bucket, so tagged messages owned by other processors are no longer round-tripped through the replacement API. `MessageList.replaceAllSystemMessages()` replaces only the untagged bucket and leaves tagged buckets intact. Final model input still receives both via `messageList.getAllSystemMessages()`.

‎.changeset/quiet-maps-fix.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'@mastra/ai-sdk': patch
3+
---
4+
5+
Fixed processor middleware so `args.systemMessages` only contains untagged system messages. Tagged processor-owned system messages stay on the message list and are still included in the final model input.

‎.changeset/soft-walls-fix.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
'@mastra/inngest': patch
3+
---
4+
5+
Fixed processor workflow steps so `args.systemMessages` only contains untagged system messages. Tagged processor-owned system messages stay on the message list and are still included in the final model input.

‎client-sdks/ai-sdk/src/middleware.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -498,7 +498,7 @@ export function createProcessorMiddleware(options: ProcessorMiddlewareOptions):
498498
// but the messageList reference takes precedence for preserving source info.
499499
await processor.processInput({
500500
messages: messageList.get.input.db(),
501-
systemMessages: messageList.getAllSystemMessages(),
501+
systemMessages: messageList.getSystemMessages(),
502502
messageList,
503503
requestContext,
504504
abort: (reason?: string): never => {

‎packages/core/src/agent/message-list/message-list.ts‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1231,20 +1231,16 @@ export class MessageList {
12311231
}
12321232

12331233
/**
1234-
* Replace all system messages with new ones
1235-
* This clears both tagged and untagged system messages and replaces them with the provided array
1236-
* @param messages - Array of system messages to set
1234+
* Replace the untagged system message bucket with the provided array while
1235+
* leaving tagged system message buckets (owned by other processors) intact.
1236+
* @param messages - Array of system messages to set as untagged
12371237
*/
12381238
public replaceAllSystemMessages(messages: CoreMessageV4[]): this {
1239-
// Clear existing system messages
12401239
this.systemMessages = [];
1241-
this.taggedSystemMessages = {};
12421240

1243-
// Add all new messages as untagged (processors don't need to preserve tags)
12441241
for (const message of messages) {
1245-
if (message.role === 'system') {
1246-
this.systemMessages.push(message);
1247-
}
1242+
if (message.role !== 'system') continue;
1243+
this.systemMessages.push(message);
12481244
}
12491245

12501246
return this;

‎packages/core/src/agent/message-list/tests/message-list.test.ts‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4281,10 +4281,10 @@ describe('MessageList', () => {
42814281
});
42824282

42834283
describe('replaceAllSystemMessages', () => {
4284-
it('should replace all system messages with new ones', () => {
4284+
it('should replace untagged system messages and preserve tagged ones', () => {
42854285
const list = new MessageList();
42864286
list.addSystem('Original instruction 1');
4287-
list.addSystem('Original instruction 2', 'memory');
4287+
list.addSystem('Memory context', 'memory');
42884288

42894289
const newSystemMessages: AIV4Type.CoreSystemMessage[] = [
42904290
{ role: 'system', content: 'New instruction 1' },
@@ -4293,22 +4293,21 @@ describe('MessageList', () => {
42934293

42944294
list.replaceAllSystemMessages(newSystemMessages);
42954295

4296-
const systemMessages = list.getAllSystemMessages();
4297-
expect(systemMessages).toHaveLength(2);
4298-
expect(systemMessages[0].content).toBe('New instruction 1');
4299-
expect(systemMessages[1].content).toBe('New instruction 2');
4296+
expect(list.getSystemMessages().map(m => m.content)).toEqual(['New instruction 1', 'New instruction 2']);
4297+
expect(list.getSystemMessages('memory').map(m => m.content)).toEqual(['Memory context']);
43004298
});
43014299

4302-
it('should clear all existing system messages including tagged ones', () => {
4300+
it('should preserve tagged system messages when called with an empty array', () => {
43034301
const list = new MessageList();
43044302
list.addSystem('Instruction');
43054303
list.addSystem('Memory context', 'memory');
43064304
list.addSystem('User provided', 'user-provided');
43074305

43084306
list.replaceAllSystemMessages([]);
43094307

4310-
const systemMessages = list.getAllSystemMessages();
4311-
expect(systemMessages).toHaveLength(0);
4308+
expect(list.getSystemMessages()).toHaveLength(0);
4309+
expect(list.getSystemMessages('memory').map(m => m.content)).toEqual(['Memory context']);
4310+
expect(list.getSystemMessages('user-provided').map(m => m.content)).toEqual(['User provided']);
43124311
});
43134312

43144313
it('should not affect non-system messages', () => {

‎packages/core/src/channels/processor.test.ts‎

Lines changed: 6 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -53,9 +53,9 @@ describe('ChatChannelProcessor', () => {
5353
const args = createArgs({
5454
channel: { platform: 'slack', isDM: true, userName: 'caleb' } as ChannelContext,
5555
messageList,
56-
// The runner builds args.systemMessages from getAllSystemMessages(), which
57-
// flattens tagged messages into a single untagged array.
58-
systemMessages: messageList.getAllSystemMessages(),
56+
// args.systemMessages exposes only the untagged bucket; tagged messages
57+
// owned by other processors remain accessible via messageList.
58+
systemMessages: messageList.getSystemMessages(),
5959
});
6060

6161
processor.processInputStep(args);
@@ -69,8 +69,7 @@ describe('ChatChannelProcessor', () => {
6969
expect(instructionsAfter).toHaveLength(1);
7070
expect(instructionsAfter[0]!.content).toBe('agent base instructions');
7171

72-
// And the channel system message must be present under its own tag — not
73-
// merged into the flat untagged array. This is the core of the fix.
72+
// And the channel system message must be present under its own tag.
7473
const channelAfter = args.messageList.getSystemMessages('chat-channel-context');
7574
expect(channelAfter).toHaveLength(1);
7675
expect(channelAfter[0]!.content).toContain('communicating via slack');
@@ -82,11 +81,11 @@ describe('ChatChannelProcessor', () => {
8281
const channel: ChannelContext = { platform: 'slack', isDM: true, userName: 'caleb' } as ChannelContext;
8382

8483
// Step 1
85-
const args1 = createArgs({ channel, messageList, systemMessages: messageList.getAllSystemMessages() });
84+
const args1 = createArgs({ channel, messageList, systemMessages: messageList.getSystemMessages() });
8685
processor.processInputStep(args1);
8786

8887
// Step 2 — same channel context, system messages list is re-derived
89-
const args2 = createArgs({ channel, messageList, systemMessages: messageList.getAllSystemMessages() });
88+
const args2 = createArgs({ channel, messageList, systemMessages: messageList.getSystemMessages() });
9089
processor.processInputStep(args2);
9190

9291
const channelMessages = messageList.getSystemMessages('chat-channel-context');

‎packages/core/src/loop/workflows/agentic-execution/llm-execution-step.ts‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -761,7 +761,7 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
761761
outputWriter,
762762
mastra,
763763
}: OuterLLMRun<TOOLS, OUTPUT> & { toolCallForeachOptions?: ToolCallForeachOptions }) {
764-
const initialSystemMessages = messageList.getAllSystemMessages();
764+
const initialUntaggedSystemMessages = messageList.getSystemMessages();
765765
const configuredToolCallConcurrency = resolveConfiguredToolCallConcurrency(toolCallConcurrency);
766766

767767
let currentIteration = 0;
@@ -823,11 +823,10 @@ export function createLLMExecutionStep<TOOLS extends ToolSet = ToolSet, OUTPUT =
823823
},
824824
});
825825
}
826-
// Reset system messages to original before each step execution
827-
// This ensures that system message modifications in prepareStep/processInputStep/processors
828-
// don't persist across steps - each step starts fresh with original system messages
829-
if (initialSystemMessages) {
830-
messageList.replaceAllSystemMessages(initialSystemMessages);
826+
// Reset the mutable untagged bucket before each step execution. Tagged
827+
// processor-owned buckets remain on messageList and are assembled later.
828+
if (initialUntaggedSystemMessages) {
829+
messageList.replaceAllSystemMessages(initialUntaggedSystemMessages);
831830
}
832831

833832
if (inputData.processorRetryFeedback) {

‎packages/core/src/processors/index.ts‎

Lines changed: 9 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,8 @@ export interface ProcessorMessageContext<TTripwireMetadata = unknown> extends Pr
9393
}
9494

9595
/**
96-
* Return type for processInput that includes modified system messages
96+
* Return type for processInput that includes modified untagged system messages.
97+
* Tagged system messages owned by other processors are preserved.
9798
*/
9899
export interface ProcessInputResultWithSystemMessages {
99100
messages: MastraDBMessage[];
@@ -116,7 +117,7 @@ export type ProcessInputResult = MessageList | MastraDBMessage[] | ProcessInputR
116117
* Arguments for processInput method
117118
*/
118119
export interface ProcessInputArgs<TTripwireMetadata = unknown> extends ProcessorMessageContext<TTripwireMetadata> {
119-
/** All system messages (agent instructions, user-provided, memory) for read/modify access */
120+
/** Untagged system messages for read/modify access. Tagged processor-owned messages remain on messageList. */
120121
systemMessages: CoreMessageV4[];
121122
/** Per-processor state that persists across all method calls within this request */
122123
state: Record<string, unknown>;
@@ -165,7 +166,7 @@ export interface ProcessInputStepArgs<TTripwireMetadata = unknown> extends Proce
165166
/** Mark the current assistant response message ID as complete and rotate to a fresh one, when supported by the caller */
166167
rotateResponseMessageId?: () => string;
167168

168-
/** All system messages (agent instructions, user-provided, memory) for read/modify access */
169+
/** Untagged system messages for read/modify access. Tagged processor-owned messages remain on messageList. */
169170
systemMessages: CoreMessageV4[];
170171
/** Per-processor state that persists across all method calls within this request */
171172
state: Record<string, unknown>;
@@ -220,7 +221,10 @@ export type ProcessInputStepResult = {
220221

221222
messages?: MastraDBMessage[];
222223
messageList?: MessageList;
223-
/** Replace all system messages with these */
224+
/**
225+
* Replace untagged system messages with these while preserving tagged system messages
226+
* owned by other processors.
227+
*/
224228
systemMessages?: CoreMessageV4[];
225229
providerOptions?: SharedProviderOptions;
226230
modelSettings?: Omit<CallSettings, 'abortSignal'>;
@@ -405,7 +409,7 @@ export interface ProcessOutputStepArgs<TTripwireMetadata = unknown> extends Proc
405409
text?: string;
406410
/** Token usage for the current step (input tokens, output tokens, etc.) */
407411
usage: LanguageModelUsage;
408-
/** All system messages */
412+
/** Untagged system messages. Tagged processor-owned messages remain on messageList. */
409413
systemMessages: CoreMessageV4[];
410414
/** All completed steps so far (including the current step) */
411415
steps: Array<StepResult<any>>;

‎packages/core/src/processors/process-input-step.test.ts‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1741,6 +1741,128 @@ describe('processInputStep', () => {
17411741
const systemMessages = messageList.getAllSystemMessages();
17421742
expect(systemMessages.length).toBe(3);
17431743
});
1744+
1745+
it('should preserve tagged system messages when processor returns systemMessages', async () => {
1746+
const processor: Processor = {
1747+
id: 'system-modifier',
1748+
processInputStep: async ({ systemMessages }) => {
1749+
return {
1750+
systemMessages: [...systemMessages, { role: 'system' as const, content: 'Additional instruction' }],
1751+
};
1752+
},
1753+
};
1754+
1755+
const runner = new ProcessorRunner({
1756+
inputProcessors: [processor],
1757+
outputProcessors: [],
1758+
logger: mockLogger,
1759+
agentName: 'test-agent',
1760+
});
1761+
1762+
const messageList = new MessageList({ threadId: 'test-thread' });
1763+
messageList.add([createMessage('User message')], 'input');
1764+
messageList.addSystem('Original instruction');
1765+
messageList.addSystem('Memory context', 'observational-memory');
1766+
1767+
await runner.runProcessInputStep({
1768+
messageList,
1769+
stepNumber: 0,
1770+
model: createMockModel(),
1771+
steps: [],
1772+
});
1773+
1774+
expect(messageList.getSystemMessages('observational-memory').map(message => message.content)).toEqual([
1775+
'Memory context',
1776+
]);
1777+
expect(messageList.getSystemMessages().map(message => message.content)).toEqual([
1778+
'Original instruction',
1779+
'Additional instruction',
1780+
]);
1781+
expect(messageList.getAllSystemMessages().map(message => message.content)).toEqual([
1782+
'Original instruction',
1783+
'Additional instruction',
1784+
'Memory context',
1785+
]);
1786+
});
1787+
1788+
it('should allow processor-returned systemMessages to prepend untagged system context', async () => {
1789+
const processor: Processor = {
1790+
id: 'system-prepender',
1791+
processInputStep: async ({ systemMessages }) => {
1792+
return {
1793+
systemMessages: [{ role: 'system' as const, content: 'Priority instruction' }, ...systemMessages],
1794+
};
1795+
},
1796+
};
1797+
1798+
const runner = new ProcessorRunner({
1799+
inputProcessors: [processor],
1800+
outputProcessors: [],
1801+
logger: mockLogger,
1802+
agentName: 'test-agent',
1803+
});
1804+
1805+
const messageList = new MessageList({ threadId: 'test-thread' });
1806+
messageList.add([createMessage('User message')], 'input');
1807+
messageList.addSystem('Original instruction');
1808+
messageList.addSystem('Memory context', 'observational-memory');
1809+
1810+
await runner.runProcessInputStep({
1811+
messageList,
1812+
stepNumber: 0,
1813+
model: createMockModel(),
1814+
steps: [],
1815+
});
1816+
1817+
expect(messageList.getSystemMessages('observational-memory').map(message => message.content)).toEqual([
1818+
'Memory context',
1819+
]);
1820+
expect(messageList.getSystemMessages().map(message => message.content)).toEqual([
1821+
'Priority instruction',
1822+
'Original instruction',
1823+
]);
1824+
expect(messageList.getAllSystemMessages().map(message => message.content)).toEqual([
1825+
'Priority instruction',
1826+
'Original instruction',
1827+
'Memory context',
1828+
]);
1829+
});
1830+
1831+
it('should not include tagged system messages in processor args.systemMessages', async () => {
1832+
let seenSystemMessages: any[] = [];
1833+
const processor: Processor = {
1834+
id: 'system-inspector',
1835+
processInputStep: async ({ systemMessages, messageList }) => {
1836+
seenSystemMessages = systemMessages;
1837+
return { messageList };
1838+
},
1839+
};
1840+
1841+
const runner = new ProcessorRunner({
1842+
inputProcessors: [processor],
1843+
outputProcessors: [],
1844+
logger: mockLogger,
1845+
agentName: 'test-agent',
1846+
});
1847+
1848+
const messageList = new MessageList({ threadId: 'test-thread' });
1849+
messageList.add([createMessage('User message')], 'input');
1850+
messageList.addSystem('Original instruction');
1851+
messageList.addSystem('Memory context', 'observational-memory');
1852+
1853+
await runner.runProcessInputStep({
1854+
messageList,
1855+
stepNumber: 0,
1856+
model: createMockModel(),
1857+
steps: [],
1858+
});
1859+
1860+
expect(seenSystemMessages.map(m => m.content)).toEqual(['Original instruction']);
1861+
expect(messageList.getAllSystemMessages().map(m => m.content)).toEqual([
1862+
'Original instruction',
1863+
'Memory context',
1864+
]);
1865+
});
17441866
});
17451867

17461868
describe('abort functionality', () => {

0 commit comments

Comments
 (0)