fix(memory): validate checkpoint recovery and bounded summary coverage

Author: icecrasher321

.github/workflows/test-build.yml

@@ -232,6 +232,7 @@ jobs:
           lib/table/rows/secret-provenance.postgres.test.ts
           lib/memory/message-provenance.postgres.test.ts
           lib/memory/conversation-store.postgres.test.ts
+          lib/memory/summary-store.postgres.test.ts
           executor/handlers/agent/memory-harness.postgres.test.ts
 
       - name: Verify Search vector projection upgrade in PostgreSQL

apps/sim/lib/memory/README.md

@@ -20,7 +20,7 @@ A journal belongs to one execution, workflow, block, node, and execution-order i
 
 Recorded terminal tool outcomes are reused. Calls with no recorded outcome may run again with their original Sim invocation ID, so continuation provides at-least-once execution, not exactly-once external effects. An external service must support the supplied idempotency key to deduplicate an uncertain effect. A new execution or loop iteration has a separate journal.
 
-New durability failures degrade execution to its in-memory path. They must never cause a version-2 conversation to resume legacy-array writes. A missing result payload remains a terminal outcome, with an unavailable-detail notice and its recorded success status and cost. A missing or invalid step in a reference checkpoint, or oversized saved ciphertext, prevents continuation instead of starting the invocation again. Generic database availability failures retain the existing in-memory degradation behavior. API and ordinary memory behavior keep their existing error contract. Deleting a conversation takes the conversation lock and cascades item, journal, and artifact ownership rows. Active turns retain the original memory ID, so a later conversation with the same key cannot accept their stale checkpoint writes.
+New durability failures degrade execution to its in-memory path. They must never cause a version-2 conversation to resume legacy-array writes. A missing result payload remains a terminal outcome, with an unavailable-detail notice and its recorded success status and cost. Any nonempty saved checkpoint must decrypt, match its original identity and memory owner, and contain a valid state before continuation. Damaged ciphertext, invalid bindings or state, a missing step payload, and oversized saved ciphertext stop continuation instead of starting the invocation again. Database availability failures before a checkpoint is obtained retain the existing in-memory degradation behavior. API and ordinary memory behavior keep their existing error contract. Deleting a conversation takes the conversation lock and cascades item, journal, and artifact ownership rows. Active turns retain the original memory ID, so a later conversation with the same key cannot accept their stale checkpoint writes.
 
 Large-result artifacts have conversation ownership separate from run-log retention. The cleanup predicate retains owned artifacts and their dependencies while the conversation remains active; deleting the conversation releases this ownership.
 
@@ -42,15 +42,19 @@ The shared context policy uses an explicit memory token window when configured a
 
 `agent_memory_read` is supplied only with the trusted original memory owner. It searches retained history, including a safely projectable legacy prefix, or reads a referenced result in bounded pages. Legacy-prefix data and provenance must fit the 1 MiB retrieval admission limit. Each call returns at most 6,000 UTF-8 text bytes and scans at most 10 history items; a continuation cursor may be returned even when no match appears in a page. Cursors bind to the owner and search. Artifact reads use opaque IDs and canonical conversation ownership, and return only safely projected model content. Journal envelopes, provider continuations, raw replay fields, and unrelated conversation artifacts cannot be retrieved through this tool. At most two retrievals run concurrently per execution context. Read pages sequentially and follow `nextCursor` when more detail is needed.
 
