diff --git a/docs/users/features/channels/overview.md b/docs/users/features/channels/overview.md index 6026ec0923..05301fc1c4 100644 --- a/docs/users/features/channels/overview.md +++ b/docs/users/features/channels/overview.md @@ -89,20 +89,41 @@ Controls how conversation sessions are managed: ### Channel Memory -Channel memory lets accepted channel senders save stable context for one chat or thread. Qwen Code injects that memory when a fresh channel session starts, including after `/clear`. +Channel memory stores durable context for one chat or thread. Entries have stable +IDs, so a list response can be used for deterministic follow-up operations. -Natural-language examples: +- `记住:默认使用 staging 环境` saves a new entry for the current chat or + thread. +- `查看记忆` lists entries and their stable IDs. Use `查看第 2 页记忆` to view + a later page, or `查看记忆 ` to view one entry. +- `把 改成默认使用 production` updates that entry immediately, and + `忘掉 ` removes it immediately. Neither operation needs confirmation. +- `清空记忆` starts the clear-all confirmation flow; `确认清空记忆` completes + it. -- `记住:默认使用 staging 环境` saves memory for the current chat or thread. -- `你记一下以后回复前要说 1122` saves the extracted durable memory. -- `你现在都记住了什么` shows saved memory for the current chat or thread. -- `把这个聊天的记忆清空` starts the clear flow; `确认清空记忆` confirms it. +Update and removal requests must include an entry ID. Natural-language +references without an ID, such as "忘掉刚才那条", are deferred to a later phase +and are not treated as deterministic channel-memory operations. -Channel memory follows the channel access gates. Any message accepted by `senderPolicy`, `dmPolicy`, `groupPolicy`, group settings, pairing, and mention requirements can read, write, or clear memory for that chat or thread. +The legacy slash aliases `/remember-channel`, `/channel-memory`, and +`/forget-channel` have been removed. They are no longer channel-memory +commands. -In open groups, any accepted member can update shared channel memory for that group. Use `allowlist` or `pairing` policies when memory should be limited to trusted senders. +Channel memory follows the channel access gates. Any message accepted by +`senderPolicy`, `dmPolicy`, `groupPolicy`, group settings, pairing, and mention +requirements can read, write, update, or clear memory for that chat or thread. +Accepted members of the same group share that group's target store. Use +`allowlist` or `pairing` policies when group memory should be limited to trusted +senders. -Memory is keyed to the current chat or thread, so it is not injected into `single` session scope, where every chat shares one channel-wide agent session. +Existing legacy `CHANNEL.md` memory is migrated automatically to structured +`CHANNEL.json` storage on the first mutation. Structured memory persists across +standalone channel and daemon-managed channel restarts, and is injected when a +fresh target-scoped session starts, including after `/clear`. + +Memory remains keyed to the current chat or thread. It is not injected into a +`sessionScope: single` session, because that session is shared across the whole +channel rather than scoped to one target. ### Token Security diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index 7fdbcb2384..126dbf3f8a 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -5,6 +5,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import type { ChannelConfig, + ChannelMemoryEntry, ChannelTaskLifecycleEvent, Envelope, SessionTarget, @@ -277,6 +278,35 @@ function channelMemoryPrompt(memoryText: string): string { ].join('\n'); } +function createChannelMemory(entries: ChannelMemoryEntry[] = []) { + return { + readChannelMemory: vi.fn().mockResolvedValue(''), + listChannelMemoryEntries: vi.fn().mockResolvedValue(entries), + addChannelMemoryEntries: vi + .fn() + .mockImplementation( + async ( + _target: unknown, + texts: readonly string[], + createdBy?: string, + ) => ({ + changed: true, + added: texts.map((text, index) => ({ + id: `m-${String(index + 1).padStart(12, '0')}`, + text, + createdBy, + })), + duplicateIds: [], + }), + ), + updateChannelMemoryEntry: vi.fn().mockResolvedValue({ changed: true }), + removeChannelMemoryEntries: vi + .fn() + .mockResolvedValue({ changed: true, removed: [] }), + clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), + }; +} + describe('ChannelBase', () => { let bridge: ChannelAgentBridge; @@ -1851,11 +1881,7 @@ describe('ChannelBase', () => { }); it('natural remember saves group memory for accepted group messages', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel( { allowedUsers: ['alice'], groupPolicy: 'open' }, { channelMemory }, @@ -1871,26 +1897,23 @@ describe('ChannelBase', () => { }), ); - expect(channelMemory.appendChannelMemory).toHaveBeenCalledWith( + expect(channelMemory.addChannelMemoryEntries).toHaveBeenCalledWith( { channelName: 'test-chan', chatId: 'group-1', threadId: undefined, }, - '发布前跑 npm run build', + ['发布前跑 npm run build'], + 'alice', ); expect(ch.sent).toEqual([ - { chatId: 'group-1', text: 'Channel memory updated.' }, + { chatId: 'group-1', text: 'Channel memory m-000000000001 saved.' }, ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('llm memory classifier can save natural remember requests', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi.fn().mockResolvedValue({ intent: 'remember', @@ -1913,26 +1936,23 @@ describe('ChannelBase', () => { expect( memoryIntentClassifier.classifyChannelMemoryIntent, ).toHaveBeenCalledWith('你记一下以后回复前要说 1122'); - expect(channelMemory.appendChannelMemory).toHaveBeenCalledWith( + expect(channelMemory.addChannelMemoryEntries).toHaveBeenCalledWith( { channelName: 'test-chan', chatId: 'chat1', threadId: undefined, }, - '回复前必须说 1122', + ['回复前必须说 1122'], + 'alice', ); expect(ch.sent).toEqual([ - { chatId: 'chat1', text: 'Channel memory updated.' }, + { chatId: 'chat1', text: 'Channel memory m-000000000001 saved.' }, ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('regex memory intent skips the llm classifier', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi.fn().mockResolvedValue({ intent: 'none', @@ -1954,16 +1974,17 @@ describe('ChannelBase', () => { expect( memoryIntentClassifier.classifyChannelMemoryIntent, ).not.toHaveBeenCalled(); - expect(channelMemory.appendChannelMemory).toHaveBeenCalledWith( + expect(channelMemory.addChannelMemoryEntries).toHaveBeenCalledWith( { channelName: 'test-chan', chatId: 'chat1', threadId: undefined, }, - '回复前必须说 1122', + ['回复前必须说 1122'], + 'alice', ); expect(ch.sent).toEqual([ - { chatId: 'chat1', text: 'Channel memory updated.' }, + { chatId: 'chat1', text: 'Channel memory m-000000000001 saved.' }, ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); @@ -1995,11 +2016,9 @@ describe('ChannelBase', () => { }); it('llm memory classifier can list memory for natural questions', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory([ + { id: 'm-a31f0d82c7e4', text: 'Use staging.' }, + ]); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi.fn().mockResolvedValue({ intent: 'list', @@ -2021,21 +2040,22 @@ describe('ChannelBase', () => { }), ); - expect(channelMemory.readChannelMemory).toHaveBeenCalledWith({ + expect(channelMemory.listChannelMemoryEntries).toHaveBeenCalledWith({ channelName: 'test-chan', chatId: 'group-1', threadId: undefined, }); - expect(ch.sent).toEqual([{ chatId: 'group-1', text: 'Use staging.' }]); + expect(ch.sent).toEqual([ + { + chatId: 'group-1', + text: 'Channel memory (page 1/1):\nm-a31f0d82c7e4 Use staging.', + }, + ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('llm memory classifier can start clear flow for natural requests', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi.fn().mockResolvedValue({ intent: 'clear_all', @@ -2068,11 +2088,9 @@ describe('ChannelBase', () => { }); it('llm memory classifier gate ignores platform format characters', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory([ + { id: 'm-a31f0d82c7e4', text: 'Use staging.' }, + ]); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi.fn().mockResolvedValue({ intent: 'list', @@ -2094,16 +2112,17 @@ describe('ChannelBase', () => { expect( memoryIntentClassifier.classifyChannelMemoryIntent, ).toHaveBeenCalledWith('看看你记\u200b忆里有什么'); - expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'Use staging.' }]); + expect(ch.sent).toEqual([ + { + chatId: 'chat1', + text: 'Channel memory (page 1/1):\nm-a31f0d82c7e4 Use staging.', + }, + ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('llm memory classifier low confidence falls through to agent', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi.fn().mockResolvedValue({ intent: 'list', @@ -2122,17 +2141,13 @@ describe('ChannelBase', () => { }), ); - expect(channelMemory.appendChannelMemory).not.toHaveBeenCalled(); + expect(channelMemory.addChannelMemoryEntries).not.toHaveBeenCalled(); expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'agent response' }]); expect(bridge.prompt).toHaveBeenCalled(); }); it('llm memory classifier none intent falls through to agent', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi.fn().mockResolvedValue({ intent: 'none', @@ -2151,18 +2166,14 @@ describe('ChannelBase', () => { }), ); - expect(channelMemory.appendChannelMemory).not.toHaveBeenCalled(); + expect(channelMemory.addChannelMemoryEntries).not.toHaveBeenCalled(); expect(channelMemory.clearChannelMemory).not.toHaveBeenCalled(); expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'agent response' }]); expect(bridge.prompt).toHaveBeenCalled(); }); it('llm memory classifier errors fall through to agent', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const memoryIntentClassifier = { classifyChannelMemoryIntent: vi .fn() @@ -2183,7 +2194,7 @@ describe('ChannelBase', () => { }), ); - expect(channelMemory.appendChannelMemory).not.toHaveBeenCalled(); + expect(channelMemory.addChannelMemoryEntries).not.toHaveBeenCalled(); expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'agent response' }]); expect(bridge.prompt).toHaveBeenCalled(); expect(stderrSpy).toHaveBeenCalledWith( @@ -2193,13 +2204,10 @@ describe('ChannelBase', () => { }); it('natural remember reports append failures', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi - .fn() - .mockRejectedValue(new Error('Channel memory exceeds maximum size')), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); + channelMemory.addChannelMemoryEntries.mockRejectedValue( + new Error('Channel memory exceeds maximum size'), + ); const ch = createChannel( { allowedUsers: ['alice'], groupPolicy: 'open' }, { channelMemory }, @@ -2226,11 +2234,9 @@ describe('ChannelBase', () => { }); it('natural memory management follows open sender policy without allowedUsers', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory([ + { id: 'm-a31f0d82c7e4', text: 'Use staging.' }, + ]); const ch = createChannel( { senderPolicy: 'open', allowedUsers: [] }, { channelMemory }, @@ -2241,29 +2247,29 @@ describe('ChannelBase', () => { ); await ch.handleInbound(envelope({ text: '查看记忆', senderId: 'alice' })); - expect(channelMemory.appendChannelMemory).toHaveBeenCalledWith( + expect(channelMemory.addChannelMemoryEntries).toHaveBeenCalledWith( { channelName: 'test-chan', chatId: 'chat1', threadId: undefined, }, - 'Use staging.', + ['Use staging.'], + 'alice', ); expect(ch.sent).toEqual([ - { chatId: 'chat1', text: 'Channel memory updated.' }, - { chatId: 'chat1', text: 'Use staging.' }, + { chatId: 'chat1', text: 'Channel memory m-000000000001 saved.' }, + { + chatId: 'chat1', + text: 'Channel memory (page 1/1):\nm-a31f0d82c7e4 Use staging.', + }, ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('natural memory list shows trimmed memory for an allowed user', async () => { - const channelMemory = { - readChannelMemory: vi - .fn() - .mockResolvedValue('Use staging by default.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory([ + { id: 'm-a31f0d82c7e4', text: 'Use staging by default.\n' }, + ]); const ch = createChannel( { allowedUsers: ['alice'], groupPolicy: 'open' }, { channelMemory }, @@ -2272,17 +2278,16 @@ describe('ChannelBase', () => { await ch.handleInbound(envelope({ text: '查看记忆', senderId: 'alice' })); expect(ch.sent).toEqual([ - { chatId: 'chat1', text: 'Use staging by default.' }, + { + chatId: 'chat1', + text: 'Channel memory (page 1/1):\nm-a31f0d82c7e4 Use staging by default.', + }, ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('natural group remember follows open group access control', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel( { senderPolicy: 'open', @@ -2302,26 +2307,23 @@ describe('ChannelBase', () => { }), ); - expect(channelMemory.appendChannelMemory).toHaveBeenCalledWith( + expect(channelMemory.addChannelMemoryEntries).toHaveBeenCalledWith( { channelName: 'test-chan', chatId: 'group-1', threadId: undefined, }, - '这个群默认讨论 qwen-code', + ['这个群默认讨论 qwen-code'], + 'alice', ); expect(ch.sent).toEqual([ - { chatId: 'group-1', text: 'Channel memory updated.' }, + { chatId: 'group-1', text: 'Channel memory m-000000000001 saved.' }, ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('natural group memory commands do not run without a required mention', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel( { senderPolicy: 'open', @@ -2342,33 +2344,34 @@ describe('ChannelBase', () => { }), ); - expect(channelMemory.appendChannelMemory).not.toHaveBeenCalled(); - expect(channelMemory.readChannelMemory).not.toHaveBeenCalled(); + expect(channelMemory.addChannelMemoryEntries).not.toHaveBeenCalled(); + expect(channelMemory.listChannelMemoryEntries).not.toHaveBeenCalled(); expect(channelMemory.clearChannelMemory).not.toHaveBeenCalled(); expect(ch.sent).toEqual([]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('natural memory list sanitizes stored memory before showing it', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('safe\u202Ehidden\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory([ + { id: 'm-a31f0d82c7e4', text: 'safe\u202Ehidden\n' }, + ]); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); await ch.handleInbound(envelope({ text: '查看记忆', senderId: 'alice' })); - expect(ch.sent).toEqual([{ chatId: 'chat1', text: 'safe hidden' }]); + expect(ch.sent).toEqual([ + { + chatId: 'chat1', + text: 'Channel memory (page 1/1):\nm-a31f0d82c7e4 safe hidden', + }, + ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('natural memory list ignores platform format characters', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory([ + { id: 'm-a31f0d82c7e4', text: 'Use staging.' }, + ]); const ch = createChannel( { allowedUsers: ['alice'], groupPolicy: 'open' }, { channelMemory }, @@ -2384,16 +2387,20 @@ describe('ChannelBase', () => { }), ); - expect(ch.sent).toEqual([{ chatId: 'group-1', text: 'Use staging.' }]); + expect(ch.sent).toEqual([ + { + chatId: 'group-1', + text: 'Channel memory (page 1/1):\nm-a31f0d82c7e4 Use staging.', + }, + ]); expect(bridge.prompt).not.toHaveBeenCalled(); }); it('natural memory list reports read failures', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockRejectedValue(new Error('disk full')), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); + channelMemory.listChannelMemoryEntries.mockRejectedValue( + new Error('disk full'), + ); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); const stderrSpy = vi .spyOn(process.stderr, 'write') @@ -2415,11 +2422,7 @@ describe('ChannelBase', () => { }); it('natural clear requires confirmation and then clears memory', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); await ch.handleInbound(envelope({ text: '清空记忆', senderId: 'alice' })); @@ -2445,11 +2448,7 @@ describe('ChannelBase', () => { }); it('natural group clear uses the current group target and requires confirmation', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel( { allowedUsers: ['alice'], groupPolicy: 'open' }, { channelMemory }, @@ -2496,11 +2495,7 @@ describe('ChannelBase', () => { }); it('natural clear rejects confirm from a different sender', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel( { allowedUsers: ['alice', 'bob'] }, { channelMemory }, @@ -2526,11 +2521,7 @@ describe('ChannelBase', () => { }); it('natural clear rejects confirm from a different thread', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); await ch.handleInbound( @@ -2565,11 +2556,7 @@ describe('ChannelBase', () => { it('natural clear confirm expires after the TTL window', async () => { vi.useFakeTimers(); try { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); const ch = createChannel( { allowedUsers: ['alice'] }, { channelMemory }, @@ -2599,11 +2586,8 @@ describe('ChannelBase', () => { }); it('natural clear reports when no memory was saved', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: false }), - }; + const channelMemory = createChannelMemory(); + channelMemory.clearChannelMemory.mockResolvedValue({ changed: false }); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); await ch.handleInbound(envelope({ text: '清空记忆', senderId: 'alice' })); @@ -2622,11 +2606,8 @@ describe('ChannelBase', () => { }); it('natural clear confirm reports clear failures', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue(''), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockRejectedValue(new Error('EACCES')), - }; + const channelMemory = createChannelMemory(); + channelMemory.clearChannelMemory.mockRejectedValue(new Error('EACCES')); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); const stderrSpy = vi .spyOn(process.stderr, 'write') @@ -2666,14 +2647,423 @@ describe('ChannelBase', () => { expect(bridge.prompt).not.toHaveBeenCalled(); }); - it('keeps legacy channel memory slash commands as hidden aliases', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + it('/help does not expose channel memory commands', async () => { + const ch = createChannel(); + + await ch.handleInbound(envelope({ text: '/help' })); + + expect(ch.sent[0]!.text).not.toContain('/remember-channel'); + expect(ch.sent[0]!.text).not.toContain('/channel-memory'); + expect(ch.sent[0]!.text).not.toContain('/forget-channel'); + expect(bridge.prompt).not.toHaveBeenCalled(); + }); + + it('lists stable pages of sanitized channel memory previews', async () => { + const entries = Array.from({ length: 21 }, (_, index) => ({ + id: `m-${index.toString(16).padStart(12, '0')}`, + text: + index === 0 ? `${'🎉'.repeat(161)}\nignored` : `Memory ${index + 1}`, + createdBy: 'internal-user-id', + })); + const channelMemory = createChannelMemory(entries); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + await ch.handleInbound(envelope({ text: '查看记忆', senderId: 'alice' })); + + const firstPage = ch.sent[0]!.text; + expect(channelMemory.listChannelMemoryEntries).toHaveBeenCalledWith({ + channelName: 'test-chan', + chatId: 'chat1', + threadId: undefined, + }); + expect(firstPage).toMatch(/^Channel memory \(page 1\/2\):/u); + expect(firstPage).toContain(`m-000000000000 ${'🎉'.repeat(160)}`); + expect(firstPage).toContain('m-000000000013 Memory 20'); + expect(firstPage).not.toContain('ignored'); + expect(firstPage).not.toContain('internal-user-id'); + + ch.sent = []; + await ch.handleInbound( + envelope({ text: '查看第 2 页记忆', senderId: 'alice' }), + ); + expect(ch.sent).toEqual([ + { + chatId: 'chat1', + text: 'Channel memory (page 2/2):\nm-000000000014 Memory 21', + }, + ]); + + ch.sent = []; + await ch.handleInbound( + envelope({ text: '查看第 3 页记忆', senderId: 'alice' }), + ); + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'Channel memory page 3 does not exist.' }, + ]); + }); + + it('inspects the full entry without exposing its creator and reports empty lists', async () => { + const channelMemory = createChannelMemory([ + { + id: 'm-a31f0d82c7e4', + text: 'Run tests before release.\nThen deploy.', + createdBy: 'internal-user-id', + }, + ]); + const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + + await ch.handleInbound( + envelope({ + text: '查看记忆 m-a31f0d82c7e4', + senderId: 'alice', + threadId: 'thread-1', + }), + ); + expect(channelMemory.listChannelMemoryEntries).toHaveBeenCalledWith({ + channelName: 'test-chan', + chatId: 'chat1', + threadId: 'thread-1', + }); + expect(ch.sent).toEqual([ + { + chatId: 'chat1', + text: 'Channel memory m-a31f0d82c7e4:\nRun tests before release. Then deploy.', + }, + ]); + + channelMemory.listChannelMemoryEntries.mockResolvedValueOnce([]); + ch.sent = []; + await ch.handleInbound(envelope({ text: '查看记忆', senderId: 'alice' })); + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'No channel memory saved.' }, + ]); + }); + + it('reports missing inspected entries', async () => { + const channelMemory = createChannelMemory(); + const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + + await ch.handleInbound( + envelope({ text: '查看记忆 m-a31f0d82c7e4', senderId: 'alice' }), + ); + + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'No channel memory entry m-a31f0d82c7e4.' }, + ]); + }); + + it('reports empty page one but rejects later pages for empty memory', async () => { + const channelMemory = createChannelMemory(); + const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + + await ch.handleInbound(envelope({ text: '查看记忆', senderId: 'alice' })); + await ch.handleInbound( + envelope({ text: '查看第 2 页记忆', senderId: 'alice' }), + ); + + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'No channel memory saved.' }, + { chatId: 'chat1', text: 'Channel memory page 2 does not exist.' }, + ]); + }); + + it('adds one remembered entry with the sender and reports exact duplicates', async () => { + const channelMemory = createChannelMemory(); + const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + + await ch.handleInbound( + envelope({ text: '记住:Use staging by default.', senderId: 'alice' }), + ); + expect(channelMemory.addChannelMemoryEntries).toHaveBeenCalledWith( + { channelName: 'test-chan', chatId: 'chat1', threadId: undefined }, + ['Use staging by default.'], + 'alice', + ); + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'Channel memory m-000000000001 saved.' }, + ]); + + channelMemory.addChannelMemoryEntries.mockResolvedValueOnce({ + changed: false, + added: [], + duplicateIds: ['m-a31f0d82c7e4'], + }); + ch.sent = []; + await ch.handleInbound( + envelope({ text: '记住:Use staging by default.', senderId: 'alice' }), + ); + expect(ch.sent).toEqual([ + { + chatId: 'chat1', + text: 'Channel memory already contains m-a31f0d82c7e4.', + }, + ]); + }); + + it('only invalidates injected memory after a changed remember result', async () => { + const channelMemory = createChannelMemory(); + channelMemory.readChannelMemory.mockResolvedValue('old memory'); + channelMemory.addChannelMemoryEntries.mockResolvedValue({ + changed: false, + added: [], + duplicateIds: ['m-a31f0d82c7e4'], + }); + const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + + await ch.handleInbound(envelope({ text: 'first', senderId: 'alice' })); + await ch.handleInbound( + envelope({ text: '记住:old memory', senderId: 'alice' }), + ); + await ch.handleInbound(envelope({ text: 'second', senderId: 'alice' })); + + expect(channelMemory.readChannelMemory).toHaveBeenCalledTimes(1); + }); + + it('updates and removes exact entries immediately for current DM and group targets', async () => { + const channelMemory = createChannelMemory(); + channelMemory.updateChannelMemoryEntry.mockResolvedValue({ + changed: true, + entry: { + id: 'm-a31f0d82c7e4', + text: 'Use production.', + createdBy: 'original-author', + }, + }); + channelMemory.removeChannelMemoryEntries.mockResolvedValue({ + changed: true, + removed: [{ id: 'm-b82c4e190a6f', text: 'Old rule.' }], + }); + const ch = createChannel( + { allowedUsers: ['alice'], groupPolicy: 'open' }, + { channelMemory }, + ); + + await ch.handleInbound( + envelope({ + text: '把 m-a31f0d82c7e4 改成Use production.', + senderId: 'alice', + }), + ); + await ch.handleInbound( + envelope({ + text: '忘掉 m-b82c4e190a6f', + senderId: 'alice', + chatId: 'group-1', + isGroup: true, + isMentioned: true, + }), + ); + + expect(channelMemory.updateChannelMemoryEntry).toHaveBeenCalledWith( + { channelName: 'test-chan', chatId: 'chat1', threadId: undefined }, + { id: 'm-a31f0d82c7e4', text: 'Use production.' }, + ); + expect(channelMemory.removeChannelMemoryEntries).toHaveBeenCalledWith( + { channelName: 'test-chan', chatId: 'group-1', threadId: undefined }, + { ids: ['m-b82c4e190a6f'] }, + ); + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'Channel memory m-a31f0d82c7e4 updated.' }, + { + chatId: 'group-1', + text: 'Channel memory m-b82c4e190a6f removed.', + }, + ]); + }); + + it.each([ + { + operation: 'update', + text: '把 m-a31f0d82c7e4 改成Use production.', + }, + { + operation: 'remove', + text: '忘掉 m-a31f0d82c7e4', + }, + ])( + '$operation invalidates matching sessions without invalidating other targets', + async ({ operation, text }) => { + const channelMemory = createChannelMemory(); + channelMemory.readChannelMemory.mockImplementation( + async (target) => `memory for ${target.chatId}`, + ); + channelMemory.updateChannelMemoryEntry.mockResolvedValue({ + changed: true, + entry: { id: 'm-a31f0d82c7e4', text: 'Use production.' }, + }); + channelMemory.removeChannelMemoryEntries.mockResolvedValue({ + changed: true, + removed: [{ id: 'm-a31f0d82c7e4', text: 'Use staging.' }], + }); + const ch = createChannel( + { allowedUsers: ['alice'] }, + { channelMemory }, + ); + + await ch.handleInbound( + envelope({ + text: 'alice first', + senderId: 'alice', + chatId: 'chat-1', + }), + ); + await ch.handleInbound( + envelope({ text: 'bob first', senderId: 'bob', chatId: 'chat-1' }), + ); + await ch.handleInbound( + envelope({ + text: 'carol first', + senderId: 'carol', + chatId: 'chat-2', + }), + ); + expect(channelMemory.readChannelMemory).toHaveBeenCalledTimes(3); + + await ch.handleInbound( + envelope({ text, senderId: 'alice', chatId: 'chat-1' }), + ); + if (operation === 'update') { + expect(channelMemory.updateChannelMemoryEntry).toHaveBeenCalledTimes( + 1, + ); + } else { + expect( + channelMemory.removeChannelMemoryEntries, + ).toHaveBeenCalledTimes(1); + } + + channelMemory.readChannelMemory.mockClear(); + await ch.handleInbound( + envelope({ + text: 'alice second', + senderId: 'alice', + chatId: 'chat-1', + }), + ); + await ch.handleInbound( + envelope({ text: 'bob second', senderId: 'bob', chatId: 'chat-1' }), + ); + await ch.handleInbound( + envelope({ + text: 'carol second', + senderId: 'carol', + chatId: 'chat-2', + }), + ); + + expect(channelMemory.readChannelMemory.mock.calls).toEqual([ + [{ channelName: 'test-chan', chatId: 'chat-1', threadId: undefined }], + [{ channelName: 'test-chan', chatId: 'chat-1', threadId: undefined }], + ]); + }, + ); + + it('does not mutate or invalidate on missing, rejected, or failed item operations', async () => { + const channelMemory = createChannelMemory(); + channelMemory.updateChannelMemoryEntry.mockResolvedValue({ + changed: false, + }); + const ch = createChannel( + { senderPolicy: 'allowlist', allowedUsers: ['alice'] }, + { channelMemory }, + ); + + await ch.handleInbound( + envelope({ + text: '把 m-a31f0d82c7e4 改成Use production.', + senderId: 'alice', + }), + ); + await ch.handleInbound( + envelope({ text: '忘掉 m-b82c4e190a6f', senderId: 'bob' }), + ); + expect(channelMemory.removeChannelMemoryEntries).not.toHaveBeenCalled(); + expect(ch.sent).toEqual([ + { chatId: 'chat1', text: 'No channel memory entry m-a31f0d82c7e4.' }, + ]); + + channelMemory.updateChannelMemoryEntry.mockRejectedValueOnce( + new Error('unsafe\nbackend failure'), + ); + ch.sent = []; + await ch.handleInbound( + envelope({ + text: '把 m-a31f0d82c7e4 改成Use production.', + senderId: 'alice', + }), + ); + expect(ch.sent).toEqual([ + { + chatId: 'chat1', + text: 'Failed to update channel memory: An error occurred while accessing channel memory.', + }, + ]); + }); + + it('allows management but not Recall injection for sessionScope single', async () => { + const channelMemory = createChannelMemory(); + const ch = createChannel( + { allowedUsers: ['alice'], sessionScope: 'single' }, + { channelMemory }, + ); + + await ch.handleInbound( + envelope({ + text: '把 m-a31f0d82c7e4 改成Use production.', + senderId: 'alice', + }), + ); + await ch.handleInbound( + envelope({ text: 'normal prompt', senderId: 'alice' }), + ); + + expect(channelMemory.updateChannelMemoryEntry).toHaveBeenCalled(); + expect(channelMemory.readChannelMemory).not.toHaveBeenCalled(); + }); + + it('does not invalidate injected memory after failed update or unchanged clear', async () => { + const channelMemory = createChannelMemory(); + channelMemory.readChannelMemory.mockResolvedValue('old memory'); + channelMemory.updateChannelMemoryEntry.mockRejectedValue( + new Error('backend unavailable'), + ); + channelMemory.clearChannelMemory.mockResolvedValue({ changed: false }); + const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + + await ch.handleInbound(envelope({ text: 'first', senderId: 'alice' })); + await ch.handleInbound( + envelope({ + text: '把 m-a31f0d82c7e4 改成Use production.', + senderId: 'alice', + }), + ); + await ch.handleInbound(envelope({ text: 'second', senderId: 'alice' })); + await ch.handleInbound(envelope({ text: '清空记忆', senderId: 'alice' })); + await ch.handleInbound( + envelope({ text: '确认清空记忆', senderId: 'alice' }), + ); + await ch.handleInbound(envelope({ text: 'third', senderId: 'alice' })); + + expect(channelMemory.readChannelMemory).toHaveBeenCalledTimes(1); + }); + + it('forwards hidden memory slash aliases after Recall without invoking memory management', async () => { + const channelMemory = createChannelMemory(); + const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); + + await ch.handleInbound( + envelope({ text: 'prewarm session', senderId: 'alice' }), + ); + expect(channelMemory.readChannelMemory).toHaveBeenCalledTimes(1); + + channelMemory.readChannelMemory.mockClear(); + channelMemory.listChannelMemoryEntries.mockClear(); + channelMemory.addChannelMemoryEntries.mockClear(); + channelMemory.updateChannelMemoryEntry.mockClear(); + channelMemory.removeChannelMemoryEntries.mockClear(); + channelMemory.clearChannelMemory.mockClear(); + (bridge.prompt as ReturnType).mockClear(); + await ch.handleInbound( envelope({ text: '/remember-channel Use staging.', senderId: 'alice' }), ); @@ -2684,101 +3074,22 @@ describe('ChannelBase', () => { envelope({ text: '/forget-channel confirm', senderId: 'alice' }), ); - expect(channelMemory.appendChannelMemory).toHaveBeenCalledWith( - { - channelName: 'test-chan', - chatId: 'chat1', - threadId: undefined, - }, - 'Use staging.', - ); - expect(channelMemory.readChannelMemory).toHaveBeenCalledWith({ - channelName: 'test-chan', - chatId: 'chat1', - threadId: undefined, - }); - expect(channelMemory.clearChannelMemory).toHaveBeenCalledWith({ - channelName: 'test-chan', - chatId: 'chat1', - threadId: undefined, - }); - expect(ch.sent).toEqual([ - { chatId: 'chat1', text: 'Channel memory updated.' }, - { chatId: 'chat1', text: 'Use staging.' }, - { chatId: 'chat1', text: 'Channel memory cleared.' }, + expect(bridge.prompt).toHaveBeenCalledTimes(3); + expect( + (bridge.prompt as ReturnType).mock.calls.map( + (call) => call[1], + ), + ).toEqual([ + '/remember-channel Use staging.', + '/channel-memory', + '/forget-channel confirm', ]); - expect(bridge.prompt).not.toHaveBeenCalled(); - }); - - it('slash memory aliases follow open group access control', async () => { - const channelMemory = { - readChannelMemory: vi.fn().mockResolvedValue('Use staging.\n'), - appendChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; - const ch = createChannel( - { - senderPolicy: 'open', - allowedUsers: [], - groupPolicy: 'open', - }, - { channelMemory }, - ); - const groupEnvelope = { - senderId: 'alice', - isGroup: true, - chatId: 'group-1', - isMentioned: true, - }; - - await ch.handleInbound( - envelope({ - ...groupEnvelope, - text: '/remember-channel Use staging.', - }), - ); - await ch.handleInbound( - envelope({ ...groupEnvelope, text: '/channel-memory' }), - ); - await ch.handleInbound( - envelope({ ...groupEnvelope, text: '/forget-channel confirm' }), - ); - - expect(channelMemory.appendChannelMemory).toHaveBeenCalledWith( - { - channelName: 'test-chan', - chatId: 'group-1', - threadId: undefined, - }, - 'Use staging.', - ); - expect(channelMemory.readChannelMemory).toHaveBeenCalledWith({ - channelName: 'test-chan', - chatId: 'group-1', - threadId: undefined, - }); - expect(channelMemory.clearChannelMemory).toHaveBeenCalledWith({ - channelName: 'test-chan', - chatId: 'group-1', - threadId: undefined, - }); - expect(ch.sent).toEqual([ - { chatId: 'group-1', text: 'Channel memory updated.' }, - { chatId: 'group-1', text: 'Use staging.' }, - { chatId: 'group-1', text: 'Channel memory cleared.' }, - ]); - expect(bridge.prompt).not.toHaveBeenCalled(); - }); - - it('/help does not expose channel memory commands', async () => { - const ch = createChannel(); - - await ch.handleInbound(envelope({ text: '/help' })); - - expect(ch.sent[0]!.text).not.toContain('/remember-channel'); - expect(ch.sent[0]!.text).not.toContain('/channel-memory'); - expect(ch.sent[0]!.text).not.toContain('/forget-channel'); - expect(bridge.prompt).not.toHaveBeenCalled(); + expect(channelMemory.readChannelMemory).not.toHaveBeenCalled(); + expect(channelMemory.addChannelMemoryEntries).not.toHaveBeenCalled(); + expect(channelMemory.listChannelMemoryEntries).not.toHaveBeenCalled(); + expect(channelMemory.updateChannelMemoryEntry).not.toHaveBeenCalled(); + expect(channelMemory.removeChannelMemoryEntries).not.toHaveBeenCalled(); + expect(channelMemory.clearChannelMemory).not.toHaveBeenCalled(); }); it('/clear removes session and confirms', async () => { @@ -6268,19 +6579,21 @@ describe('ChannelBase', () => { it('re-reads memory for a collect followup buffered after memory changes', async () => { let memory = 'old memory'; let reads = 0; - const channelMemory = { - readChannelMemory: vi.fn().mockImplementation(() => { - reads += 1; - return memory; - }), - appendChannelMemory: vi - .fn() - .mockImplementation(async (_target: unknown, text: string) => { - memory = `${memory}\n${text}`; - return { changed: true }; - }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); + channelMemory.readChannelMemory.mockImplementation(() => { + reads += 1; + return memory; + }); + channelMemory.addChannelMemoryEntries.mockImplementation( + async (_target: unknown, texts: readonly string[]) => { + memory = `${memory}\n${texts[0]}`; + return { + changed: true, + added: [{ id: 'm-a31f0d82c7e4', text: texts[0]! }], + duplicateIds: [], + }; + }, + ); let resolveFirst!: (value: string) => void; const firstPrompt = new Promise((resolve) => { resolveFirst = resolve; @@ -6433,19 +6746,21 @@ describe('ChannelBase', () => { it('natural remember invalidates current session context after append', async () => { let memory = 'old memory'; let reads = 0; - const channelMemory = { - readChannelMemory: vi.fn().mockImplementation(() => { - reads += 1; - return memory; - }), - appendChannelMemory: vi - .fn() - .mockImplementation(async (_target: unknown, text: string) => { - memory = `${memory}\n${text}`; - return { changed: true }; - }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); + channelMemory.readChannelMemory.mockImplementation(() => { + reads += 1; + return memory; + }); + channelMemory.addChannelMemoryEntries.mockImplementation( + async (_target: unknown, texts: readonly string[]) => { + memory = `${memory}\n${texts[0]}`; + return { + changed: true, + added: [{ id: 'm-a31f0d82c7e4', text: texts[0]! }], + duplicateIds: [], + }; + }, + ); const ch = createChannel({ allowedUsers: ['alice'] }, { channelMemory }); await ch.handleInbound(envelope({ text: 'first', senderId: 'alice' })); @@ -6463,19 +6778,21 @@ describe('ChannelBase', () => { it('natural group remember invalidates other sender sessions for the same group memory', async () => { let memory = 'old memory'; let reads = 0; - const channelMemory = { - readChannelMemory: vi.fn().mockImplementation(() => { - reads += 1; - return memory; - }), - appendChannelMemory: vi - .fn() - .mockImplementation(async (_target: unknown, text: string) => { - memory = text; - return { changed: true }; - }), - clearChannelMemory: vi.fn().mockResolvedValue({ changed: true }), - }; + const channelMemory = createChannelMemory(); + channelMemory.readChannelMemory.mockImplementation(() => { + reads += 1; + return memory; + }); + channelMemory.addChannelMemoryEntries.mockImplementation( + async (_target: unknown, texts: readonly string[]) => { + memory = texts[0]!; + return { + changed: true, + added: [{ id: 'm-a31f0d82c7e4', text: texts[0]! }], + duplicateIds: [], + }; + }, + ); const ch = createChannel( { groupPolicy: 'open', sessionScope: 'user' }, { channelMemory }, diff --git a/packages/channels/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index 93cd2a69a9..aaf0ef6f7c 100644 --- a/packages/channels/base/src/ChannelBase.ts +++ b/packages/channels/base/src/ChannelBase.ts @@ -73,6 +73,8 @@ const GROUP_HISTORY_ENTRY_TEXT_LIMIT = 1000; const GROUP_HISTORY_ENTRY_METADATA_LIMIT = 256; const LOOP_CANCEL_GRACE_MS = 5000; const CHANNEL_MEMORY_PROMPT_CODE_POINT_LIMIT = 12_000; +const CHANNEL_MEMORY_PAGE_SIZE = 20; +const CHANNEL_MEMORY_PREVIEW_CODE_POINT_LIMIT = 160; const CHANNEL_MEMORY_CLASSIFIER_MIN_CONFIDENCE = 0.7; const CHANNEL_MEMORY_CLASSIFIER_TRIGGER_RE = /(记住|记得|记一下|记忆|忘掉|忘记|清空|清除|删除|保存|remember|memory|forget)/iu; @@ -2036,43 +2038,6 @@ export abstract class ChannelBase { this.handlePermissionResponseCommand(envelope, args, 'deny'), ); - this.registerCommand('remember-channel', async (envelope, args) => { - const text = args.trim(); - if (text === '') { - await this.sendMessage( - envelope.chatId, - 'Usage: /remember-channel ', - ); - return true; - } - await this.handleChannelMemoryIntent(envelope, { - kind: 'remember', - text, - }); - return true; - }); - - this.registerCommand('channel-memory', async (envelope) => { - await this.handleChannelMemoryIntent(envelope, { kind: 'list' }); - return true; - }); - - this.registerCommand('forget-channel', async (envelope, args) => { - if (args.toLowerCase() !== 'confirm') { - await this.sendMessage( - envelope.chatId, - 'This clears channel memory for this chat. Re-send with "confirm" (e.g. /forget-channel confirm) to proceed.', - ); - return true; - } - await this.handleChannelMemoryIntent( - envelope, - { kind: 'clear_confirm' }, - { skipPendingClear: true }, - ); - return true; - }); - // Read-only: report the current (possibly group-shared) session and workspace. // For a shared session, gate it to authorized senders like /clear — /who // leaks the workspace basename, so non-members shouldn't see it either. @@ -2751,7 +2716,6 @@ export abstract class ChannelBase { private async handleChannelMemoryIntent( envelope: Envelope, intent: ChannelMemoryIntent, - options: { skipPendingClear?: boolean } = {}, ): Promise { if (intent.kind === 'clear_request') { this.setPendingClear(this.clearPendingKey(envelope)); @@ -2768,10 +2732,16 @@ export abstract class ChannelBase { } if (intent.kind === 'remember') { + let result: { + changed: boolean; + added: Array<{ id: string }>; + duplicateIds: string[]; + }; try { - await channelMemory.appendChannelMemory( + result = await channelMemory.addChannelMemoryEntries( this.channelMemoryTarget(envelope), - intent.text, + [intent.text], + envelope.senderId, ); } catch (error) { const message = this.channelMemoryErrorMessage(error); @@ -2782,19 +2752,34 @@ export abstract class ChannelBase { ); return; } - this.invalidateSessionContext(envelope); - await this.sendMessage(envelope.chatId, 'Channel memory updated.'); + if (result.changed) { + this.invalidateSessionContext(envelope); + } + if (result.added.length > 0) { + const ids = result.added.map((entry) => entry.id); + await this.sendMessage( + envelope.chatId, + ids.length === 1 + ? `Channel memory ${ids[0]} saved.` + : `Channel memory saved: ${ids.join(', ')}.`, + ); + } else if (result.duplicateIds.length > 0) { + await this.sendMessage( + envelope.chatId, + `Channel memory already contains ${result.duplicateIds.join(', ')}.`, + ); + } else { + await this.sendMessage(envelope.chatId, 'Channel memory updated.'); + } return; } - if (intent.kind === 'list') { - let text: string; + if (intent.kind === 'list' || intent.kind === 'inspect') { + let entries; try { - text = ( - await channelMemory.readChannelMemory( - this.channelMemoryTarget(envelope), - ) - ).trim(); + entries = await channelMemory.listChannelMemoryEntries( + this.channelMemoryTarget(envelope), + ); } catch (error) { const message = this.channelMemoryErrorMessage(error); this.logChannelMemoryError('read', envelope, message); @@ -2804,25 +2789,104 @@ export abstract class ChannelBase { ); return; } + if (intent.kind === 'inspect') { + const entry = entries.find((candidate) => candidate.id === intent.id); + await this.sendMessage( + envelope.chatId, + entry + ? `Channel memory ${entry.id}:\n${sanitizePromptText(entry.text).trim()}` + : `No channel memory entry ${intent.id}.`, + ); + return; + } + const totalPages = Math.max( + 1, + Math.ceil(entries.length / CHANNEL_MEMORY_PAGE_SIZE), + ); + if (intent.page > totalPages) { + await this.sendMessage( + envelope.chatId, + `Channel memory page ${intent.page} does not exist.`, + ); + return; + } + if (entries.length === 0) { + await this.sendMessage(envelope.chatId, 'No channel memory saved.'); + return; + } + const pageStart = (intent.page - 1) * CHANNEL_MEMORY_PAGE_SIZE; + const lines = entries + .slice(pageStart, pageStart + CHANNEL_MEMORY_PAGE_SIZE) + .map((entry) => { + const preview = truncateCodePoints( + sanitizePromptText(entry.text) + .replace(/[\r\n]+/gu, ' ') + .trim(), + CHANNEL_MEMORY_PREVIEW_CODE_POINT_LIMIT, + ); + return `${entry.id} ${preview}`; + }); await this.sendMessage( envelope.chatId, - text === '' ? 'No channel memory saved.' : sanitizePromptText(text), + [`Channel memory (page ${intent.page}/${totalPages}):`, ...lines].join( + '\n', + ), + ); + return; + } + + if (intent.kind === 'update' || intent.kind === 'remove') { + let changed: boolean; + try { + if (intent.kind === 'update') { + ({ changed } = await channelMemory.updateChannelMemoryEntry( + this.channelMemoryTarget(envelope), + { id: intent.id, text: intent.text }, + )); + } else { + ({ changed } = await channelMemory.removeChannelMemoryEntries( + this.channelMemoryTarget(envelope), + { ids: [intent.id] }, + )); + } + } catch (error) { + const message = this.channelMemoryErrorMessage(error); + this.logChannelMemoryError( + intent.kind === 'update' ? 'update' : 'remove', + envelope, + message, + ); + await this.sendMessage( + envelope.chatId, + `Failed to ${intent.kind === 'update' ? 'update' : 'remove'} channel memory: ${this.channelMemoryUserErrorMessage()}`, + ); + return; + } + if (!changed) { + await this.sendMessage( + envelope.chatId, + `No channel memory entry ${intent.id}.`, + ); + return; + } + this.invalidateSessionContext(envelope); + await this.sendMessage( + envelope.chatId, + `Channel memory ${intent.id} ${intent.kind === 'update' ? 'updated' : 'removed'}.`, ); return; } if (intent.kind === 'clear_confirm') { - if (!options.skipPendingClear) { - const pendingKey = this.clearPendingKey(envelope); - const expiresAt = this.pendingClears.get(pendingKey); - this.pendingClears.delete(pendingKey); - if (expiresAt === undefined || expiresAt < Date.now()) { - await this.sendMessage( - envelope.chatId, - 'No pending clear request. Say "清空记忆" first.', - ); - return; - } + const pendingKey = this.clearPendingKey(envelope); + const expiresAt = this.pendingClears.get(pendingKey); + this.pendingClears.delete(pendingKey); + if (expiresAt === undefined || expiresAt < Date.now()) { + await this.sendMessage( + envelope.chatId, + 'No pending clear request. Say "清空记忆" first.', + ); + return; } let result: { changed: boolean }; @@ -2839,7 +2903,9 @@ export abstract class ChannelBase { ); return; } - this.invalidateSessionContext(envelope); + if (result.changed) { + this.invalidateSessionContext(envelope); + } await this.sendMessage( envelope.chatId, result.changed ? 'Channel memory cleared.' : 'No channel memory saved.', @@ -2909,7 +2975,7 @@ export abstract class ChannelBase { return memory ? { kind: 'remember', text: memory } : null; } if (classified.intent === 'list') { - return { kind: 'list' }; + return { kind: 'list', page: 1 }; } if (classified.intent === 'clear_all') { return { kind: 'clear_request' }; @@ -2926,7 +2992,7 @@ export abstract class ChannelBase { } private logChannelMemoryError( - action: 'save' | 'read' | 'clear', + action: 'save' | 'read' | 'update' | 'remove' | 'clear', envelope: Envelope, message: string, ): void { diff --git a/packages/channels/base/src/channel-memory-intent.test.ts b/packages/channels/base/src/channel-memory-intent.test.ts index ae125917a7..9c31ad18d5 100644 --- a/packages/channels/base/src/channel-memory-intent.test.ts +++ b/packages/channels/base/src/channel-memory-intent.test.ts @@ -34,13 +34,79 @@ describe('parseChannelMemoryIntent', () => { it('parses list requests', () => { expect(parseChannelMemoryIntent('你现在记住了什么?')).toEqual({ kind: 'list', + page: 1, + }); + expect(parseChannelMemoryIntent('查看记忆')).toEqual({ + kind: 'list', + page: 1, }); - expect(parseChannelMemoryIntent('查看记忆')).toEqual({ kind: 'list' }); expect(parseChannelMemoryIntent('查看记忆\u200b')).toEqual({ kind: 'list', + page: 1, }); expect(parseChannelMemoryIntent('what do you remember?')).toEqual({ kind: 'list', + page: 1, + }); + }); + + it('parses deterministic item intents', () => { + expect(parseChannelMemoryIntent('查看第 2 页记忆')).toEqual({ + kind: 'list', + page: 2, + }); + expect(parseChannelMemoryIntent('show memory page 3')).toEqual({ + kind: 'list', + page: 3, + }); + expect(parseChannelMemoryIntent('查看记忆 m-a31f0d82c7e4')).toEqual({ + kind: 'inspect', + id: 'm-a31f0d82c7e4', + }); + expect(parseChannelMemoryIntent('show memory m-a31f0d82c7e4')).toEqual({ + kind: 'inspect', + id: 'm-a31f0d82c7e4', + }); + expect(parseChannelMemoryIntent('忘掉 m-a31f0d82c7e4')).toEqual({ + kind: 'remove', + id: 'm-a31f0d82c7e4', + }); + expect(parseChannelMemoryIntent('forget m-a31f0d82c7e4')).toEqual({ + kind: 'remove', + id: 'm-a31f0d82c7e4', + }); + expect( + parseChannelMemoryIntent('把 m-a31f0d82c7e4 改成默认使用 production'), + ).toEqual({ + kind: 'update', + id: 'm-a31f0d82c7e4', + text: '默认使用 production', + }); + expect( + parseChannelMemoryIntent('update m-a31f0d82c7e4 to use production'), + ).toEqual({ + kind: 'update', + id: 'm-a31f0d82c7e4', + text: 'use production', + }); + }); + + it('rejects invalid item intent arguments', () => { + expect(parseChannelMemoryIntent('忘掉 m-not-valid')).toBeNull(); + expect(parseChannelMemoryIntent('查看第 0 页记忆')).toBeNull(); + expect(parseChannelMemoryIntent('查看第 -1 页记忆')).toBeNull(); + expect(parseChannelMemoryIntent('show memory page 1.5')).toBeNull(); + expect(parseChannelMemoryIntent('把 m-a31f0d82c7e4 改成 ')).toBeNull(); + expect(parseChannelMemoryIntent('/forget m-a31f0d82c7e4')).toBeNull(); + }); + + it('parses item updates before broad clear requests', () => { + expect( + parseChannelMemoryIntent('把 m-a31f0d82c7e4 改成默认的记忆清空'), + ).toEqual({ + kind: 'update', + id: 'm-a31f0d82c7e4', + text: '默认的记忆清空', }); }); diff --git a/packages/channels/base/src/channel-memory-intent.ts b/packages/channels/base/src/channel-memory-intent.ts index e799d97a56..1df0394781 100644 --- a/packages/channels/base/src/channel-memory-intent.ts +++ b/packages/channels/base/src/channel-memory-intent.ts @@ -2,7 +2,10 @@ import { PROMPT_UNSAFE_INVISIBLES } from './sanitize.js'; export type ChannelMemoryIntent = | { kind: 'remember'; text: string } - | { kind: 'list' } + | { kind: 'list'; page: number } + | { kind: 'inspect'; id: string } + | { kind: 'remove'; id: string } + | { kind: 'update'; id: string; text: string } | { kind: 'clear_request' } | { kind: 'clear_confirm' }; @@ -23,6 +26,25 @@ const LIST_PATTERNS: RegExp[] = [ /^what do you remember[??]?$/iu, ]; +const LIST_PAGE_PATTERNS: RegExp[] = [ + /^查看第\s*(\d+)\s*页记忆$/u, + /^show memory page\s+(\d+)$/iu, +]; + +const INSPECT_PATTERNS: RegExp[] = [ + /^查看记忆\s+(\S+)$/u, + /^show memory\s+(\S+)$/iu, +]; + +const REMOVE_PATTERNS: RegExp[] = [/^忘掉\s+(\S+)$/u, /^forget\s+(\S+)$/iu]; + +const UPDATE_PATTERNS: RegExp[] = [ + /^把\s+(\S+)\s+改成\s*(.+)$/su, + /^update\s+(\S+)\s+to\s+(.+)$/isu, +]; + +const MEMORY_ID_PATTERN = /^m-[a-f0-9]{12}$/u; + const CLEAR_REQUEST_PATTERNS: RegExp[] = [ /^清空记忆$/u, /^清除记忆$/u, @@ -50,14 +72,43 @@ export function parseChannelMemoryIntent( return { kind: 'clear_confirm' }; } } + for (const pattern of REMOVE_PATTERNS) { + const match = trimmed.match(pattern); + if (match?.[1] && MEMORY_ID_PATTERN.test(match[1])) { + return { kind: 'remove', id: match[1] }; + } + } + for (const pattern of UPDATE_PATTERNS) { + const match = trimmed.match(pattern); + const id = match?.[1]; + const updated = match?.[2]?.trim(); + if (id && updated && MEMORY_ID_PATTERN.test(id)) { + return { kind: 'update', id, text: updated }; + } + } for (const pattern of CLEAR_REQUEST_PATTERNS) { if (pattern.test(trimmed)) { return { kind: 'clear_request' }; } } + for (const pattern of INSPECT_PATTERNS) { + const match = trimmed.match(pattern); + if (match?.[1] && MEMORY_ID_PATTERN.test(match[1])) { + return { kind: 'inspect', id: match[1] }; + } + } + for (const pattern of LIST_PAGE_PATTERNS) { + const match = trimmed.match(pattern); + if (match?.[1]) { + const page = Number(match[1]); + return Number.isSafeInteger(page) && page > 0 + ? { kind: 'list', page } + : null; + } + } for (const pattern of LIST_PATTERNS) { if (pattern.test(trimmed)) { - return { kind: 'list' }; + return { kind: 'list', page: 1 }; } } for (const pattern of REMEMBER_PATTERNS) { diff --git a/packages/channels/base/src/types.ts b/packages/channels/base/src/types.ts index efa016c356..8d58dfde0c 100644 --- a/packages/channels/base/src/types.ts +++ b/packages/channels/base/src/types.ts @@ -197,20 +197,39 @@ export interface ChannelMemoryTarget { threadId?: string; } -export interface ChannelMemoryWriteResult { - changed: boolean; - filePath?: string; +export interface ChannelMemoryEntry { + id: string; + text: string; + createdAt?: string; + updatedAt?: string; + createdBy?: string; } export interface ChannelMemoryCallbacks { readChannelMemory(target: ChannelMemoryTarget): Promise; - appendChannelMemory( + listChannelMemoryEntries( target: ChannelMemoryTarget, - text: string, - ): Promise; - clearChannelMemory( + ): Promise; + addChannelMemoryEntries( target: ChannelMemoryTarget, - ): Promise; + texts: readonly string[], + createdBy?: string, + ): Promise<{ + changed: boolean; + added: ChannelMemoryEntry[]; + duplicateIds: string[]; + }>; + updateChannelMemoryEntry( + target: ChannelMemoryTarget, + mutation: { id: string; text: string }, + ): Promise<{ changed: boolean; entry?: ChannelMemoryEntry }>; + removeChannelMemoryEntries( + target: ChannelMemoryTarget, + mutation: { ids: readonly string[] }, + ): Promise<{ changed: boolean; removed: ChannelMemoryEntry[] }>; + clearChannelMemory(target: ChannelMemoryTarget): Promise<{ + changed: boolean; + }>; } export interface ChannelMemoryIntentClassifierResult { diff --git a/packages/cli/src/commands/channel/daemon-worker.test.ts b/packages/cli/src/commands/channel/daemon-worker.test.ts index 6cc8611729..8aa1638a97 100644 --- a/packages/cli/src/commands/channel/daemon-worker.test.ts +++ b/packages/cli/src/commands/channel/daemon-worker.test.ts @@ -6,7 +6,10 @@ const mockLoadChannelsFromExtensions = vi.hoisted(() => vi.fn()); const mockParseConfiguredChannels = vi.hoisted(() => vi.fn()); const mockCreateChannel = vi.hoisted(() => vi.fn()); const mockReadChannelMemory = vi.hoisted(() => vi.fn()); -const mockAppendChannelMemory = vi.hoisted(() => vi.fn()); +const mockListChannelMemoryEntries = vi.hoisted(() => vi.fn()); +const mockAddChannelMemoryEntries = vi.hoisted(() => vi.fn()); +const mockUpdateChannelMemoryEntry = vi.hoisted(() => vi.fn()); +const mockRemoveChannelMemoryEntries = vi.hoisted(() => vi.fn()); const mockClearChannelMemory = vi.hoisted(() => vi.fn()); const mockRegisterToolCallDispatch = vi.hoisted(() => vi.fn()); const mockRegisterPermissionRelay = vi.hoisted(() => vi.fn()); @@ -130,9 +133,12 @@ vi.mock('@qwen-code/acp-bridge/workspacePaths', () => ({ })); vi.mock('@qwen-code/qwen-code-core', () => ({ - appendChannelMemory: mockAppendChannelMemory, + addChannelMemoryEntries: mockAddChannelMemoryEntries, clearChannelMemory: mockClearChannelMemory, + listChannelMemoryEntries: mockListChannelMemoryEntries, readChannelMemory: mockReadChannelMemory, + removeChannelMemoryEntries: mockRemoveChannelMemoryEntries, + updateChannelMemoryEntry: mockUpdateChannelMemoryEntry, })); vi.mock('../../utils/stdioHelpers.js', () => ({ @@ -633,9 +639,12 @@ describe('runChannelDaemonWorker', () => { proxy: 'http://settings-proxy:8080', router: mockSessionRouter.mock.results[0]!.value, channelMemory: { - appendChannelMemory: mockAppendChannelMemory, - clearChannelMemory: mockClearChannelMemory, readChannelMemory: mockReadChannelMemory, + listChannelMemoryEntries: mockListChannelMemoryEntries, + addChannelMemoryEntries: mockAddChannelMemoryEntries, + updateChannelMemoryEntry: mockUpdateChannelMemoryEntry, + removeChannelMemoryEntries: mockRemoveChannelMemoryEntries, + clearChannelMemory: mockClearChannelMemory, }, memoryIntentClassifier: expect.objectContaining({ classifyChannelMemoryIntent: expect.any(Function), diff --git a/packages/cli/src/commands/channel/daemon-worker.ts b/packages/cli/src/commands/channel/daemon-worker.ts index 32047fc2eb..0da3fa43b7 100644 --- a/packages/cli/src/commands/channel/daemon-worker.ts +++ b/packages/cli/src/commands/channel/daemon-worker.ts @@ -1,9 +1,12 @@ import type { CommandModule } from 'yargs'; import { canonicalizeWorkspace } from '@qwen-code/acp-bridge/workspacePaths'; import { - appendChannelMemory, + addChannelMemoryEntries, clearChannelMemory, + listChannelMemoryEntries, readChannelMemory, + removeChannelMemoryEntries, + updateChannelMemoryEntry, } from '@qwen-code/qwen-code-core'; import { loadSettings } from '../../config/settings.js'; import { @@ -415,7 +418,10 @@ export async function runChannelDaemonWorker( router: createdRouter, channelMemory: { readChannelMemory, - appendChannelMemory, + listChannelMemoryEntries, + addChannelMemoryEntries, + updateChannelMemoryEntry, + removeChannelMemoryEntries, clearChannelMemory, }, memoryIntentClassifier: new BridgeChannelMemoryIntentClassifier( diff --git a/packages/cli/src/commands/channel/start.test.ts b/packages/cli/src/commands/channel/start.test.ts index 8a9c8cc401..f1553cd1d4 100644 --- a/packages/cli/src/commands/channel/start.test.ts +++ b/packages/cli/src/commands/channel/start.test.ts @@ -14,7 +14,10 @@ const mockStorageGetGlobalQwenDir = vi.hoisted(() => vi.fn(() => '/tmp/qwen-home'), ); const mockReadChannelMemory = vi.hoisted(() => vi.fn()); -const mockAppendChannelMemory = vi.hoisted(() => vi.fn()); +const mockListChannelMemoryEntries = vi.hoisted(() => vi.fn()); +const mockAddChannelMemoryEntries = vi.hoisted(() => vi.fn()); +const mockUpdateChannelMemoryEntry = vi.hoisted(() => vi.fn()); +const mockRemoveChannelMemoryEntries = vi.hoisted(() => vi.fn()); const mockClearChannelMemory = vi.hoisted(() => vi.fn()); const mockParseCron = vi.hoisted(() => vi.fn()); const mockNextFireTime = vi.hoisted(() => @@ -101,12 +104,15 @@ vi.mock('undici', () => ({ })); vi.mock('@qwen-code/qwen-code-core', () => ({ - appendChannelMemory: mockAppendChannelMemory, + addChannelMemoryEntries: mockAddChannelMemoryEntries, clearChannelMemory: mockClearChannelMemory, + listChannelMemoryEntries: mockListChannelMemoryEntries, nextFireTime: mockNextFireTime, normalizeProxyUrl: mockNormalizeProxyUrl, parseCron: mockParseCron, readChannelMemory: mockReadChannelMemory, + removeChannelMemoryEntries: mockRemoveChannelMemoryEntries, + updateChannelMemoryEntry: mockUpdateChannelMemoryEntry, Storage: { getGlobalQwenDir: mockStorageGetGlobalQwenDir, }, @@ -865,9 +871,12 @@ describe('startCommand.handler', () => { expect.any(Object), expect.objectContaining({ channelMemory: { - appendChannelMemory: mockAppendChannelMemory, - clearChannelMemory: mockClearChannelMemory, readChannelMemory: mockReadChannelMemory, + listChannelMemoryEntries: mockListChannelMemoryEntries, + addChannelMemoryEntries: mockAddChannelMemoryEntries, + updateChannelMemoryEntry: mockUpdateChannelMemoryEntry, + removeChannelMemoryEntries: mockRemoveChannelMemoryEntries, + clearChannelMemory: mockClearChannelMemory, }, memoryIntentClassifier: expect.objectContaining({ classifyChannelMemoryIntent: expect.any(Function), @@ -903,9 +912,12 @@ describe('startCommand.handler', () => { expect.any(Object), expect.objectContaining({ channelMemory: { - appendChannelMemory: mockAppendChannelMemory, - clearChannelMemory: mockClearChannelMemory, readChannelMemory: mockReadChannelMemory, + listChannelMemoryEntries: mockListChannelMemoryEntries, + addChannelMemoryEntries: mockAddChannelMemoryEntries, + updateChannelMemoryEntry: mockUpdateChannelMemoryEntry, + removeChannelMemoryEntries: mockRemoveChannelMemoryEntries, + clearChannelMemory: mockClearChannelMemory, }, memoryIntentClassifier: expect.objectContaining({ classifyChannelMemoryIntent: expect.any(Function), @@ -919,9 +931,12 @@ describe('startCommand.handler', () => { expect.any(Object), expect.objectContaining({ channelMemory: { - appendChannelMemory: mockAppendChannelMemory, - clearChannelMemory: mockClearChannelMemory, readChannelMemory: mockReadChannelMemory, + listChannelMemoryEntries: mockListChannelMemoryEntries, + addChannelMemoryEntries: mockAddChannelMemoryEntries, + updateChannelMemoryEntry: mockUpdateChannelMemoryEntry, + removeChannelMemoryEntries: mockRemoveChannelMemoryEntries, + clearChannelMemory: mockClearChannelMemory, }, }), ); diff --git a/packages/cli/src/commands/channel/start.ts b/packages/cli/src/commands/channel/start.ts index 95b4bd6b47..f5d14050ed 100644 --- a/packages/cli/src/commands/channel/start.ts +++ b/packages/cli/src/commands/channel/start.ts @@ -1,10 +1,13 @@ import type { CommandModule } from 'yargs'; import { - appendChannelMemory, + addChannelMemoryEntries, clearChannelMemory, + listChannelMemoryEntries, nextFireTime, parseCron, readChannelMemory, + removeChannelMemoryEntries, + updateChannelMemoryEntry, } from '@qwen-code/qwen-code-core'; import { loadSettings } from '../../config/settings.js'; import { writeStderrLine, writeStdoutLine } from '../../utils/stdioHelpers.js'; @@ -62,7 +65,10 @@ function channelMemoryOptions( return { channelMemory: { readChannelMemory, - appendChannelMemory, + listChannelMemoryEntries, + addChannelMemoryEntries, + updateChannelMemoryEntry, + removeChannelMemoryEntries, clearChannelMemory, }, memoryIntentClassifier: new BridgeChannelMemoryIntentClassifier( diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 8cd45bf157..6b0b57510b 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -323,6 +323,7 @@ export * from './memory/types.js'; export * from './memory/paths.js'; export * from './memory/store.js'; export * from './memory/const.js'; +export * from './memory/channel-memory-document.js'; export * from './memory/channel-memory.js'; export * from './memory/remember.js'; export * from './memory/refresh.js'; diff --git a/packages/core/src/memory/channel-memory-document.test.ts b/packages/core/src/memory/channel-memory-document.test.ts new file mode 100644 index 0000000000..748fee5371 --- /dev/null +++ b/packages/core/src/memory/channel-memory-document.test.ts @@ -0,0 +1,300 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it } from 'vitest'; +import { + CHANNEL_MEMORY_ID_RE, + MAX_CHANNEL_MEMORY_ENTRIES, + MAX_CHANNEL_MEMORY_ENTRY_CODE_POINTS, + createChannelMemoryEntry, + normalizeChannelMemoryText, + parseChannelMemoryDocument, + parseLegacyChannelMemory, + renderChannelMemoryRecall, + serializeChannelMemoryDocument, +} from './channel-memory-document.js'; + +describe('channel memory document', () => { + it('normalizes channel memory text', () => { + expect(normalizeChannelMemoryText(' USE\u00a0staging ')).toBe( + 'use staging', + ); + }); + + it('rejects unsupported document versions', () => { + expect(() => + parseChannelMemoryDocument('{"version":2,"entries":[]}'), + ).toThrow('Unsupported channel memory version'); + }); + + it('rejects entries with invalid ids', () => { + expect(() => + parseChannelMemoryDocument( + JSON.stringify({ + version: 1, + entries: [ + { id: 'bad', text: 'x' }, + { id: 'm-123456789abc', text: 'y' }, + ], + }), + ), + ).toThrow('Invalid channel memory entry'); + }); + + it('validates the complete version-1 document shape', () => { + const document = parseChannelMemoryDocument( + JSON.stringify({ + version: 1, + migration: { legacySha256: 'a'.repeat(64) }, + entries: [ + { + id: 'm-123456789abc', + text: 'Use staging', + createdAt: '2026-07-14T00:00:00.000Z', + updatedAt: '2026-07-14T00:01:00.000Z', + createdBy: 'alice', + }, + ], + }), + ); + + expect(document).toEqual({ + version: 1, + migration: { legacySha256: 'a'.repeat(64) }, + entries: [ + { + id: 'm-123456789abc', + text: 'Use staging', + createdAt: '2026-07-14T00:00:00.000Z', + updatedAt: '2026-07-14T00:01:00.000Z', + createdBy: 'alice', + }, + ], + }); + expect(CHANNEL_MEMORY_ID_RE.test(document.entries[0].id)).toBe(true); + }); + + it.each([ + ['missing entries', { version: 1 }], + ['entries is not an array', { version: 1, entries: {} }], + [ + 'empty text', + { version: 1, entries: [{ id: 'm-123456789abc', text: ' ' }] }, + ], + [ + 'oversized text', + { + version: 1, + entries: [ + { + id: 'm-123456789abc', + text: 'x'.repeat(MAX_CHANNEL_MEMORY_ENTRY_CODE_POINTS + 1), + }, + ], + }, + ], + [ + 'duplicate ids', + { + version: 1, + entries: [ + { id: 'm-123456789abc', text: 'x' }, + { id: 'm-123456789abc', text: 'y' }, + ], + }, + ], + [ + 'invalid optional fields', + { + version: 1, + migration: { legacySha256: 'A'.repeat(64) }, + entries: [ + { + id: 'm-123456789abc', + text: 'x', + createdAt: null, + }, + ], + }, + ], + ])('rejects %s', (_name, value) => { + expect(() => parseChannelMemoryDocument(JSON.stringify(value))).toThrow( + 'Invalid channel memory', + ); + }); + + it('rejects duplicate JSON object keys', () => { + expect(() => + parseChannelMemoryDocument('{"version":1,"entries":[],"entries":[]}'), + ).toThrow('Invalid channel memory document'); + }); + + it.each([ + ['top-level', { version: 1, entries: [], futureMetadata: 'preserve me' }], + [ + 'migration', + { + version: 1, + migration: { + legacySha256: 'a'.repeat(64), + futureMetadata: 'preserve me', + }, + entries: [], + }, + ], + [ + 'entry', + { + version: 1, + entries: [ + { + id: 'm-123456789abc', + text: 'Use staging', + futureMetadata: 'preserve me', + }, + ], + }, + ], + ])('rejects unknown %s keys', (_level, value) => { + expect(() => parseChannelMemoryDocument(JSON.stringify(value))).toThrow( + 'Invalid channel memory', + ); + }); + + it('rejects documents exceeding the entry limit', () => { + const entries = Array.from( + { length: MAX_CHANNEL_MEMORY_ENTRIES + 1 }, + (_, index) => ({ + id: `m-${index.toString(16).padStart(12, '0')}`, + text: 'x', + }), + ); + expect(() => + parseChannelMemoryDocument(JSON.stringify({ version: 1, entries })), + ).toThrow('maximum number of entries'); + }); + + it('counts astral Unicode text by code point', () => { + const astralCharacter = '\u{1f600}'; + const acceptedText = astralCharacter.repeat( + MAX_CHANNEL_MEMORY_ENTRY_CODE_POINTS, + ); + + expect( + parseChannelMemoryDocument( + JSON.stringify({ + version: 1, + entries: [{ id: 'm-123456789abc', text: acceptedText }], + }), + ).entries[0].text, + ).toBe(acceptedText); + expect(() => + parseChannelMemoryDocument( + JSON.stringify({ + version: 1, + entries: [ + { + id: 'm-123456789abc', + text: astralCharacter.repeat( + MAX_CHANNEL_MEMORY_ENTRY_CODE_POINTS + 1, + ), + }, + ], + }), + ), + ).toThrow('Invalid channel memory entry'); + }); + + it('converts legacy lines with stable ids and a migration hash', () => { + const raw = Buffer.from('Use staging\n\n use STAGING \nRun tests\n'); + const first = parseLegacyChannelMemory(raw); + const second = parseLegacyChannelMemory(raw); + + expect(first).toEqual(second); + expect(first.entries).toHaveLength(2); + expect(first.entries.map((entry) => entry.text)).toEqual([ + 'Use staging', + 'Run tests', + ]); + expect(first.entries[0].id).toBe('m-5c1888e97dc2'); + expect( + first.entries.every((entry) => CHANNEL_MEMORY_ID_RE.test(entry.id)), + ).toBe(true); + expect(first.entries.every((entry) => !('createdAt' in entry))).toBe(true); + expect(first.migration?.legacySha256).toMatch(/^[a-f0-9]{64}$/u); + }); + + it('preserves surrounding whitespace in legacy entry text', () => { + const document = parseLegacyChannelMemory( + Buffer.from(' Keep surrounding whitespace \n'), + ); + + expect(document.entries[0].text).toBe(' Keep surrounding whitespace '); + }); + + it('rejects invalid UTF-8 in legacy bytes', () => { + expect(() => parseLegacyChannelMemory(Buffer.from([0xff]))).toThrow(); + }); + + it('creates a timestamped channel memory entry', () => { + expect( + createChannelMemoryEntry({ + text: ' Use staging ', + createdBy: 'alice', + now: '2026-07-14T00:00:00.000Z', + randomHex: 'abcdef012345', + }), + ).toEqual({ + id: 'm-abcdef012345', + text: 'Use staging', + createdAt: '2026-07-14T00:00:00.000Z', + updatedAt: '2026-07-14T00:00:00.000Z', + createdBy: 'alice', + }); + }); + + it('rejects random values that cannot form a channel memory id', () => { + expect(() => + createChannelMemoryEntry({ + text: 'Use staging', + now: '2026-07-14T00:00:00.000Z', + randomHex: 'ABCDEF012345', + }), + ).toThrow('randomHex'); + }); + + it('renders recall text without entry metadata', () => { + expect( + renderChannelMemoryRecall([ + { + id: 'm-abcdef012345', + text: 'Use staging', + createdBy: 'alice', + }, + { id: 'm-123456789abc', text: 'Run tests', updatedAt: 'now' }, + ]), + ).toBe('Use staging\nRun tests\n'); + expect(renderChannelMemoryRecall([])).toBe(''); + }); + + it('serializes a document as stable pretty JSON', () => { + expect(serializeChannelMemoryDocument({ version: 1, entries: [] })).toBe( + '{\n "version": 1,\n "entries": []\n}\n', + ); + }); + + it('does not silently discard unknown keys during serialization', () => { + const document = { + version: 1 as const, + entries: [], + futureMetadata: 'preserve me', + }; + + expect(() => serializeChannelMemoryDocument(document)).toThrow( + 'Invalid channel memory document', + ); + }); +}); diff --git a/packages/core/src/memory/channel-memory-document.ts b/packages/core/src/memory/channel-memory-document.ts new file mode 100644 index 0000000000..2f86a3a17e --- /dev/null +++ b/packages/core/src/memory/channel-memory-document.ts @@ -0,0 +1,373 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { createHash } from 'node:crypto'; + +export const CHANNEL_MEMORY_DOCUMENT_VERSION = 1; +export const MAX_CHANNEL_MEMORY_ENTRIES = 500; +export const MAX_CHANNEL_MEMORY_ENTRIES_PER_REQUEST = 10; +export const MAX_CHANNEL_MEMORY_ENTRY_CODE_POINTS = 2_000; +export const CHANNEL_MEMORY_ID_RE = /^m-[a-f0-9]{12}$/u; + +export interface ChannelMemoryEntry { + id: string; + text: string; + createdAt?: string; + updatedAt?: string; + createdBy?: string; +} + +export interface ChannelMemoryDocument { + version: 1; + migration?: { legacySha256: string }; + entries: ChannelMemoryEntry[]; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function invalidDocument(message = 'Invalid channel memory document'): Error { + return new Error(message); +} + +function validateKeys( + value: Record, + allowedKeys: readonly string[], + message?: string, +): void { + const allowed = new Set(allowedKeys); + if (Object.keys(value).some((key) => !allowed.has(key))) { + throw invalidDocument(message); + } +} + +function validateEntry(value: unknown): ChannelMemoryEntry { + if (!isRecord(value)) { + throw invalidDocument('Invalid channel memory entry'); + } + validateKeys( + value, + ['id', 'text', 'createdAt', 'updatedAt', 'createdBy'], + 'Invalid channel memory entry', + ); + + const { id, text } = value; + if ( + typeof id !== 'string' || + !CHANNEL_MEMORY_ID_RE.test(id) || + typeof text !== 'string' || + text.trim().length === 0 || + Array.from(text).length > MAX_CHANNEL_MEMORY_ENTRY_CODE_POINTS + ) { + throw invalidDocument('Invalid channel memory entry'); + } + + const entry: ChannelMemoryEntry = { id, text }; + for (const key of ['createdAt', 'updatedAt', 'createdBy'] as const) { + if (key in value) { + if (typeof value[key] !== 'string') { + throw invalidDocument('Invalid channel memory entry'); + } + entry[key] = value[key]; + } + } + return entry; +} + +function parseDocumentValue(value: unknown): ChannelMemoryDocument { + if (!isRecord(value)) { + throw invalidDocument(); + } + validateKeys(value, ['version', 'migration', 'entries']); + if (!('version' in value) || typeof value['version'] !== 'number') { + throw invalidDocument(); + } + if (value['version'] !== CHANNEL_MEMORY_DOCUMENT_VERSION) { + throw invalidDocument('Unsupported channel memory version'); + } + if (!Array.isArray(value['entries'])) { + throw invalidDocument(); + } + if (value['entries'].length > MAX_CHANNEL_MEMORY_ENTRIES) { + throw invalidDocument('Channel memory exceeds maximum number of entries'); + } + + const ids = new Set(); + const entries = value['entries'].map((value) => { + const entry = validateEntry(value); + if (ids.has(entry.id)) { + throw invalidDocument('Invalid channel memory entry'); + } + ids.add(entry.id); + return entry; + }); + + let migration: ChannelMemoryDocument['migration']; + if ('migration' in value) { + if ( + !isRecord(value['migration']) || + typeof value['migration']['legacySha256'] !== 'string' || + !/^[a-f0-9]{64}$/u.test(value['migration']['legacySha256']) + ) { + throw invalidDocument(); + } + validateKeys(value['migration'], ['legacySha256']); + migration = { legacySha256: value['migration']['legacySha256'] }; + } + + return migration + ? { version: 1, migration, entries } + : { version: 1, entries }; +} + +function parseJson(raw: string): unknown { + return new JsonParser(raw).parse(); +} + +class JsonParser { + private index = 0; + + constructor(private readonly input: string) {} + + parse(): unknown { + const value = this.parseValue(); + this.skipWhitespace(); + if (this.index !== this.input.length) { + throw new Error('Unexpected trailing JSON'); + } + return value; + } + + private parseValue(): unknown { + this.skipWhitespace(); + const character = this.input[this.index]; + if (character === '{') { + return this.parseObject(); + } + if (character === '[') { + return this.parseArray(); + } + if (character === '"') { + return this.parseString(); + } + if (character === '-' || /\d/u.test(character ?? '')) { + return this.parseNumber(); + } + for (const [literal, value] of [ + ['true', true], + ['false', false], + ['null', null], + ] as const) { + if (this.input.startsWith(literal, this.index)) { + this.index += literal.length; + return value; + } + } + throw new Error('Invalid JSON value'); + } + + private parseObject(): Record { + this.index++; + const object: Record = Object.create(null) as Record< + string, + unknown + >; + const keys = new Set(); + this.skipWhitespace(); + if (this.input[this.index] === '}') { + this.index++; + return object; + } + + while (true) { + this.skipWhitespace(); + if (this.input[this.index] !== '"') { + throw new Error('Invalid JSON object key'); + } + const key = this.parseString(); + if (keys.has(key)) { + throw new Error('Duplicate JSON object key'); + } + keys.add(key); + this.skipWhitespace(); + if (this.input[this.index] !== ':') { + throw new Error('Invalid JSON object separator'); + } + this.index++; + Object.defineProperty(object, key, { + configurable: true, + enumerable: true, + value: this.parseValue(), + writable: true, + }); + this.skipWhitespace(); + if (this.input[this.index] === '}') { + this.index++; + return object; + } + if (this.input[this.index] !== ',') { + throw new Error('Invalid JSON object delimiter'); + } + this.index++; + } + } + + private parseArray(): unknown[] { + this.index++; + const array: unknown[] = []; + this.skipWhitespace(); + if (this.input[this.index] === ']') { + this.index++; + return array; + } + + while (true) { + array.push(this.parseValue()); + this.skipWhitespace(); + if (this.input[this.index] === ']') { + this.index++; + return array; + } + if (this.input[this.index] !== ',') { + throw new Error('Invalid JSON array delimiter'); + } + this.index++; + } + } + + private parseString(): string { + const start = this.index; + this.index++; + while (this.index < this.input.length) { + const character = this.input[this.index]; + if (character === '\\') { + this.index += 2; + continue; + } + if (character === '"') { + this.index++; + const value = JSON.parse( + this.input.slice(start, this.index), + ) as unknown; + if (typeof value !== 'string') { + throw new Error('Invalid JSON string'); + } + return value; + } + if (character < ' ') { + throw new Error('Invalid JSON string'); + } + this.index++; + } + throw new Error('Unterminated JSON string'); + } + + private parseNumber(): number { + const match = this.input + .slice(this.index) + .match(/^-?(?:0|[1-9]\d*)(?:\.\d+)?(?:[eE][+-]?\d+)?/u); + if (!match) { + throw new Error('Invalid JSON number'); + } + this.index += match[0].length; + return Number(match[0]); + } + + private skipWhitespace(): void { + while (/[ \t\r\n]/u.test(this.input[this.index] ?? '')) { + this.index++; + } + } +} + +export function normalizeChannelMemoryText(text: string): string { + return text.normalize('NFKC').trim().replace(/\s+/gu, ' ').toLowerCase(); +} + +export function parseChannelMemoryDocument(raw: string): ChannelMemoryDocument { + if (typeof raw !== 'string') { + throw invalidDocument(); + } + let value: unknown; + try { + value = parseJson(raw); + } catch { + throw invalidDocument(); + } + return parseDocumentValue(value); +} + +export function parseLegacyChannelMemory(raw: Buffer): ChannelMemoryDocument { + const entries: ChannelMemoryEntry[] = []; + const normalizedTexts = new Set(); + const ids = new Set(); + const decoded = new TextDecoder('utf-8', { fatal: true }).decode(raw); + + for (const [sourceLineIndex, line] of decoded + .split(/\r\n|\n|\r/u) + .entries()) { + const normalizedText = normalizeChannelMemoryText(line); + if (!normalizedText || normalizedTexts.has(normalizedText)) { + continue; + } + normalizedTexts.add(normalizedText); + + const digest = createHash('sha256') + .update(`${normalizedText}\0${sourceLineIndex}`) + .digest('hex'); + const id = `m-${digest.slice(0, 12)}`; + if (ids.has(id)) { + throw new Error('Channel memory legacy ID collision'); + } + ids.add(id); + entries.push(validateEntry({ id, text: line })); + if (entries.length > MAX_CHANNEL_MEMORY_ENTRIES) { + throw new Error('Channel memory exceeds maximum number of entries'); + } + } + + return { + version: 1, + migration: { + legacySha256: createHash('sha256').update(raw).digest('hex'), + }, + entries, + }; +} + +export function createChannelMemoryEntry(input: { + text: string; + createdBy?: string; + now: string; + randomHex: string; +}): ChannelMemoryEntry { + if (!/^[a-f0-9]{12}$/u.test(input.randomHex)) { + throw new Error('Invalid randomHex for channel memory entry'); + } + const text = input.text.trim(); + const entry = validateEntry({ id: `m-${input.randomHex}`, text }); + entry.createdAt = input.now; + entry.updatedAt = input.now; + if (input.createdBy !== undefined) { + entry.createdBy = input.createdBy; + } + return entry; +} + +export function renderChannelMemoryRecall( + entries: readonly ChannelMemoryEntry[], +): string { + return entries.length === 0 + ? '' + : `${entries.map((entry) => entry.text).join('\n')}\n`; +} + +export function serializeChannelMemoryDocument( + document: ChannelMemoryDocument, +): string { + return `${JSON.stringify(parseDocumentValue(document), null, 2)}\n`; +} diff --git a/packages/core/src/memory/channel-memory.test.ts b/packages/core/src/memory/channel-memory.test.ts index c5f7fbba04..744174e081 100644 --- a/packages/core/src/memory/channel-memory.test.ts +++ b/packages/core/src/memory/channel-memory.test.ts @@ -1,34 +1,169 @@ /** * @license - * Copyright 2025 Google LLC + * Copyright 2026 Qwen Team * SPDX-License-Identifier: Apache-2.0 */ +import { createHash } from 'node:crypto'; import * as fs from 'node:fs'; import * as os from 'node:os'; import * as path from 'node:path'; import lockfile from 'proper-lockfile'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { + addChannelMemoryEntries, appendChannelMemory, CHANNEL_MEMORY_FILE_NAME, clearChannelMemory, getChannelMemoryFilePath, + getLegacyChannelMemoryFilePath, + listChannelMemoryEntries, MAX_CHANNEL_MEMORY_BYTES, readChannelMemory, + removeChannelMemoryEntries, type ChannelMemoryTarget, + updateChannelMemoryEntry, } from './channel-memory.js'; +import { + parseChannelMemoryDocument, + parseLegacyChannelMemory, + serializeChannelMemoryDocument, +} from './channel-memory-document.js'; + +interface ReadRace { + jsonPath: string; + legacyPath: string; + jsonRead: () => void; + waitToReadLegacy: () => Promise; + jsonIntercepted: boolean; + legacyIntercepted: boolean; +} + +const fsFailure = vi.hoisted(() => ({ + tempSync: false, + tempBytesAtSync: 0, + rename: false, + legacyUnlinkPath: undefined as string | undefined, + readErrorPath: undefined as string | undefined, + readRace: undefined as ReadRace | undefined, + legacyAppendAfterRename: undefined as + | { path: string; text: string } + | undefined, +})); + +const lockObservation = vi.hoisted(() => ({ + path: undefined as string | undefined, + attempted: undefined as (() => void) | undefined, +})); + +vi.mock('proper-lockfile', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + default: { + ...actual, + async lock(...args: Parameters) { + if (String(args[0]) === lockObservation.path) { + lockObservation.attempted?.(); + } + return actual.lock(...args); + }, + }, + }; +}); + +vi.mock('node:fs/promises', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + async open(...args: Parameters) { + const handle = await actual.open(...args); + if (fsFailure.tempSync && String(args[0]).endsWith('.tmp')) { + return new Proxy(handle, { + get(target, property) { + if (property === 'sync') { + return async () => { + fsFailure.tempBytesAtSync = (await target.stat()).size; + throw new Error('temp sync failed'); + }; + } + const value = Reflect.get(target, property, target) as unknown; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + } + return handle; + }, + async readFile(...args: Parameters) { + const race = fsFailure.readRace; + const filePath = String(args[0]); + if (filePath === fsFailure.readErrorPath) { + throw Object.assign(new Error('read failed'), { code: 'EIO' }); + } + if ( + race !== undefined && + filePath === race.jsonPath && + !race.jsonIntercepted + ) { + race.jsonIntercepted = true; + race.jsonRead(); + throw Object.assign(new Error('missing'), { code: 'ENOENT' }); + } + if ( + race !== undefined && + filePath === race.legacyPath && + !race.legacyIntercepted + ) { + race.legacyIntercepted = true; + await race.waitToReadLegacy(); + } + return actual.readFile(...args); + }, + async rename(...args: Parameters) { + if (fsFailure.rename) { + throw new Error('rename failed'); + } + await actual.rename(...args); + const append = fsFailure.legacyAppendAfterRename; + if (append !== undefined) { + fsFailure.legacyAppendAfterRename = undefined; + await actual.appendFile(append.path, append.text); + } + }, + async unlink(...args: Parameters) { + if (String(args[0]) === fsFailure.legacyUnlinkPath) { + throw new Error('legacy unlink failed'); + } + return actual.unlink(...args); + }, + }; +}); describe('channel memory', () => { const originalQwenHome = process.env['QWEN_HOME']; let qwenHome: string; + const target: ChannelMemoryTarget = { + channelName: 'prod', + chatId: 'chat-1', + }; + beforeEach(() => { qwenHome = fs.mkdtempSync(path.join(os.tmpdir(), 'qwen-channel-memory-')); process.env['QWEN_HOME'] = qwenHome; }); afterEach(() => { + fsFailure.tempSync = false; + fsFailure.tempBytesAtSync = 0; + fsFailure.rename = false; + fsFailure.legacyUnlinkPath = undefined; + fsFailure.readErrorPath = undefined; + fsFailure.readRace = undefined; + fsFailure.legacyAppendAfterRename = undefined; + lockObservation.path = undefined; + lockObservation.attempted = undefined; + vi.restoreAllMocks(); if (originalQwenHome === undefined) { delete process.env['QWEN_HOME']; } else { @@ -37,16 +172,38 @@ describe('channel memory', () => { fs.rmSync(qwenHome, { recursive: true, force: true }); }); - it('returns a path under QWEN_HOME ending with CHANNEL.md', () => { - const filePath = getChannelMemoryFilePath({ - channelName: 'prod', - chatId: 'chat-1', + function writeLegacy(text: string): string { + const legacyPath = getLegacyChannelMemoryFilePath(target); + fs.mkdirSync(path.dirname(legacyPath), { recursive: true }); + fs.writeFileSync(legacyPath, text); + return legacyPath; + } + + function writeJson(raw: string): string { + const filePath = getChannelMemoryFilePath(target); + fs.mkdirSync(path.dirname(filePath), { recursive: true }); + fs.writeFileSync(filePath, raw); + return filePath; + } + + function deferred(): { promise: Promise; resolve: () => void } { + let resolve = () => {}; + const promise = new Promise((done) => { + resolve = done; }); + return { promise, resolve }; + } + + it('uses JSON for canonical storage and Markdown for legacy storage', () => { + const filePath = getChannelMemoryFilePath(target); + const legacyPath = getLegacyChannelMemoryFilePath(target); expect(filePath.startsWith(qwenHome + path.sep)).toBe(true); expect(filePath.endsWith(path.join('', CHANNEL_MEMORY_FILE_NAME))).toBe( true, ); + expect(filePath.endsWith('CHANNEL.json')).toBe(true); + expect(legacyPath.endsWith('CHANNEL.md')).toBe(true); }); it('keeps channel names and chat/thread identifiers safe', () => { @@ -79,10 +236,10 @@ describe('channel memory', () => { channelName, chatId: 'chat-1', }); - const relativePath = path.relative(qwenHome, filePath); - const relativeSegments = relativePath.split(path.sep); + const relativeSegments = path + .relative(qwenHome, filePath) + .split(path.sep); - expect(filePath.startsWith(qwenHome + path.sep)).toBe(true); expect(relativeSegments).not.toContain('.'); expect(relativeSegments).not.toContain('..'); expect(relativeSegments[0]).toBe('channels'); @@ -91,152 +248,619 @@ describe('channel memory', () => { }, ); - it('uses different paths for colliding sanitized channel names', () => { - const first = getChannelMemoryFilePath({ - channelName: 'ops/alerts', - chatId: 'chat-1', - }); - const second = getChannelMemoryFilePath({ - channelName: 'ops alerts', - chatId: 'chat-1', - }); - - expect(first).not.toBe(second); - }); - - it('uses different paths for different thread ids', () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; - + it('uses different paths for colliding sanitized channel names and threads', () => { + expect( + getChannelMemoryFilePath({ channelName: 'ops/alerts', chatId: 'chat-1' }), + ).not.toBe( + getChannelMemoryFilePath({ channelName: 'ops alerts', chatId: 'chat-1' }), + ); expect( getChannelMemoryFilePath({ ...target, threadId: 'thread-1' }), ).not.toBe(getChannelMemoryFilePath({ ...target, threadId: 'thread-2' })); }); - it('appends entries and reads the exact content', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; - - await appendChannelMemory(target, 'Use staging cluster by default.'); - await appendChannelMemory(target, 'Ask before running deploy commands.'); + it('renders JSON entries through the compatibility read API', async () => { + writeJson( + serializeChannelMemoryDocument({ + version: 1, + entries: [ + { id: 'm-111111111111', text: 'Use staging' }, + { id: 'm-222222222222', text: 'Run tests' }, + ], + }), + ); await expect(readChannelMemory(target)).resolves.toBe( - 'Use staging cluster by default.\nAsk before running deploy commands.\n', + 'Use staging\nRun tests\n', ); }); - it('does not create memory for whitespace-only appends', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', + it('lists deterministic legacy entries without creating JSON', async () => { + writeLegacy('Use staging\nUse staging\n Run tests \n'); + + const entries = await listChannelMemoryEntries(target); + + expect(entries.map((entry) => entry.text)).toEqual([ + 'Use staging', + ' Run tests ', + ]); + expect(entries.map((entry) => entry.id)).toEqual([ + 'm-5c1888e97dc2', + 'm-477e65662a6b', + ]); + expect(fs.existsSync(getChannelMemoryFilePath(target))).toBe(false); + }); + + it('migrates legacy content on add and cleans up the legacy file after commit', async () => { + const legacyPath = writeLegacy('Use staging\n'); + + const result = await addChannelMemoryEntries( + target, + ['Run tests'], + 'alice', + ); + + expect(result.added).toHaveLength(1); + expect(result.added[0].createdBy).toBe('alice'); + expect(fs.existsSync(getChannelMemoryFilePath(target))).toBe(true); + expect(fs.existsSync(legacyPath)).toBe(false); + await expect(listChannelMemoryEntries(target)).resolves.toMatchObject([ + { text: 'Use staging' }, + { text: 'Run tests' }, + ]); + }); + + it('waits for an old worker lock and migrates its final append', async () => { + const legacyPath = writeLegacy('Use staging\n'); + let releaseOldWorker = await lockfile.lock(legacyPath, { + realpath: false, + stale: 5000, + }); + const legacyLockAttempted = deferred(); + lockObservation.path = legacyPath; + lockObservation.attempted = legacyLockAttempted.resolve; + + const migration = addChannelMemoryEntries(target, ['Run tests']); + try { + const first = await Promise.race([ + legacyLockAttempted.promise.then(() => 'legacy-lock'), + migration.then(() => 'migration-completed'), + ]); + expect(first).toBe('legacy-lock'); + + fs.appendFileSync(legacyPath, 'Old worker append\n'); + await releaseOldWorker(); + releaseOldWorker = async () => {}; + + await expect(migration).resolves.toMatchObject({ changed: true }); + await expect(readChannelMemory(target)).resolves.toBe( + 'Use staging\nOld worker append\nRun tests\n', + ); + expect(fs.existsSync(legacyPath)).toBe(false); + } finally { + await releaseOldWorker(); + } + }); + + it('does not delete legacy bytes changed after canonical commit', async () => { + const legacyPath = writeLegacy('Use staging\n'); + fsFailure.legacyAppendAfterRename = { + path: legacyPath, + text: 'Late append\n', }; - const result = await appendChannelMemory(target, ' \n\t '); + await expect( + addChannelMemoryEntries(target, ['Run tests']), + ).resolves.toMatchObject({ changed: true }); - expect(result).toEqual({ + expect(fs.readFileSync(legacyPath, 'utf8')).toBe( + 'Use staging\nLate append\n', + ); + expect(fs.existsSync(getChannelMemoryFilePath(target))).toBe(true); + }); + + it('skips normalized duplicate additions and returns their existing IDs', async () => { + const first = await addChannelMemoryEntries( + target, + ['Use staging'], + 'alice', + ); + const duplicate = await addChannelMemoryEntries( + target, + [' use STAGING '], + 'alice', + ); + + expect(duplicate).toEqual({ + changed: false, + filePath: getChannelMemoryFilePath(target), + added: [], + duplicateIds: [first.added[0].id], + }); + }); + + it('keeps append as a compatibility wrapper', async () => { + await expect(appendChannelMemory(target, 'Use staging')).resolves.toEqual({ + changed: true, + filePath: getChannelMemoryFilePath(target), + }); + await expect(appendChannelMemory(target, ' use STAGING ')).resolves.toEqual( + { + changed: false, + filePath: getChannelMemoryFilePath(target), + }, + ); + await expect(readChannelMemory(target)).resolves.toBe('Use staging\n'); + }); + + it('does not create memory for whitespace-only appends', async () => { + await expect(appendChannelMemory(target, ' \n\t ')).resolves.toEqual({ changed: false, filePath: getChannelMemoryFilePath(target), }); await expect(readChannelMemory(target)).resolves.toBe(''); }); - it('clears memory when present', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; + it('updates only text and updatedAt while preserving identity metadata', async () => { + const [entry] = ( + await addChannelMemoryEntries(target, ['Use staging'], 'alice') + ).added; + await new Promise((resolve) => setTimeout(resolve, 1)); + + const result = await updateChannelMemoryEntry(target, { + id: entry.id, + text: 'Use production', + expectedText: 'Use staging', + }); + + expect(result.changed).toBe(true); + expect(result.entry).toMatchObject({ + id: entry.id, + text: 'Use production', + createdAt: entry.createdAt, + createdBy: 'alice', + }); + expect(result.entry?.updatedAt).not.toBe(entry.updatedAt); + }); + + it('rejects updates that duplicate another entry after normalization', async () => { + const [first, second] = ( + await addChannelMemoryEntries(target, ['Use staging', 'Run tests']) + ).added; + + await expect( + updateChannelMemoryEntry(target, { + id: second.id, + text: ' use STAGING ', + }), + ).rejects.toThrow('Channel memory entry already exists'); + + await expect(listChannelMemoryEntries(target)).resolves.toMatchObject([ + { id: first.id, text: 'Use staging' }, + { id: second.id, text: 'Run tests' }, + ]); + }); + + it('returns no change for missing update IDs', async () => { + await expect( + updateChannelMemoryEntry(target, { + id: 'm-111111111111', + text: 'Use prod', + }), + ).resolves.toEqual({ + changed: false, + filePath: getChannelMemoryFilePath(target), + }); + }); + + it('rejects update CAS when the entry was deleted', async () => { + const [entry] = (await addChannelMemoryEntries(target, ['Use staging'])) + .added; + await removeChannelMemoryEntries(target, { ids: [entry.id] }); + + await expect( + updateChannelMemoryEntry(target, { + id: entry.id, + text: 'Use production', + expectedText: 'Use staging', + }), + ).rejects.toThrow('Channel memory entry changed'); + }); + + it('rejects remove CAS when any expected entry was deleted', async () => { + const [first, second] = ( + await addChannelMemoryEntries(target, ['Use staging', 'Run tests']) + ).added; + await removeChannelMemoryEntries(target, { ids: [first.id] }); + + await expect( + removeChannelMemoryEntries(target, { + ids: [first.id, second.id], + expectedTextById: { + [first.id]: first.text, + [second.id]: second.text, + }, + }), + ).rejects.toThrow('Channel memory entry changed'); + await expect(listChannelMemoryEntries(target)).resolves.toEqual([second]); + }); + + it('rejects stale update and remove compare-and-swap requests', async () => { + const [entry] = (await addChannelMemoryEntries(target, ['Use staging'])) + .added; + + await expect( + updateChannelMemoryEntry(target, { + id: entry.id, + text: 'Use production', + expectedText: 'stale text', + }), + ).rejects.toThrow('Channel memory entry changed'); + await expect( + removeChannelMemoryEntries(target, { + ids: [entry.id], + expectedTextById: { [entry.id]: 'stale text' }, + }), + ).rejects.toThrow('Channel memory entry changed'); + }); + + it('removes requested IDs once and ignores missing IDs', async () => { + const [entry] = (await addChannelMemoryEntries(target, ['Use staging'])) + .added; + + const result = await removeChannelMemoryEntries(target, { + ids: [entry.id, entry.id, 'm-111111111111'], + }); + + expect(result.removed).toEqual([entry]); + expect(result.changed).toBe(true); + await expect( + removeChannelMemoryEntries(target, { ids: ['m-111111111111'] }), + ).resolves.toEqual({ + changed: false, + filePath: getChannelMemoryFilePath(target), + removed: [], + }); + }); + + it('clears entries while preserving migration metadata', async () => { + const legacyPath = writeLegacy('Use staging\n'); + await addChannelMemoryEntries(target, ['Run tests']); + const before = parseChannelMemoryDocument( + fs.readFileSync(getChannelMemoryFilePath(target), 'utf8'), + ); - await appendChannelMemory(target, 'Use staging cluster by default.'); await expect(clearChannelMemory(target)).resolves.toEqual({ changed: true, filePath: getChannelMemoryFilePath(target), }); - await expect(readChannelMemory(target)).resolves.toBe(''); + expect(fs.existsSync(legacyPath)).toBe(false); + expect( + parseChannelMemoryDocument( + fs.readFileSync(getChannelMemoryFilePath(target), 'utf8'), + ), + ).toEqual({ version: 1, migration: before.migration, entries: [] }); }); - it('reports no change when clearing missing memory', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; - + it('reports no change when clearing sources with no entries', async () => { + await expect(clearChannelMemory(target)).resolves.toEqual({ + changed: false, + filePath: getChannelMemoryFilePath(target), + }); + writeLegacy('\n\n'); await expect(clearChannelMemory(target)).resolves.toEqual({ changed: false, filePath: getChannelMemoryFilePath(target), }); }); - it('rejects writes over the maximum size', async () => { + it('rejects additions beyond request, entry, and text limits', async () => { await expect( - appendChannelMemory( - { channelName: 'prod', chatId: 'chat-1' }, - 'a'.repeat(MAX_CHANNEL_MEMORY_BYTES), + addChannelMemoryEntries( + target, + Array.from({ length: 11 }, () => 'entry'), + ), + ).rejects.toThrow(); + await expect( + addChannelMemoryEntries(target, ['a'.repeat(2_001)]), + ).rejects.toThrow('Invalid channel memory entry'); + + for (let index = 0; index < 50; index++) { + await addChannelMemoryEntries( + target, + Array.from( + { length: 10 }, + (_, offset) => `entry ${index * 10 + offset}`, + ), + ); + } + await expect(addChannelMemoryEntries(target, ['too many'])).rejects.toThrow( + 'Channel memory exceeds maximum number of entries', + ); + }); + + it('rejects oversized serialized JSON without creating canonical storage', async () => { + await expect( + addChannelMemoryEntries( + target, + ['entry'], + 'x'.repeat(MAX_CHANNEL_MEMORY_BYTES), ), ).rejects.toThrow('Channel memory exceeds maximum size'); + expect(fs.existsSync(getChannelMemoryFilePath(target))).toBe(false); }); - it('continues appends after a rejected append', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', + it('fails closed for malformed JSON and unsupported JSON versions', async () => { + writeLegacy('Use staging\n'); + writeJson('{'); + await expect(listChannelMemoryEntries(target)).rejects.toThrow( + 'Invalid channel memory document', + ); + await expect( + addChannelMemoryEntries(target, ['Run tests']), + ).rejects.toThrow('Invalid channel memory document'); + expect( + fs.readFileSync(getLegacyChannelMemoryFilePath(target), 'utf8'), + ).toBe('Use staging\n'); + + writeJson('{"version":2,"entries":[]}'); + await expect(listChannelMemoryEntries(target)).rejects.toThrow( + 'Unsupported channel memory version', + ); + }); + + it('rejects unknown JSON keys without modifying either source', async () => { + const legacyPath = writeLegacy('Use staging\n'); + const legacyBefore = fs.readFileSync(legacyPath); + const filePath = writeJson( + JSON.stringify({ + version: 1, + migration: { + legacySha256: createHash('sha256').update(legacyBefore).digest('hex'), + }, + entries: [{ id: 'm-111111111111', text: 'Use staging' }], + futureMetadata: 'preserve me', + }), + ); + const canonicalBefore = fs.readFileSync(filePath); + + const readResult = await readChannelMemory(target).then( + () => 'fulfilled', + () => 'rejected', + ); + const mutationResult = await addChannelMemoryEntries(target, [ + 'Run tests', + ]).then( + () => 'fulfilled', + () => 'rejected', + ); + + expect(readResult).toBe('rejected'); + expect(mutationResult).toBe('rejected'); + expect(fs.readFileSync(filePath)).toEqual(canonicalBefore); + expect(fs.readFileSync(legacyPath)).toEqual(legacyBefore); + }); + + it('rejects invalid UTF-8 JSON without modifying either source', async () => { + const legacyPath = writeLegacy('Use staging\n'); + const legacyBefore = fs.readFileSync(legacyPath); + const migrationHash = createHash('sha256') + .update(legacyBefore) + .digest('hex'); + const filePath = getChannelMemoryFilePath(target); + fs.writeFileSync( + filePath, + Buffer.concat([ + Buffer.from( + `{"version":1,"migration":{"legacySha256":"${migrationHash}"},"entries":[{"id":"m-111111111111","text":"`, + ), + Buffer.from([0xff]), + Buffer.from('"}]}'), + ]), + ); + const canonicalBefore = fs.readFileSync(filePath); + + await expect(readChannelMemory(target)).rejects.toThrow(); + await expect( + addChannelMemoryEntries(target, ['Run tests']), + ).rejects.toThrow(); + + expect(fs.readFileSync(filePath)).toEqual(canonicalBefore); + expect(fs.readFileSync(legacyPath)).toEqual(legacyBefore); + }); + + it('rejects invalid UTF-8 legacy bytes without migrating or modifying them', async () => { + const legacyPath = getLegacyChannelMemoryFilePath(target); + fs.mkdirSync(path.dirname(legacyPath), { recursive: true }); + fs.writeFileSync(legacyPath, Buffer.from([0xff])); + const legacyBefore = fs.readFileSync(legacyPath); + + await expect(readChannelMemory(target)).rejects.toThrow(); + await expect( + addChannelMemoryEntries(target, ['Run tests']), + ).rejects.toThrow(); + + expect(fs.readFileSync(legacyPath)).toEqual(legacyBefore); + expect(fs.existsSync(getChannelMemoryFilePath(target))).toBe(false); + }); + + it('rejects non-missing read errors without modifying either source', async () => { + const legacyPath = writeLegacy('Use staging\n'); + const legacy = fs.readFileSync(legacyPath); + const filePath = writeJson( + serializeChannelMemoryDocument(parseLegacyChannelMemory(legacy)), + ); + const canonicalBefore = fs.readFileSync(filePath); + const legacyBefore = fs.readFileSync(legacyPath); + fsFailure.readErrorPath = filePath; + + await expect(listChannelMemoryEntries(target)).rejects.toMatchObject({ + code: 'EIO', + message: 'read failed', + }); + await expect(readChannelMemory(target)).rejects.toMatchObject({ + code: 'EIO', + message: 'read failed', + }); + expect(fs.readFileSync(filePath)).toEqual(canonicalBefore); + expect(fs.readFileSync(legacyPath)).toEqual(legacyBefore); + }); + + it('accepts matching dual files and cleans up legacy only after a mutation', async () => { + const legacyPath = writeLegacy('Use staging\n'); + const legacy = fs.readFileSync(legacyPath); + writeJson(serializeChannelMemoryDocument(parseLegacyChannelMemory(legacy))); + + await expect(listChannelMemoryEntries(target)).resolves.toMatchObject([ + { text: 'Use staging' }, + ]); + expect(fs.existsSync(legacyPath)).toBe(true); + await addChannelMemoryEntries(target, ['Run tests']); + expect(fs.existsSync(legacyPath)).toBe(false); + }); + + it('re-reads canonical JSON across the first-migration rename race', async () => { + const legacyPath = writeLegacy('Use staging\n'); + const filePath = getChannelMemoryFilePath(target); + const jsonRead = deferred(); + const releaseLegacyRead = deferred(); + fsFailure.readRace = { + jsonPath: filePath, + legacyPath, + jsonRead: jsonRead.resolve, + waitToReadLegacy: () => releaseLegacyRead.promise, + jsonIntercepted: false, + legacyIntercepted: false, }; + const entriesPromise = listChannelMemoryEntries(target); + await jsonRead.promise; + await addChannelMemoryEntries(target, ['Run tests']); + releaseLegacyRead.resolve(); + + await expect(entriesPromise).resolves.toMatchObject([ + { text: 'Use staging' }, + { text: 'Run tests' }, + ]); + }); + + it('rejects divergent or unhashed dual files', async () => { + const legacyPath = writeLegacy('Use staging\n'); + const legacy = fs.readFileSync(legacyPath); + const document = parseLegacyChannelMemory(legacy); + writeJson( + serializeChannelMemoryDocument({ + ...document, + migration: { + legacySha256: createHash('sha256').update('different').digest('hex'), + }, + }), + ); + await expect(listChannelMemoryEntries(target)).rejects.toThrow( + 'Channel memory migration conflict', + ); + + writeJson(serializeChannelMemoryDocument({ version: 1, entries: [] })); + await expect( + addChannelMemoryEntries(target, ['Run tests']), + ).rejects.toThrow('Channel memory migration conflict'); + }); + + it('cleans a written temp file and recovers lock and queue after sync failure', async () => { + const legacyPath = writeLegacy('Use staging\n'); + const directory = path.dirname(legacyPath); + fsFailure.tempSync = true; + await expect( - appendChannelMemory(target, 'a'.repeat(MAX_CHANNEL_MEMORY_BYTES)), - ).rejects.toThrow('Channel memory exceeds maximum size'); - await appendChannelMemory(target, 'after failure'); + addChannelMemoryEntries(target, ['Run tests']), + ).rejects.toThrow('temp sync failed'); + expect(fsFailure.tempBytesAtSync).toBeGreaterThan(0); + expect( + fs.readdirSync(directory).filter((name) => name.endsWith('.tmp')), + ).toEqual([]); + expect(fs.existsSync(getChannelMemoryFilePath(target))).toBe(false); + expect(fs.readFileSync(legacyPath, 'utf8')).toBe('Use staging\n'); - await expect(readChannelMemory(target)).resolves.toBe('after failure\n'); + fsFailure.tempSync = false; + await expect( + addChannelMemoryEntries(target, ['after failure']), + ).resolves.toMatchObject({ + changed: true, + }); + await expect(readChannelMemory(target)).resolves.toBe( + 'Use staging\nafter failure\n', + ); }); - it('retries append when the file disappears before locking', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; + it('preserves the previous JSON when atomic rename fails', async () => { + await addChannelMemoryEntries(target, ['Use staging']); const filePath = getChannelMemoryFilePath(target); - const realLock = lockfile.lock.bind(lockfile); - let deletedBeforeLock = false; - const lockSpy = vi - .spyOn(lockfile, 'lock') - .mockImplementation(async (targetPath, options) => { - if (!deletedBeforeLock && targetPath === filePath) { - deletedBeforeLock = true; - fs.rmSync(filePath, { force: true }); - throw Object.assign(new Error('missing'), { code: 'ENOENT' }); - } - return realLock(targetPath, options); - }); + const previous = fs.readFileSync(filePath, 'utf8'); + fsFailure.rename = true; - try { - await expect(appendChannelMemory(target, 'after clear')).resolves.toEqual( - { - changed: true, - filePath, - }, - ); - await expect(readChannelMemory(target)).resolves.toBe('after clear\n'); - expect(lockSpy).toHaveBeenCalledTimes(2); - } finally { - lockSpy.mockRestore(); - } + await expect( + addChannelMemoryEntries(target, ['Run tests']), + ).rejects.toThrow('rename failed'); + expect(fs.readFileSync(filePath, 'utf8')).toBe(previous); }); - it('keeps concurrent appends within the maximum size', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; - const firstEntry = 'a'.repeat(MAX_CHANNEL_MEMORY_BYTES - 3); - await appendChannelMemory(target, firstEntry); + it('reports a committed migration when legacy cleanup fails and retries later', async () => { + const legacyPath = writeLegacy('Use staging\n'); + fsFailure.legacyUnlinkPath = legacyPath; + await expect( + addChannelMemoryEntries(target, ['Run tests']), + ).resolves.toMatchObject({ + changed: true, + added: [{ text: 'Run tests' }], + }); + expect(fs.existsSync(legacyPath)).toBe(true); + await expect(readChannelMemory(target)).resolves.toBe( + 'Use staging\nRun tests\n', + ); + + fsFailure.legacyUnlinkPath = undefined; + await addChannelMemoryEntries(target, ['Review diff']); + expect(fs.existsSync(legacyPath)).toBe(false); + }); + + it('serializes concurrent additions without losing entries', async () => { + const additions = await Promise.all( + Array.from({ length: 20 }, (_, index) => + addChannelMemoryEntries(target, [`entry ${index}`]), + ), + ); + const entries = await listChannelMemoryEntries(target); + + expect(entries).toHaveLength(20); + expect(new Set(entries.map((entry) => entry.id)).size).toBe(20); + expect( + new Set( + additions.flatMap((result) => result.added.map((entry) => entry.text)), + ), + ).toEqual( + new Set(Array.from({ length: 20 }, (_, index) => `entry ${index}`)), + ); + expect(() => + parseChannelMemoryDocument( + fs.readFileSync(getChannelMemoryFilePath(target), 'utf8'), + ), + ).not.toThrow(); + }); + + it('allows only one stale compare-and-swap operation to win', async () => { + const [entry] = (await addChannelMemoryEntries(target, ['Use staging'])) + .added; const results = await Promise.allSettled([ - appendChannelMemory(target, 'b'), - appendChannelMemory(target, 'c'), + updateChannelMemoryEntry(target, { + id: entry.id, + text: 'Use production', + expectedText: 'Use staging', + }), + removeChannelMemoryEntries(target, { + ids: [entry.id], + expectedTextById: { [entry.id]: 'Use staging' }, + }), ]); expect( @@ -245,34 +869,55 @@ describe('channel memory', () => { expect( results.filter((result) => result.status === 'rejected'), ).toHaveLength(1); - expect( - fs.statSync(getChannelMemoryFilePath(target)).size, - ).toBeLessThanOrEqual(MAX_CHANNEL_MEMORY_BYTES); }); - it('serializes clear after pending appends', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; + it('rejects an update CAS when a racing remove is queued first', async () => { + const [entry] = (await addChannelMemoryEntries(target, ['Use staging'])) + .added; + const results = await Promise.allSettled([ + removeChannelMemoryEntries(target, { + ids: [entry.id], + expectedTextById: { [entry.id]: 'Use staging' }, + }), + updateChannelMemoryEntry(target, { + id: entry.id, + text: 'Use production', + expectedText: 'Use staging', + }), + ]); - const appends = Array.from({ length: 20 }, (_, index) => - appendChannelMemory(target, `entry ${index}`), + expect(results[0].status).toBe('fulfilled'); + expect(results[1]).toMatchObject({ + status: 'rejected', + reason: new Error('Channel memory entry changed'), + }); + }); + + it('serializes clear racing additions and first migration racing another add', async () => { + writeLegacy('Use staging\n'); + await Promise.all([ + addChannelMemoryEntries(target, ['Run tests']), + addChannelMemoryEntries(target, ['Review diff']), + ]); + await expect(readChannelMemory(target)).resolves.toBe( + 'Use staging\nRun tests\nReview diff\n', ); - await Promise.all([...appends, clearChannelMemory(target)]); + await Promise.all([ + clearChannelMemory(target), + ...Array.from({ length: 10 }, (_, index) => + addChannelMemoryEntries(target, [`entry ${index}`]), + ), + ]); - await expect(readChannelMemory(target)).resolves.toBe(''); - }); - - it('reads oversized existing memory as empty', async () => { - const target: ChannelMemoryTarget = { - channelName: 'prod', - chatId: 'chat-1', - }; - const filePath = getChannelMemoryFilePath(target); - fs.mkdirSync(path.dirname(filePath), { recursive: true }); - fs.writeFileSync(filePath, Buffer.alloc(MAX_CHANNEL_MEMORY_BYTES + 1)); - - await expect(readChannelMemory(target)).resolves.toBe(''); + const entries = await listChannelMemoryEntries(target); + expect(entries.map((entry) => entry.text)).toEqual( + Array.from({ length: 10 }, (_, index) => `entry ${index}`), + ); + expect(new Set(entries.map((entry) => entry.id)).size).toBe(entries.length); + expect(() => + parseChannelMemoryDocument( + fs.readFileSync(getChannelMemoryFilePath(target), 'utf8'), + ), + ).not.toThrow(); }); }); diff --git a/packages/core/src/memory/channel-memory.ts b/packages/core/src/memory/channel-memory.ts index 14e7348b2f..e3eaeb5d10 100644 --- a/packages/core/src/memory/channel-memory.ts +++ b/packages/core/src/memory/channel-memory.ts @@ -1,14 +1,25 @@ /** * @license - * Copyright 2025 Google LLC + * Copyright 2026 Qwen Team * SPDX-License-Identifier: Apache-2.0 */ -import { createHash } from 'node:crypto'; +import { createHash, randomBytes } from 'node:crypto'; import * as fs from 'node:fs/promises'; import * as path from 'node:path'; import lockfile from 'proper-lockfile'; import { Storage } from '../config/storage.js'; +import { + createChannelMemoryEntry, + MAX_CHANNEL_MEMORY_ENTRIES_PER_REQUEST, + normalizeChannelMemoryText, + parseChannelMemoryDocument, + parseLegacyChannelMemory, + renderChannelMemoryRecall, + serializeChannelMemoryDocument, + type ChannelMemoryDocument, + type ChannelMemoryEntry, +} from './channel-memory-document.js'; export interface ChannelMemoryTarget { channelName: string; @@ -16,14 +27,31 @@ export interface ChannelMemoryTarget { threadId?: string; } -export interface ChannelMemoryWriteResult { +export interface ChannelMemoryMutationResult { changed: boolean; filePath: string; } -export const CHANNEL_MEMORY_FILE_NAME = 'CHANNEL.md'; +export interface AddChannelMemoryResult extends ChannelMemoryMutationResult { + added: ChannelMemoryEntry[]; + duplicateIds: string[]; +} + +export interface UpdateChannelMemoryResult extends ChannelMemoryMutationResult { + entry?: ChannelMemoryEntry; +} + +export interface RemoveChannelMemoryResult extends ChannelMemoryMutationResult { + removed: ChannelMemoryEntry[]; +} + +export type ChannelMemoryWriteResult = ChannelMemoryMutationResult; + +export const CHANNEL_MEMORY_FILE_NAME = 'CHANNEL.json'; +export const LEGACY_CHANNEL_MEMORY_FILE_NAME = 'CHANNEL.md'; export const MAX_CHANNEL_MEMORY_BYTES = 1024 * 1024; -const pendingAppends = new Map>(); + +const pendingMutations = new Map>(); const LOCK_OPTIONS: lockfile.LockOptions = { realpath: false, retries: { @@ -36,6 +64,17 @@ const LOCK_OPTIONS: lockfile.LockOptions = { stale: 5000, }; +interface LoadedChannelMemory { + document: ChannelMemoryDocument; + legacyBytes?: Buffer; + legacyHasEntries: boolean; +} + +interface Mutation { + changed: boolean; + result: T; +} + function isMissingFile(error: unknown): boolean { return (error as NodeJS.ErrnoException).code === 'ENOENT'; } @@ -48,6 +87,24 @@ async function releaseLock(release: () => Promise): Promise { } } +async function cleanupLegacyAfterCommit( + legacyPath: string, + expectedBytes: Buffer, +): Promise { + try { + const currentBytes = await fs.readFile(legacyPath); + if ( + legacyHash(currentBytes) !== legacyHash(expectedBytes) || + !currentBytes.equals(expectedBytes) + ) { + return; + } + await fs.unlink(legacyPath); + } catch { + // Canonical data is committed; a matching legacy file is safe to retry. + } +} + function safeChannelName(channelName: string): string { const slug = channelName.replace(/[^a-zA-Z0-9._-]/g, '_').slice(0, 20) || '_'; const hash = createHash('sha256') @@ -66,152 +123,384 @@ function hashedThreadPath(target: ChannelMemoryTarget): string { .slice(0, 32); } -export function getChannelMemoryFilePath(target: ChannelMemoryTarget): string { +function getChannelMemoryDirectory(target: ChannelMemoryTarget): string { return path.join( Storage.getGlobalQwenDir(), 'channels', 'memory', safeChannelName(target.channelName), hashedThreadPath(target), - CHANNEL_MEMORY_FILE_NAME, ); } -async function serializeAppend( - filePath: string, +export function getChannelMemoryFilePath(target: ChannelMemoryTarget): string { + return path.join(getChannelMemoryDirectory(target), CHANNEL_MEMORY_FILE_NAME); +} + +export function getLegacyChannelMemoryFilePath( + target: ChannelMemoryTarget, +): string { + return path.join( + getChannelMemoryDirectory(target), + LEGACY_CHANNEL_MEMORY_FILE_NAME, + ); +} + +async function serializeMutation( + directory: string, task: () => Promise, ): Promise { - const previous = pendingAppends.get(filePath) ?? Promise.resolve(); - let release: () => void = () => {}; + const previous = pendingMutations.get(directory) ?? Promise.resolve(); + let resolveCurrent: () => void = () => {}; const current = new Promise((resolve) => { - release = resolve; + resolveCurrent = resolve; }); const queued = previous.then( () => current, () => current, ); - pendingAppends.set(filePath, queued); + pendingMutations.set(directory, queued); await previous.catch(() => {}); try { return await task(); } finally { - release(); - if (pendingAppends.get(filePath) === queued) { - pendingAppends.delete(filePath); + resolveCurrent(); + if (pendingMutations.get(directory) === queued) { + pendingMutations.delete(directory); } } } +async function readFileIfExists(filePath: string): Promise { + try { + return await fs.readFile(filePath); + } catch (error) { + if (isMissingFile(error)) { + return undefined; + } + throw error; + } +} + +function legacyHash(legacyBytes: Buffer): string { + return createHash('sha256').update(legacyBytes).digest('hex'); +} + +async function lockLegacyIfExists( + legacyPath: string, +): Promise<(() => Promise) | undefined> { + try { + return await lockfile.lock(legacyPath, LOCK_OPTIONS); + } catch (error) { + if (isMissingFile(error)) { + return undefined; + } + throw error; + } +} + +function verifyDualFileState( + document: ChannelMemoryDocument, + legacyBytes: Buffer | undefined, +): void { + if ( + legacyBytes !== undefined && + document.migration?.legacySha256 !== legacyHash(legacyBytes) + ) { + throw new Error('Channel memory migration conflict'); + } +} + +async function loadChannelMemory( + target: ChannelMemoryTarget, +): Promise { + const filePath = getChannelMemoryFilePath(target); + const [initialJsonBytes, legacyBytes] = await Promise.all([ + readFileIfExists(filePath), + readFileIfExists(getLegacyChannelMemoryFilePath(target)), + ]); + const jsonBytes = + initialJsonBytes ?? + (legacyBytes === undefined ? await readFileIfExists(filePath) : undefined); + + if (jsonBytes !== undefined) { + if (jsonBytes.length > MAX_CHANNEL_MEMORY_BYTES) { + throw new Error('Channel memory exceeds maximum size'); + } + const document = parseChannelMemoryDocument( + new TextDecoder('utf-8', { fatal: true }).decode(jsonBytes), + ); + verifyDualFileState(document, legacyBytes); + return { + document, + legacyBytes, + legacyHasEntries: + legacyBytes !== undefined && + parseLegacyChannelMemory(legacyBytes).entries.length > 0, + }; + } + + if (legacyBytes !== undefined) { + const document = parseLegacyChannelMemory(legacyBytes); + return { + document, + legacyBytes, + legacyHasEntries: document.entries.length > 0, + }; + } + + return { document: { version: 1, entries: [] }, legacyHasEntries: false }; +} + +async function writeChannelMemory( + filePath: string, + serialized: string, +): Promise { + const tempPath = `${filePath}.${process.pid}.${randomBytes(6).toString('hex')}.tmp`; + let committed = false; + try { + const handle = await fs.open(tempPath, 'wx', 0o600); + try { + await handle.writeFile(serialized, 'utf8'); + await handle.sync(); + } finally { + await handle.close(); + } + await fs.rename(tempPath, filePath); + committed = true; + } finally { + if (!committed) { + await fs.unlink(tempPath).catch((error: unknown) => { + if (!isMissingFile(error)) { + throw error; + } + }); + } + } +} + +async function mutateChannelMemory( + target: ChannelMemoryTarget, + apply: ( + document: ChannelMemoryDocument, + sourceHasEntries: boolean, + ) => Mutation, +): Promise { + const filePath = getChannelMemoryFilePath(target); + const directory = path.dirname(filePath); + const legacyPath = getLegacyChannelMemoryFilePath(target); + + return serializeMutation(directory, async () => { + await fs.mkdir(directory, { recursive: true }); + const lockPath = path.join(directory, '.channel-memory.lock'); + const lockHandle = await fs.open(lockPath, 'a', 0o600); + await lockHandle.close(); + const release = await lockfile.lock(lockPath, LOCK_OPTIONS); + let releaseLegacy: (() => Promise) | undefined; + try { + releaseLegacy = await lockLegacyIfExists(legacyPath); + const loaded = await loadChannelMemory(target); + const mutation = apply( + loaded.document, + loaded.document.entries.length > 0 || loaded.legacyHasEntries, + ); + if (!mutation.changed) { + return mutation.result; + } + + const serialized = serializeChannelMemoryDocument(loaded.document); + if (Buffer.byteLength(serialized, 'utf8') > MAX_CHANNEL_MEMORY_BYTES) { + throw new Error('Channel memory exceeds maximum size'); + } + await writeChannelMemory(filePath, serialized); + if (loaded.legacyBytes !== undefined) { + await cleanupLegacyAfterCommit(legacyPath, loaded.legacyBytes); + } + return mutation.result; + } finally { + if (releaseLegacy !== undefined) { + await releaseLock(releaseLegacy); + } + await releaseLock(release); + } + }); +} + +export async function listChannelMemoryEntries( + target: ChannelMemoryTarget, +): Promise { + const { document } = await loadChannelMemory(target); + return document.entries; +} + +export async function addChannelMemoryEntries( + target: ChannelMemoryTarget, + texts: readonly string[], + createdBy?: string, +): Promise { + if (texts.length > MAX_CHANNEL_MEMORY_ENTRIES_PER_REQUEST) { + throw new Error('Channel memory accepts at most 10 entries per request'); + } + + const filePath = getChannelMemoryFilePath(target); + return mutateChannelMemory(target, (document) => { + const entriesByNormalizedText = new Map( + document.entries.map((entry) => [ + normalizeChannelMemoryText(entry.text), + entry, + ]), + ); + const ids = new Set(document.entries.map((entry) => entry.id)); + const added: ChannelMemoryEntry[] = []; + const duplicateIds: string[] = []; + + for (const text of texts) { + const normalizedText = normalizeChannelMemoryText(text); + if (!normalizedText) { + continue; + } + const duplicate = entriesByNormalizedText.get(normalizedText); + if (duplicate !== undefined) { + duplicateIds.push(duplicate.id); + continue; + } + + let randomHex: string; + do { + randomHex = randomBytes(6).toString('hex'); + } while (ids.has(`m-${randomHex}`)); + const entry = createChannelMemoryEntry({ + text, + createdBy, + now: new Date().toISOString(), + randomHex, + }); + ids.add(entry.id); + entriesByNormalizedText.set(normalizedText, entry); + document.entries.push(entry); + added.push(entry); + } + + return { + changed: added.length > 0, + result: { + changed: added.length > 0, + filePath, + added, + duplicateIds, + }, + }; + }); +} + +export async function updateChannelMemoryEntry( + target: ChannelMemoryTarget, + mutation: { id: string; text: string; expectedText?: string }, +): Promise { + const filePath = getChannelMemoryFilePath(target); + return mutateChannelMemory(target, (document) => { + const entry = document.entries.find( + (candidate) => candidate.id === mutation.id, + ); + if (entry === undefined) { + if (mutation.expectedText !== undefined) { + throw new Error('Channel memory entry changed'); + } + return { changed: false, result: { changed: false, filePath } }; + } + if ( + mutation.expectedText !== undefined && + entry.text !== mutation.expectedText + ) { + throw new Error('Channel memory entry changed'); + } + + const replacement = createChannelMemoryEntry({ + text: mutation.text, + now: new Date().toISOString(), + randomHex: '000000000000', + }); + if ( + document.entries.some( + (candidate) => + candidate.id !== entry.id && + normalizeChannelMemoryText(candidate.text) === + normalizeChannelMemoryText(replacement.text), + ) + ) { + throw new Error('Channel memory entry already exists'); + } + entry.text = replacement.text; + entry.updatedAt = replacement.updatedAt; + return { + changed: true, + result: { changed: true, filePath, entry: { ...entry } }, + }; + }); +} + +export async function removeChannelMemoryEntries( + target: ChannelMemoryTarget, + mutation: { + ids: readonly string[]; + expectedTextById?: Readonly>; + }, +): Promise { + const filePath = getChannelMemoryFilePath(target); + return mutateChannelMemory(target, (document) => { + const requestedIds = new Set(mutation.ids); + const entriesById = new Map( + document.entries.map((entry) => [entry.id, entry]), + ); + for (const id of requestedIds) { + const expectedText = mutation.expectedTextById?.[id]; + if ( + expectedText !== undefined && + entriesById.get(id)?.text !== expectedText + ) { + throw new Error('Channel memory entry changed'); + } + } + const removed = document.entries.filter((entry) => + requestedIds.has(entry.id), + ); + if (removed.length === 0) { + return { changed: false, result: { changed: false, filePath, removed } }; + } + document.entries = document.entries.filter( + (entry) => !requestedIds.has(entry.id), + ); + return { changed: true, result: { changed: true, filePath, removed } }; + }); +} + export async function readChannelMemory( target: ChannelMemoryTarget, ): Promise { - const filePath = getChannelMemoryFilePath(target); - return serializeAppend(filePath, async () => { - let size: number; - try { - size = (await fs.stat(filePath)).size; - } catch (error) { - if (isMissingFile(error)) { - return ''; - } - throw error; - } - if (size > MAX_CHANNEL_MEMORY_BYTES) { - process.stderr.write( - `[channel-memory] ${filePath} is ${size} bytes, exceeding ${MAX_CHANNEL_MEMORY_BYTES}; treating as empty\n`, - ); - return ''; - } - try { - return await fs.readFile(filePath, 'utf8'); - } catch (error) { - if (isMissingFile(error)) { - return ''; - } - throw error; - } - }); + return renderChannelMemoryRecall(await listChannelMemoryEntries(target)); } export async function appendChannelMemory( target: ChannelMemoryTarget, text: string, -): Promise { - const filePath = getChannelMemoryFilePath(target); - const entry = text.trim(); - if (!entry) { - return { changed: false, filePath }; - } - - return serializeAppend(filePath, async () => { - const appendBytes = Buffer.byteLength(`${entry}\n`, 'utf8'); - await fs.mkdir(path.dirname(filePath), { recursive: true }); - // proper-lockfile requires the target file to exist before locking it. - const initialHandle = await fs.open(filePath, 'a+'); - await initialHandle.close(); - let release: () => Promise; - try { - release = await lockfile.lock(filePath, LOCK_OPTIONS); - } catch (error) { - if (!isMissingFile(error)) { - throw error; - } - const retryHandle = await fs.open(filePath, 'a+'); - await retryHandle.close(); - release = await lockfile.lock(filePath, LOCK_OPTIONS); - } - try { - const handle = await fs.open(filePath, 'a+'); - try { - const existingSize = (await handle.stat()).size; - if (existingSize + appendBytes > MAX_CHANNEL_MEMORY_BYTES) { - throw new Error('Channel memory exceeds maximum size'); - } - await handle.appendFile(`${entry}\n`, 'utf8'); - } finally { - await handle.close(); - } - } finally { - await releaseLock(release); - } - return { changed: true, filePath }; - }); +): Promise { + const { changed, filePath } = await addChannelMemoryEntries(target, [text]); + return { changed, filePath }; } export async function clearChannelMemory( target: ChannelMemoryTarget, -): Promise { +): Promise { const filePath = getChannelMemoryFilePath(target); - return serializeAppend(filePath, async () => { - try { - await fs.access(filePath); - } catch (error) { - if (isMissingFile(error)) { - return { changed: false, filePath }; + return mutateChannelMemory( + target, + (document, sourceHasEntries) => { + if (!sourceHasEntries) { + return { changed: false, result: { changed: false, filePath } }; } - throw error; - } - - let release: () => Promise; - try { - release = await lockfile.lock(filePath, LOCK_OPTIONS); - } catch (error) { - if (isMissingFile(error)) { - return { changed: false, filePath }; - } - throw error; - } - try { - await fs.unlink(filePath); - return { changed: true, filePath }; - } catch (error) { - if (isMissingFile(error)) { - return { changed: false, filePath }; - } - throw error; - } finally { - await releaseLock(release); - } - }); + document.entries = []; + return { changed: true, result: { changed: true, filePath } }; + }, + ); }