Activity events that a chat middleware emits are streamed but never recorded or persisted

TanStack AI version

@tanstack/ai 0.69.0, @tanstack/ai-persistence 0.9.0 (the same behavior is on main at 85c2956)

Framework/Library version

Node 26.3.1, server only (no UI framework involved)

Describe the bug and the steps to reproduce it

An ACTIVITY_SNAPSHOT (or ACTIVITY_DELTA) that a chat middleware returns from onChunk reaches the client live, but the engine does not record it in ctx.activities. withPersistence therefore never saves it, and reconstructChat returns no activity row for it after a reload. The same event yielded by the adapter is recorded, saved and restored.

Activity rows and the activities store came in #1323 (issue #1286).

The cause is the order in the chat engine. In packages/ai/src/activities/chat/index.ts, each adapter chunk goes through handleStreamChunk(raw) before middleware runs. handleStreamChunk is the only place that applies ACTIVITY_SNAPSHOT / ACTIVITY_DELTA to the activity records (setActivities). Chunks that middleware adds go through emitPublicChunks, which normalizes and yields them but does not apply activity events. withPersistence saves ctx.activities, so it only ever sees adapter activity.

This is about the activity store (stores.activities, which withPersistence writes from ctx.activities and reconstructChat reads). It is not about the durability log, the delivery stream that toServerSentEventsResponse({ durability }) appends to. From the code, the middleware chunk is on the public stream, so I expect the durability log to carry it like any other chunk. I did not test that. The gap is that reconstructChat has no row for it.

The docs in docs/persistence/chat-persistence.md ("AG-UI activity rows on reload") present activity rows as the place for "a search progress card", and say withPersistence and reconstructChat "save and restore the rows when stores.activities is set". docs/advanced/middleware.md documents that onChunk can return StreamChunk[] to add chunks. A progress card is often made by middleware (for example, middleware that wraps a search tool), not by the provider adapter, so the two documented features do not work together today.

Steps:

1. Install @tanstack/[email protected] and @tanstack/[email protected].
2. Save the script below as repro.mjs in an ESM package and run node repro.mjs.

repro.mjs:

import { chat, defineChatMiddleware } from '@tanstack/ai'
import { memoryPersistence, reconstructChat, withPersistence } from '@tanstack/ai-persistence'

const progress = {
  type: 'ACTIVITY_SNAPSHOT',
  messageId: 'search-1',
  activityType: 'search',
  content: { status: 'done', hits: 3 },
  timestamp: Date.now(),
}

function adapter(withActivity) {
  return {
    kind: 'text',
    name: 'fixed',
    model: 'test-model',
    '~types': {},
    async *chatStream({ threadId, runId }) {
      const t = Date.now()
      yield { type: 'RUN_STARTED', threadId, runId, timestamp: t }
      if (withActivity) yield progress
      yield { type: 'TEXT_MESSAGE_START', messageId: 'msg-1', role: 'assistant', timestamp: t }
      yield { type: 'TEXT_MESSAGE_CONTENT', messageId: 'msg-1', delta: 'answer', timestamp: t }
      yield { type: 'TEXT_MESSAGE_END', messageId: 'msg-1', timestamp: t }
      yield { type: 'RUN_FINISHED', threadId, runId, finishReason: 'stop', timestamp: t }
    },
  }
}

// The same activity, produced by a middleware instead of the adapter.
const searchProgress = defineChatMiddleware({
  name: 'search-progress',
  onChunk(ctx, chunk) {
    if (chunk.type === 'TEXT_MESSAGE_START') return [progress, chunk]
  },
})

async function run(label, fromMiddleware) {
  const persistence = memoryPersistence()
  const threadId = `thread-${label}`
  const stream = chat({
    adapter: adapter(!fromMiddleware),
    messages: [{ role: 'user', content: 'search' }],
    threadId,
    runId: `run-${label}`,
    middleware: [
      withPersistence(persistence),
      ...(fromMiddleware ? [searchProgress] : []),
    ],
  })
  const wire = []
  for await (const chunk of stream) wire.push(chunk.type)
  const stored = await persistence.stores.activities.loadActivities(threadId)
  const res = await reconstructChat(persistence, new Request(`https://x/chat?threadId=${threadId}`))
  const roles = (await res.json()).messages.map((m) => m.role)
  console.log(label)
  console.log('  ACTIVITY_SNAPSHOT on the stream:', wire.includes('ACTIVITY_SNAPSHOT'))
  console.log('  stored activities:', JSON.stringify(stored))
  console.log('  reconstructChat roles:', JSON.stringify(roles))
}

await run('from adapter', false)
await run('from middleware', true)

Expected

The activity row is stored and restored in both runs:

from adapter
  ACTIVITY_SNAPSHOT on the stream: true
  stored activities: [{"id":"search-1","activityType":"search","content":{"status":"done","hits":3},"index":1}]
  reconstructChat roles: ["user","activity","assistant"]
from middleware
  ACTIVITY_SNAPSHOT on the stream: true
  stored activities: [{"id":"search-1", ...}]
  reconstructChat roles: ["user","activity","assistant"]

Actual

from adapter
  ACTIVITY_SNAPSHOT on the stream: true
  stored activities: [{"id":"search-1","activityType":"search","content":{"status":"done","hits":3},"index":1}]
  reconstructChat roles: ["user","activity","assistant"]
from middleware
  ACTIVITY_SNAPSHOT on the stream: true
  stored activities: []
  reconstructChat roles: ["user","assistant"]

The expected contract, from the docs above, is that an activity event on the stream is saved and restored no matter which layer added it. A possible fix is to apply ACTIVITY_SNAPSHOT / ACTIVITY_DELTA to ctx.activities for middleware output in emitPublicChunks too. If activity rows are meant to come only from adapters, the activity docs should say so.

Your Minimal, Reproducible Example - (Sandbox Highly Recommended)

Inline above. Plain Node script, no network, no model.

Screenshots or Videos (Optional)

None.

Do you intend to try to help solve this bug with your own PR?

Maybe, I'll investigate and start debugging

Terms & Code of Conduct

• I agree to follow this project's Code of Conduct
• I understand that if my bug cannot be reliable reproduced in a debuggable environment, it will probably not be fixed and this issue may even be closed.