Your framework keeps its own state next to the conversation: signals for the model, state writes, a compaction summary. Kept in a second store, it can drift from the transcript after a crash. Put these records in the harness session log. They share its atomic appends, its order, and its recovery.
session.append adds records to the session log, in one batch after the events that wait:
import { createHarnessHost, defineHarness } from '@tanstack/ai-harness'
import { memoryPersistence } from '@tanstack/ai-persistence'
import { openaiText } from '@tanstack/ai-openai'
const persistence = memoryPersistence()
const { log } = persistence.stores
const assistant = defineHarness({
name: 'acme/assistant',
adapter: openaiText('gpt-6.1-sol'),
})
const host = createHarnessHost({ persistence })
const session = await host.open(assistant, { threadId: 'thread-1' })
await session.appendRecords({ records: [{ type: 'app.signal', text: 'The build failed.' }] })With keepLogRecords: 'since-snapshot', the store drops the records before each snapshot. A plugin can make the snapshot keep the last records of its own types with keepRecords, and read them back with ctx.session.keptRecords(type):
import { defineCommand, definePlugin } from '@tanstack/ai-harness'
export const steps = definePlugin({
name: 'acme/steps',
keepRecords: { 'acme/steps:step': 100 },
setup: (ctx) => ({
commands: {
steps: defineCommand({
description: 'Count the kept steps',
run: async () => (await ctx.session.keptRecords('acme/steps:step')).length,
}),
},
}),
})By default, the model does not see your records. Set durability.addRecordsToModelContext on the harness to fold a record into the messages. Its record function gets one object, { messages, record }, and returns the new list, or undefined for no change:
import type { ProjectOptions } from '@tanstack/ai-harness'
const project: ProjectOptions = {
version: 'v1',
record: ({ messages, record }) => {
if (record.type === 'app.signal' && typeof record.text === 'string') {
return [...messages, { role: 'user', content: record.text }]
}
return undefined
},
}
export const projected = defineHarness({
name: 'acme/projected',
adapter: openaiText('gpt-6.1-sol'),
durability: { addRecordsToModelContext: project },
})A webhook or a browser knows something that the model needs: the build failed, or a payment arrived. Let it add a record to the thread, and the next turn reads the record through durability.addRecordsToModelContext.
import { createHarnessHandler } from '@tanstack/ai-harness/server'
export const watcher = defineHarness({
name: 'acme/build-watcher',
adapter: openaiText('gpt-6.1-sol'),
expose: { records: ['app.signal'] },
durability: { addRecordsToModelContext: project },
})
const watcherHost = createHarnessHost({
persistence: memoryPersistence(),
access: {
getUser: ({ request }) =>
request.headers.get('authorization') === `Bearer ${process.env.HARNESS_TOKEN}`
? { id: 'ci-bot' }
: null,
// The CI bot is the only user, so it can use every thread.
rules: { isAllowed: ({ user }) => user.id === 'ci-bot' },
},
})
// Every instance signs media URLs with the same secret.
const mediaSecret = process.env.HARNESS_MEDIA_SECRET
if (!mediaSecret) throw new Error('Set HARNESS_MEDIA_SECRET.')
export const handler = createHarnessHandler({
host: watcherHost,
harness: watcher,
media: { secret: mediaSecret },
})import { createHarnessClient } from '@tanstack/ai-harness/client'
import type { watcher } from './harness'
const client = createHarnessClient<typeof watcher>({
url: '/api/harness',
threadId: 'build-thread',
headers: { authorization: 'Bearer my-token' },
})
await client.appendRecords({ records: [{ type: 'app.signal', text: 'The build failed.' }] })
await client.continueWithoutMessage()The store is yours, so read your records back from it, for example to rebuild your state when a session opens. The store keeps each log under an id with its tenant, so build the id with logIdOf:
import { logIdOf } from '@tanstack/ai-harness'
// The default tenant. For a tenant, pass `tenantId` too: 'acme:thread-1'.
const entries = await log.read({ logId: logIdOf({ logId: 'thread-1' }) })
const signals = entries.filter((entry) => entry.record.type === 'app.signal')
console.log(signals.map((entry) => entry.seq))Each harness.input record has the user of its sender: its id, name, and tenantId. Every kind of input has it when the sender is known, also a cancel, an answer, a delivery change, and a change of settings.
Two harness records tell who answers a thread and where a conversation went. See Hand off a conversation.
const handoffs = entries.filter((entry) => entry.record.type === 'harness.handoff')
console.log(handoffs.at(-1)?.record.to)A reader outside a session that needs a MessageStore, for example reconstructChat, uses logMessageStore. It folds the log with the same addRecordsToModelContext, and it reads and writes the transcript that readers see, not the model context. Pass tenantId to read the threads of a tenant:
import { logMessageStore } from '@tanstack/ai-harness'
const messages = logMessageStore({ store: log, project })
console.log(await messages.loadThread('thread-1'))Your dashboard shows every agent run of a thread, also the runs that ended long ago. The log has one harness.run.node record for each run that the session admits:
const records = (await log.read({ logId: logIdOf({ logId: 'thread-1' }) })).map(
(entry) => entry.record,
)
const agentNodes = records.filter(
(record) => record.type === 'harness.run.node' && record.kind === 'agent',
)
console.log(agentNodes.map((record) => [record.runId, record.agent, record.parentRunId]))Wrap the LogStore to encrypt the records that hold what people wrote. The host reads and writes through the wrapper, so it only sees plain records. HarnessRecord is the type of every record that the harness writes, so a wrong record type is a type error:
import { createCipheriv, createDecipheriv, createHmac } from 'node:crypto'
import { defineLogStore } from '@tanstack/ai-persistence'
import type { LogRecord, LogStore } from '@tanstack/ai-persistence'
import type { HarnessRecord } from '@tanstack/ai-harness'
// The harness records that hold messages, prompts, and tool results.
const SEALED: ReadonlySet<string> = new Set([
'harness.transcript',
'harness.input',
'harness.event',
'harness.tool.result',
] satisfies Array<HarnessRecord['type']>)
interface Keys {
aes: Buffer // 32 bytes
iv: Buffer // a second secret
}
// AES-256-GCM. The IV comes from the text, so the same text gives the same output.
function seal(keys: Keys, text: string) {
const iv = createHmac('sha256', keys.iv).update(text).digest().subarray(0, 12)
const cipher = createCipheriv('aes-256-gcm', keys.aes, iv)
const data = Buffer.concat([cipher.update(text, 'utf8'), cipher.final()])
return Buffer.concat([iv, cipher.getAuthTag(), data]).toString('base64')
}
function unseal(keys: Keys, sealed: string) {
const bytes = Buffer.from(sealed, 'base64')
const decipher = createDecipheriv('aes-256-gcm', keys.aes, bytes.subarray(0, 12))
decipher.setAuthTag(bytes.subarray(12, 28))
return Buffer.concat([decipher.update(bytes.subarray(28)), decipher.final()]).toString('utf8')
}
const isLogRecord = (value: unknown): value is LogRecord =>
typeof value === 'object' && value !== null && 'type' in value && typeof value.type === 'string'
export function sealedLogStore(store: LogStore, keys: Keys): LogStore {
const sealRecord = (record: LogRecord): LogRecord =>
SEALED.has(record.type)
? { type: record.type, sealed: seal(keys, JSON.stringify(record)) }
: record
const openRecord = (record: LogRecord): LogRecord => {
if (!SEALED.has(record.type) || typeof record.sealed !== 'string') return record
const opened: unknown = JSON.parse(unseal(keys, record.sealed))
if (!isLogRecord(opened)) throw new Error('A sealed record did not open.')
return opened
}
return defineLogStore({
append: (batch) => store.append({ ...batch, records: batch.records.map(sealRecord) }),
read: async (query) =>
(await store.read(query)).map((entry) => ({ ...entry, record: openRecord(entry.record) })),
offerInput: (input) => store.offerInput({ ...input, record: sealRecord(input.record) }),
listOfferedInputs: async (query) =>
(await store.listOfferedInputs(query)).map((offer) => ({
...offer,
record: openRecord(offer.record),
})),
// The snapshot holds the transcript too.
putSnapshot: (snapshot) =>
store.putSnapshot({ ...snapshot, state: seal(keys, JSON.stringify(snapshot.state)) }),
getSnapshot: async (query) => {
const snapshot = await store.getSnapshot(query)
if (!snapshot || typeof snapshot.state !== 'string') return snapshot
const state: unknown = JSON.parse(unseal(keys, snapshot.state))
return { ...snapshot, state }
},
// These calls carry no records.
subscribe: (watch) => store.subscribe(watch),
acquireOwnership: (claim) => store.acquireOwnership(claim),
renewOwnership: (claim) => store.renewOwnership(claim),
releaseOwnership: (claim) => store.releaseOwnership(claim),
pinToWorkers: (pin) => store.pinToWorkers(pin),
markHostAlive: (beat) => store.markHostAlive(beat),
listLogsWithoutOwner: (query) => store.listLogsWithoutOwner(query),
truncate: (cut) => store.truncate(cut),
deleteLog: (target) => store.deleteLog(target),
})
}
const keys = {
aes: Buffer.from(process.env.LOG_AES_KEY ?? '', 'base64'),
iv: Buffer.from(process.env.LOG_IV_KEY ?? '', 'base64'),
}
const sealedHost = createHarnessHost({
persistence: { stores: { ...persistence.stores, log: sealedLogStore(log, keys) } },
})CAUTION: The wrapper must give back on read exactly what it got on write: the same fields, the same values, and the same key order. If it does not, the duplicate-input checks and the folds of the session break. The host compares the records that it wrote with the records that it reads back. The store compares a second offer of an input with the first one by its JSON.
A host merges streamed text and tool call arguments for log.bufferStreamWriteMs milliseconds (default 100) before it appends them. A message start or end, a tool call start or end, and your own records append at once.
const smoothHost = createHarnessHost({
persistence: memoryPersistence(),
log: { bufferStreamWriteMs: 250 },
})A larger value writes less often, and clients get text in larger parts.