-Before a generation omits older optional groups, a bound compactor can summarize both earlier conversation records and older completed exchanges from the active invocation. The current prompt and newest tool exchange remain required. A summary has priority within the same optional history budget as recent raw groups. Refreshes require additional history equal to at least half the configured history target, with a 1,024-token minimum, so every tool response does not trigger another summary call. A refreshed note can include the preceding note and newly eligible older records. The summary request has no tools, reads at most 8,000 estimated source tokens, and requests at most 1,024 output tokens, further constrained by the actual available summary budget. Its token/cost totals are recorded as `contextUsage`, without adding an execution step or a public conversation message. The latest summary is an encrypted derived cache on the original `memory` owner: at most 6,000 characters and 64 KiB of ciphertext, reused only for an exact versioned source hash. Cache replacement does not change the immutable transcript or its provenance, and generation runs outside database locks. Cache reads have a SQL byte guard and ordinary memory reads exclude the cache column. Failed summary generation or cache access falls back to bounded history selection.
+Before a generation omits older optional groups, a bound compactor can summarize earlier available conversation records and older completed exchanges from the active invocation. The current prompt and newest tool exchange remain required. A summary has priority within the same optional history budget as recent raw groups. Once the eligible backlog is covered, refreshes require additional history equal to at least half the configured history target, with a 1,024-token minimum, so every tool response does not trigger another summary call.
+
+Compaction processes at most three source batches per pressure event, in chronological order. Each batch combines the preceding note with the next contiguous older records, admits at most 8,000 estimated source tokens, and requests at most 1,024 output tokens, further constrained by the actual available summary budget. An individually large group may contribute a bounded excerpt with its call identities and artifact IDs. The coverage cursor advances only through a successfully summarized prefix; a failed or unusable response cannot mark skipped records as covered. Remaining backlog can continue at the next pressure event. Summary calls have no tools, and their token/cost totals are recorded as `contextUsage` without adding an execution step or a public conversation message. Original records remain authoritative; a derived note may omit details and never authorizes tool replay.
+
+The latest summary is an encrypted version-2 cache on the original `memory` owner, bounded to 6,000 characters and 64 KiB of ciphertext. Reuse requires validated cache metadata: its original-message count and source hash must exactly match an eligible canonical history prefix under the current summary policy. The source hash follows original records, independent of intermediate generated wording. Cache replacement does not change the immutable transcript or its provenance, and generation runs outside database locks. Cache reads have a SQL byte guard and ordinary memory reads exclude the cache column. Failed summary generation or cache access preserves any earlier usable note and falls back to bounded history selection.
 
 The public [Agent block documentation](../../../docs/content/docs/workflows/blocks/agent.mdx) describes these user-visible limits.
 
 ## Verification
 
 The optional five-family live contract suite is `providers/conversation-smoke.test.ts`. It is skipped unless `RUN_AGENT_MEMORY_PROVIDER_SMOKE=true` and `AGENT_MEMORY_PROVIDER_SMOKE_CASES` supplies an array of `{protocol, providerId, model, apiKey?, azureEndpoint?, azureApiVersion?, bedrockAccessKeyId?, bedrockSecretKey?, bedrockRegion?}` entries. Supply one entry for each protocol from `history-adapters.ts`. It executes a mocked, side-effect-free echo tool and then sends the captured native history through a second live request. These calls use provider credits; credentials remain in the environment and must not be committed. The implementation verification did not enable this paid suite.
 
-`conversation-store.postgres.test.ts` creates a disposable schema in a local database selected by `MEMORY_PROVENANCE_TEST_DATABASE_URL`. It applies migration 0368 and checks legacy-prefix preservation, public projections, CAS conflicts and rollback, append deduplication, pair-safe history windows, deletion/recreation, artifact retention, and oversized-checkpoint admission. `summary-store.postgres.test.ts` verifies exact hash reuse, scope binding, concurrent replacement/deletion, encrypted cache bounds, and exclusion from public projections and SQL reads. That suite requires the isolated local database at `127.0.0.1:5433/sim_durable_memory_e2e`. `message-provenance.postgres.test.ts` also applies this migration and verifies legacy provenance through a 17,000-message conversation and concurrent native/tool appends. All three suites drop their disposable schemas afterward.
+`conversation-store.postgres.test.ts` creates a disposable schema in a local database selected by `MEMORY_PROVENANCE_TEST_DATABASE_URL`. It applies migration 0368 and checks legacy-prefix preservation, public projections, CAS conflicts and rollback, append deduplication, pair-safe history windows, deletion/recreation, artifact retention, and oversized-checkpoint admission. `summary-store.postgres.test.ts` verifies exact hash reuse, scope binding, concurrent replacement/deletion, encrypted cache bounds, and exclusion from public projections and SQL reads. It uses the same explicitly configured local database and a disposable schema. `message-provenance.postgres.test.ts` also applies this migration and verifies legacy provenance through a 17,000-message conversation and concurrent native/tool appends. All three suites drop their disposable schemas afterward.
 
 The journal/session fixtures cover terminal sibling replay, missing artifacts, legacy snapshot migration, byte signatures, provenance, and context usage. The growth regression records 50 steps/results and checks that each payload is uploaded once while the encrypted manifest stays below 120 KiB. Run the focused suites from `apps/sim`:
 

apps/sim/lib/memory/agent-turn-session.test.ts

