fix(memory): preserve stream usage and bound portable history

Author: icecrasher321

apps/sim/executor/handlers/agent/memory.durability.test.ts

@@ -126,6 +126,29 @@ describe('optional Agent memory durability failures', () => {
     expect(mocks.append).toHaveBeenCalledOnce()
   })
 
+  it('propagates an append identity conflict instead of treating different content as saved', async () => {
+    const failure = new OrchestrationError('conflict', 'Memory append identity was already used')
+    mocks.append.mockRejectedValue(failure)
+    await expect(
+      new Memory().appendToMemory(
+        ctx,
+        inputs,
+        { role: 'user', content: 'changed question' },
+        options
+      )
+    ).rejects.toBe(failure)
+    expect(mocks.append).toHaveBeenCalledExactlyOnceWith({
+      principal: { kind: 'delegated' },
+      input: {
+        ...options,
+        workspaceId: ctx.workspaceId,
+        conversationId: inputs.conversationId,
+        data: { role: 'user', content: 'changed question' },
+        provenance: undefined,
+      },
+    })
+  })
+
   it('preserves authorization failures on reads and appends', async () => {
     const failure = new OrchestrationError('forbidden', 'Denied')
     mocks.prefix.mockRejectedValue(failure)

apps/sim/executor/handlers/agent/memory.test.ts

@@ -88,7 +88,7 @@ describe('Memory', () => {
       }
     )
 
-    it.each(['forbidden', 'unauthorized', 'validation'] as const)(
+    it.each(['forbidden', 'unauthorized', 'validation', 'conflict'] as const)(
       'propagates application %s failures even for optional rich history',
       async (code) => {
         const error = new OrchestrationError(code, 'Memory access refused')

apps/sim/executor/handlers/agent/memory.ts

@@ -95,7 +95,7 @@ function copyMemoryMessageMetadata(source: Message, target: Message): void {
 /** Optional durability tolerates storage-engine failures; application identity and projection failures propagate. */
 function isOptionalMemoryStorageFailure(error: unknown): boolean {
   if (error instanceof OrchestrationError)
-    return error.code === 'not_found' || error.code === 'conflict' || error.code === 'internal'
+    return error.code === 'not_found' || error.code === 'internal'
   const code = getPostgresErrorCode(error)
   return Boolean(
     (code &&

apps/sim/lib/memory/artifacts.test.ts

@@ -238,6 +238,35 @@ describe('encrypted memory artifacts', () => {
     expect(materializeLargeValueRef).not.toHaveBeenCalled()
   })
 
+  it('preserves the shared materializer unavailable-result contract without attempting decryption', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      {
+        key: ref.key,
+        size: 500,
+        workflowId: identity.workflowId,
+        executionId: identity.executionId,
+      },
+    ])
+    materializeLargeValueRef.mockResolvedValueOnce(undefined)
+    expect(await readMemoryArtifact({ ...scope, ref })).toBeUndefined()
+    expect(encryptionMockFns.mockDecryptSecret).not.toHaveBeenCalled()
+  })
+
+  it('does not swallow errors that escape the materialization boundary', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      {
+        key: ref.key,
+        size: 500,
+        workflowId: identity.workflowId,
+        executionId: identity.executionId,
+      },
+    ])
+    const error = new Error('Materialization access check failed')
+    materializeLargeValueRef.mockRejectedValueOnce(error)
+    await expect(readMemoryArtifact({ ...scope, ref })).rejects.toBe(error)
+    expect(encryptionMockFns.mockDecryptSecret).not.toHaveBeenCalled()
+  })
+
   it.each(['ciphertext', 'json'])(
     'treats corrupt %s as an unavailable artifact',
     async (failure) => {

apps/sim/lib/memory/conversation-store.postgres.test.ts

@@ -276,6 +276,50 @@ describe.skipIf(!databaseUrl)('conversation storage in Postgres', () => {
     expect(tail.provenance).toEqual(provenance)
   })
 
+  it('explicitly invalidates tracked legacy provenance after an untracked append', async () => {
+    await writeLegacyPrefix()
+    const untracked = { role: 'user', content: 'untracked append' }
+    await appendMemoryMessages({
+      workspaceId: identity.workspaceId,
+      key: identity.conversationId,
+      messages: [untracked],
+    })
+    const [stored] =
+      await connection!`SELECT data, secret_provenance_version FROM memory WHERE id = 'legacy-memory'`
+    expect(stored.secret_provenance_version).toBe(1)
+    expect(stored.data).toEqual([...prefix, untracked])
+    expect(
+      (
+        await connection!`SELECT content_hash, status, entries FROM memory_secret_provenance WHERE memory_id = 'legacy-memory'`
+      )[0]
+    ).toEqual({
+      content_hash: hashDurableSecretProvenanceValue(stored.data),
+      status: 'unknown',
+      entries: [],
+    })
+    const tracked = { role: 'assistant', content: 'later tracked append' }
+    await appendMemoryMessages({
+      workspaceId: identity.workspaceId,
+      key: identity.conversationId,
+      messages: [tracked],
+      provenance,
+    })
+    expect(
+      (
+        await connection!`SELECT content_hash, status FROM memory_secret_provenance WHERE memory_id = 'legacy-memory'`
+      )[0]
+    ).toEqual({
+      content_hash: hashDurableSecretProvenanceValue([...prefix, untracked, tracked]),
+      status: 'unknown',
+    })
+    await expect(
+      new Memory().fetchMemoryMessages({ workspaceId: identity.workspaceId } as ExecutionContext, {
+        memoryType: 'conversation',
+        conversationId: identity.conversationId,
+      })
+    ).rejects.toThrow('Memory content could not be safely projected')
+  })
+
   it('freezes the legacy prefix and keeps exchanges out of native/plain API history', async () => {
     await writeLegacyPrefix()
     const turn = await openAgentMemoryTurn(identity)

apps/sim/lib/memory/conversation-store.test.ts

@@ -1,7 +1,8 @@
 /** @vitest-environment node */
-import { memory, memoryItem } from '@sim/db/schema'
+import { memory, memoryItem, memorySecretProvenance } from '@sim/db/schema'
 import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { hashDurableSecretProvenanceValue } from '@/lib/execution/durable-secret-provenance'
 import {
   appendAgentMemoryMessage,
   appendMemoryMessages,
@@ -65,3 +66,51 @@ describe('bounded conversation item writes', () => {
     expect(get).not.toHaveBeenCalled()
   })
 })
+
+describe('untracked legacy appends', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('binds explicit unknown provenance to the updated JSON without declaring prior secrets public', async () => {
+    const prefix = [{ role: 'user', content: 'tracked content' }]
+    const message = { role: 'user', content: 'untracked append' }
+    const data = [...prefix, message]
+    queueTableRows(memory, [
+      { id: identity.memoryId, data: prefix, storageVersion: 1, secretProvenanceVersion: 1 },
+    ])
+    dbChainMockFns.returning
+      .mockResolvedValueOnce([{ id: identity.memoryId, data }])
+      .mockResolvedValueOnce([{ id: identity.memoryId }])
+    await writers.ordinary(message)
+    expect(dbChainMockFns.insert).toHaveBeenCalledWith(memorySecretProvenance)
+    expect(dbChainMockFns.values).toHaveBeenCalledWith(
+      expect.objectContaining({
+        memoryId: identity.memoryId,
+        contentHash: hashDurableSecretProvenanceValue(data),
+        status: 'unknown',
+        entries: [],
+      })
+    )
+    expect(dbChainMockFns.onConflictDoUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        set: expect.objectContaining({ secretProvenanceVersion: 1 }),
+      })
+    )
+  })
+
+  it('keeps wholly untracked legacy conversations on their existing compatibility path', async () => {
+    queueTableRows(memory, [
+      { id: identity.memoryId, data: [], storageVersion: 1, secretProvenanceVersion: null },
+    ])
+    dbChainMockFns.returning.mockResolvedValueOnce([{ id: identity.memoryId, data: [] }])
+    await writers.ordinary({ role: 'user', content: 'public' })
+    expect(dbChainMockFns.insert).not.toHaveBeenCalledWith(memorySecretProvenance)
+    expect(dbChainMockFns.onConflictDoUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        set: expect.objectContaining({ secretProvenanceVersion: null }),
+      })
+    )
+  })
+})

apps/sim/lib/memory/conversation-store.ts

@@ -208,6 +208,8 @@ export async function appendMemoryMessages(
             ? 'merge-provenance-limit'
             : undefined
       )
+    } else if (existing?.secretProvenanceVersion === 1) {
+      await replaceMemorySecretProvenanceInTx(tx, written.id, written.data, { status: 'unknown' })
     }
   })
 }