@@ -426,7 +426,12 @@ describe('durable Agent session', () => {
       )
 
       expect(live.rawResponse).toBe(response)
-      expect(live.modelResponse.output.text).toBe(response.output.text)
+      expect(live.modelResponse).toMatchObject({
+        success,
+        output: { memoryResultUnavailable: true },
+        ...(!success ? { error: 'Upstream rejected the operation' } : {}),
+      })
+      expect(JSON.stringify(live.modelResponse).length).toBeLessThan(2000)
       expect(executeTool).toHaveBeenCalledTimes(1)
       expect(session!.getPendingCalls()).toEqual([])
       const recorded = session!.getRecordedResult(invocationId)
@@ -467,20 +472,40 @@ describe('durable Agent session', () => {
     expect(session!.getRecordedResult(invocationId)?.modelResponse).toEqual(response)
   })
 
-  it('does not attach another invocation or recreated conversation journal', async () => {
-    const first = await openAgentTurnSession(input())
-    await first!.captureStep(step())
-    const encryptedState = save.mock.calls.at(-1)![0].input.encryptedState
-    open.mockResolvedValue({
-      memoryId: 'replacement-memory',
-      turnId: 'replacement-turn',
-      revision: 2,
-      encryptedState,
-    })
-    const replacement = await openAgentTurnSession(input(2))
-    expect(replacement!.getPendingCalls()).toEqual([])
-    expect(replacement!.getMessages('openai', 'model-a', 'binding-a')).toEqual([])
-  })
+  it.each(['damaged ciphertext', 'invocation binding', 'memory binding', 'invalid state'])(
+    'refuses an empty fresh session when a saved checkpoint has %s',
+    async (failure) => {
+      const first = (await openAgentTurnSession(input()))!
+      await first.captureStep(step())
+      const response = { success: true, output: { delivered: true } }
+      await first.recordToolResult({
+        invocationId: first.getPendingCalls()[0].invocationId,
+        rawResponse: response,
+        modelResponse: response,
+      })
+      let encryptedState: string = save.mock.calls.at(-1)![0].input.encryptedState
+      if (failure === 'damaged ciphertext') {
+        const last = encryptedState.at(-1) === '0' ? '1' : '0'
+        encryptedState = `${encryptedState.slice(0, -1)}${last}`
+      } else if (failure === 'invalid state') {
+        const envelope = await decryptMemoryCheckpoint(encryptedState)
+        if (!isRecordLike(envelope)) throw new Error('Expected checkpoint envelope')
+        encryptedState = await encryptMemoryCheckpoint({ ...envelope, state: { version: 99 } })
+      }
+      open.mockResolvedValue({
+        memoryId: failure === 'memory binding' ? 'replacement-memory' : 'memory-1',
+        turnId: 'turn-1',
+        revision: 2,
+        encryptedState,
+      })
+      const retry = input(failure === 'invocation binding' ? 2 : 1)
+      await expect(openAgentTurnSession(retry)).rejects.toMatchObject({ retryable: false })
+      await expect(openAgentTurnSession(retry)).rejects.toMatchObject({ retryable: false })
+      expect(save).toHaveBeenCalledTimes(2)
+      expect(readArtifact).not.toHaveBeenCalled()
+      expect(executeTool).not.toHaveBeenCalled()
+    }
+  )
 
   it('writes payloads once while a long invocation grows beyond the old snapshot byte limit', async () => {
     const session = (await openAgentTurnSession(input()))!

apps/sim/lib/memory/agent-turn-session.ts

@@ -412,7 +412,6 @@ export async function openAgentTurnSession(
   let record: AgentMemoryTurnRecord | undefined
   let state: AgentTurnState | undefined
   let journal: AgentTurnJournal | undefined
-  let restoringJournal = false
   let degraded = false
   const degrade = () => {
     if (!degraded)
@@ -463,15 +462,15 @@ export async function openAgentTurnSession(
         restored.memoryId !== record.memoryId
       )
         throw new Error('Invalid Agent checkpoint binding')
-      restoringJournal = isRecordLike(restored.state) && restored.state.version === 2
+      const restoringJournal = isRecordLike(restored.state) && restored.state.version === 2
       const restoredState = restoringJournal
         ? await journal.restore(restored.state)
         : restored.state
       if (!validState(restoredState)) throw new Error('Invalid Agent checkpoint state')
       state = restoredState
     }
   } catch (error) {
-    if (restoringJournal || (isRecordLike(error) && error.code === 'payload_too_large'))
+    if (record?.encryptedState || (isRecordLike(error) && error.code === 'payload_too_large'))
       throw Object.assign(new Error('Agent invocation journal could not be safely restored'), {
         retryable: false,
       })

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

@@ -11,6 +11,10 @@ const database = vi.hoisted(() => ({ current: undefined as PostgresJsDatabase |
 vi.unmock('drizzle-orm')
 vi.unmock('@sim/db/schema')
 vi.mock('@sim/db', () => ({
+  dbFor: () => {
+    if (!database.current) throw new Error('Postgres test is not initialized')
+    return database.current
+  },
   db: {
     select: (...args: unknown[]) => {
       if (!database.current) throw new Error('Postgres test is not initialized')
@@ -63,6 +67,7 @@ import {
   readPlainMemoryTail,
   saveAgentMemoryTurn,
 } from '@/lib/memory/conversation-store'
+import { retrieveMemory } from '@/lib/memory/retrieval'
 import {
   getMemoryMessageAppendKey,
   getMemoryMessageTurnId,
@@ -94,6 +99,7 @@ const identity: AgentMemoryTurnIdentity = {
 }
 const provenance = { status: 'exact', entries: [] } as const
 const journalReads: string[] = []
+const deduplicationReads: string[] = []
 const prefix = [
   { role: 'user', content: 'legacy question' },
   { role: 'assistant', content: 'legacy answer' },
@@ -156,6 +162,12 @@ describe.skipIf(!databaseUrl)('conversation storage in Postgres', () => {
         logQuery(query) {
           if (query.startsWith('select ') && query.includes('agent_memory_turn'))
             journalReads.push(query)
+          if (
+            query.startsWith('select ') &&
+            query.includes('from "memory_item"') &&
+            query.includes('"memory_item"."append_key" in')
+          )
+            deduplicationReads.push(query)
         },
       },
     })
@@ -348,6 +360,7 @@ describe.skipIf(!databaseUrl)('conversation storage in Postgres', () => {
   })
 
   it('deduplicates committed history and rolls back journal advancement on conflicting content', async () => {
+    deduplicationReads.length = 0
     const turn = await openAgentMemoryTurn(identity)
     const item = {
       appendKey: 'stable-exchange',
@@ -386,6 +399,10 @@ describe.skipIf(!databaseUrl)('conversation storage in Postgres', () => {
       (await readConversationItems({ workspaceId: identity.workspaceId, memoryId: turn.memoryId }))
         .items
     ).toHaveLength(1)
+    expect(deduplicationReads).toHaveLength(3)
+    for (const query of deduplicationReads) {
+      expect(query.split(' from ')[0]).toBe('select "append_key", "content_hash", "kind"')
+    }
   })
 
   it('deletion cascades history and forbids stale writers from resurrecting a recreated key', async () => {
@@ -698,4 +715,62 @@ describe.skipIf(!databaseUrl)('conversation storage in Postgres', () => {
       await connection!`SELECT id FROM memory_item WHERE memory_id = ${turn.memoryId}`
     ).toEqual([{ id: 'existing-large-item' }])
   })
+
+  it('finds older retained matches after a no-match page reaches the retrieval byte limit', async () => {
+    const turn = await openAgentMemoryTurn(identity)
+    await saveAgentMemoryTurn({
+      ...identity,
+      ...turn,
+      expectedRevision: 0,
+      encryptedState: 'saved',
+      items: [
+        {
+          appendKey: 'older-match',
+          kind: 'message',
+          data: { role: 'user', content: 'older retained receipt needle' },
+          provenance,
+        },
+        ...Array.from({ length: 5 }, (_, index) => ({
+          appendKey: `large-unrelated-${index}`,
+          kind: 'message' as const,
+          data: { role: 'assistant', content: 'x'.repeat(1024 * 1024 - 1024) },
+          provenance,
+        })),
+      ],
+    })
+    const scope = { workspaceId: identity.workspaceId, memoryId: turn.memoryId }
+    const contextPage = await readConversationItems({ ...scope, limit: 10 })
+    expect(contextPage.items).toHaveLength(4)
+    expect(contextPage.nextBeforeSequence).toBeUndefined()
+
+    const args = { target: 'history' as const, query: 'receipt needle' }
+    const first = await retrieveMemory({ ...scope, arguments: args, projection: {} })
+    expect(first).toMatchObject({ text: '', scannedItems: 4, nextCursor: expect.any(String) })
+    const next = await retrieveMemory({
+      ...scope,
+      arguments: { ...args, cursor: first.nextCursor },
+      projection: {},
+    })
+    expect(next.text).toContain('older retained receipt needle')
+    expect(next.scannedItems).toBe(2)
+  })
+
+  it('reports a single oversized history item and advances to the end without repeating it', async () => {
+    const turn = await openAgentMemoryTurn(identity)
+    const oversized = { role: 'user', content: 'x'.repeat(5 * 1024 * 1024) }
+    await connection!`INSERT INTO memory_item (id, memory_id, append_key, kind, data, content_hash, provenance_status, provenance_entries) VALUES ('oversized-retrieval-item', ${turn.memoryId}, 'oversized-retrieval', 'message', ${JSON.stringify(oversized)}::jsonb, ${hashDurableSecretProvenanceValue(oversized)}, 'exact', '[]')`
+    const scope = { workspaceId: identity.workspaceId, memoryId: turn.memoryId }
+    const args = { target: 'history' as const, query: 'needle' }
+    const first = await retrieveMemory({ ...scope, arguments: args, projection: {} })
+    expect(first.text).toBe('')
+    expect(first.notice).toContain('not retrievable within the safe 4 MiB')
+    expect(first.nextCursor).toEqual(expect.any(String))
+    const next = await retrieveMemory({
+      ...scope,
+      arguments: { ...args, cursor: first.nextCursor },
+      projection: {},
+    })
+    expect(next.nextCursor).toBeUndefined()
+    expect(next.scannedItems).toBe(0)
+  })
 })