apps/sim/lib/memory/execution-record.test.ts

@@ -85,6 +85,19 @@ describe('portable execution records', () => {
     expect(JSON.parse(record.content!).notice).toBe('execution record shortened')
   })
 
+  it('bounds every default execution record without altering the canonical messages', () => {
+    const messages: Message[] = Array.from({ length: 30 }, (_, index) => ({
+      role: 'tool',
+      tool_call_id: `call-${index}`,
+      content: 'retained result '.repeat(1000),
+    }))
+    const original = structuredClone(messages)
+    const record = renderConversationExecutionRecord(messages)
+    expect(record.content!.length).toBeLessThanOrEqual(4096)
+    expect(record.content).toContain('execution record shortened')
+    expect(messages).toEqual(original)
+  })
+
   it('uses the same bounded format for protocol and context-size constraints', () => {
     const messages: Message[] = [{ role: 'tool', content: 'x'.repeat(10000) }]
     const record = renderConversationExecutionRecord(messages, 256)

apps/sim/lib/memory/execution-record.ts

@@ -1,26 +1,25 @@
 import { truncate } from '@sim/utils/string'
 import type { Message } from '@/providers/types'
 
-/** A text-only representation for exchanges that cannot be replayed as protocol tool messages. */
+/** Bounded model context for exchanges that cannot be replayed as protocol tool messages. */
 export function renderConversationExecutionRecord(
   messages: readonly Message[],
-  maxCharacters?: number
+  maxCharacters = 4096
 ): Message {
-  const bounded = maxCharacters !== undefined
   let shortened = false
   const field = (value: string, limit: number): string => {
-    if (!bounded || value.length <= limit) return value
+    if (value.length <= limit) return value
     shortened = true
     return truncate(value, limit)
   }
   const take = <T>(values: readonly T[], limit: number): readonly T[] => {
-    if (!bounded || values.length <= limit) return values
+    if (values.length <= limit) return values
     shortened = true
     return values.slice(0, limit)
   }
   const recordedMessages = take(messages, 21).map((message) => ({
     role: message.role,
-    content: bounded ? field(message.content ?? '', 512) : message.content,
+    content: message.content === null ? null : field(message.content ?? '', 512),
     ...(message.name ? { name: field(message.name, 64) } : {}),
     ...(message.tool_call_id ? { callId: field(message.tool_call_id, 64) } : {}),
     ...(message.function_call
@@ -47,11 +46,11 @@ export function renderConversationExecutionRecord(
     messages: recordedMessages,
     ...(shortened ? { notice: 'execution record shortened' } : {}),
   })
-  const suffix = bounded ? truncate('… [execution record shortened]', maxCharacters, '') : ''
+  const suffix = truncate('… [execution record shortened]', maxCharacters, '')
   return {
     role: 'user',
     content:
-      bounded && record.length > maxCharacters
+      record.length > maxCharacters
         ? truncate(record, Math.max(0, maxCharacters - suffix.length), suffix)
         : record,
   }

apps/sim/lib/memory/history-window.test.ts

@@ -96,6 +96,36 @@ describe('conversation history windows', () => {
     expect(selectConversationTokenWindow(pending.flat(), 1, undefined, pending)).toEqual(groups[1])
   })
 
+  it.each(['tool_calls', 'function_call'] as const)(
+    'counts legacy ungrouped %s arguments without changing plain-message token semantics',
+    (field) => {
+      const call = { name: 'lookup', arguments: JSON.stringify({ input: 'x'.repeat(1000) }) }
+      const assistant: Message = {
+        role: 'assistant',
+        content: '',
+        ...(field === 'tool_calls'
+          ? { tool_calls: [{ id: 'call', type: 'function' as const, function: call }] }
+          : { function_call: call }),
+      }
+      expect(selectConversationTokenWindow([assistant, ...final], 100)).toEqual(final)
+      expect(selectConversationTokenWindow([...user, ...final], 6)).toEqual(final)
+    }
+  )
+
+  it('counts legacy function arguments inside explicitly grouped history too', () => {
+    const group: Message[] = [
+      {
+        role: 'assistant',
+        content: '',
+        function_call: { name: 'lookup', arguments: JSON.stringify({ input: 'x'.repeat(1000) }) },
+      },
+      { role: 'function', name: 'lookup', content: 'result' },
+    ]
+    expect(
+      selectConversationTokenWindow([...group, ...final], 200, undefined, [group, final])
+    ).toEqual(final)
+  })
+
   it('applies model context bounds to complete groups', () => {
     const groups = [user, exchange('batch'), final]
     expect(selectConversationContextWindow(groups.flat(), 'small', groups)).toEqual(final)