fix(ai-harness, ai-persistence): pass inputs to the instance that owns a thread, and other pre-release fixes

.changeset/harness-define-plugin-keys.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-harness': patch
+---
+
+`definePlugin` refuses a `setup()` contribution, such as `adapter` or `tools`, at the top level of a definition, where it would be ignored. Return it from `setup()`.

.changeset/harness-forward-inputs.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-harness': patch
+---
+
+On a durable host with `stores.workClaims`, an input that reaches a host while another host owns the thread no longer takes the log from the running turn. With `stores.inbox`, the input waits in the inbox and the owner applies it (receipt `queued`). Without an inbox, it is refused with `owned_elsewhere`. A question from `ctx.session.ask` keeps its thread claimed, so an answer that reaches another host goes to the host that asked.

.changeset/harness-log-newer-records.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-harness': patch
+---
+
+A host that folds a `harness.*` log record type it does not know (a newer harness wrote it, for example during a rolling deploy) reads the log but refuses to write it, with `LogVersionError`. A fold checkpoint from a newer harness is ignored.

.changeset/harness-recover-live-owner.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-harness': patch
+---
+
+A durable host that opens a thread while another host holds its work claim no longer recovers that host's inputs, also an input that has no run lease yet. The crash sweep recovers them after the claim expires.

.changeset/persistence-store-shapes.md

@@ -0,0 +1,5 @@
+---
+'@tanstack/ai-persistence': patch
+---
+
+`InboxStore.get`, `markApplied`, and `markRejected` take the thread id first, so a store can key an entry by thread and input id. `InboxStore.delete` removes an entry: a harness host deletes a forwarded answer after it applies it, so a secret answer does not stay in the inbox. `WorkClaimStore.get(threadId)` returns the owner of a claim and when it expires. The session store docs allow a content-hash revision (an ETag of S3 or Netlify Blobs) that comes back for the same entry.

docs/config.json

@@ -761,7 +761,7 @@
           "label": "Store Reference",
           "to": "persistence/store-reference",
           "addedAt": "2026-08-04",
-          "updatedAt": "2026-10-14"
+          "updatedAt": "2026-10-15"
         },
         {
           "label": "How Persistence Works",
@@ -950,13 +950,14 @@
         {
           "label": "Durable sessions",
           "to": "harness/durable-sessions",
-          "addedAt": "2026-10-13"
+          "addedAt": "2026-10-13",
+          "updatedAt": "2026-10-15"
         },
         {
           "label": "List, rename, and fork sessions",
           "to": "harness/sessions",
           "addedAt": "2026-10-13",
-          "updatedAt": "2026-10-14"
+          "updatedAt": "2026-10-15"
         },
         {
           "label": "Run side effects once",

docs/harness/durable-sessions.md

@@ -91,6 +91,8 @@ process.on('SIGTERM', async () => {
 - A turn that a user cancelled still ends `aborted`.
 - `session.close({ recoverable: true })` does the same for one session.
 
+During a rolling deploy, old and new hosts run at the same time. A host that reads a log record type from a newer version does not write to that log. Its session stops with `LogVersionError`, and the new hosts continue the thread.
+
 ## Resume pending work after a deploy
 
 A host recovers a thread when something opens it. After a deploy, nobody opens a thread whose user closed the tab, so its queued turn or background agent waits. Give the host `stores.workClaims`, then call `host.resumePending` to find that work and run it.
@@ -145,6 +147,10 @@ The session renews the claim on the lease timer: every `lease.renewMs`, for `lea
 
 The session gives the claim back when the thread is idle: no queued or running turn, no waiting steer, and no running agent. A thread that waits for a human (an approval or a sign-in) is idle, so a sweep never takes it.
 
+A question from `ctx.session.ask` is different. The question waits in the memory of its host, so the session keeps the claim until the answer comes. Then an answer that reaches another host goes to this host. See [One host drives a thread](#one-host-drives-a-thread).
+
+The other host cannot tell if the question is a secret, so it stores the answer in `stores.inbox` as the user typed it. The host that asked removes that entry after it applies the answer. The log keeps a secret answer as `[secret]`. Keep the inbox in a store you trust as much as the log.
+
 ### What a sweep does
 
 `host.resumePending({ harnesses, close, limit })`:
@@ -305,7 +311,35 @@ const twoNotes = defineHarness({
 
 ## One host drives a thread
 
-Only one host writes to the log of a thread at a time. When a second host writes to the thread, the first host's session stops, and its running turn fails with a clear error. Open the thread again to continue on that host.
+Only one host writes to the log of a thread at a time. On a serverless platform, two requests for one thread often reach two instances. Give every instance `workClaims` and `inbox`. Then an input that reaches the second instance goes to the first:
+
+```ts group=harness-durable
+const replicaStores = memoryPersistence().stores
+
+const replica = createHarnessHost({
+  persistence: {
+    stores: {
+      log: memoryLogStore(),
+      runs: replicaStores.runs,
+      workClaims: replicaStores.workClaims,
+      inbox: replicaStores.inbox,
+    },
+  },
+})
+```
+
+The host that runs the thread holds its work claim. An input that reaches another host:
+
+- With `inbox`: waits in the inbox, and the owner applies it within a second. The receipt is `queued`.
+- Without `inbox`: is refused. The receipt is `rejected`, with the reason `owned_elsewhere`.
+
+In both cases, the running turn continues. A client reads the events of the thread from the log, on any instance. A `POST run` on the second instance streams the turn that the owner runs.
+
+No store keeps the `overrides` and the `ephemeral` messages of a prompt. So the owner runs a forwarded prompt without them. They apply only to a prompt that reaches the owner.
+
+When the owner stops, its claim expires. Then the next host that opens the thread, or a `resumePending` sweep, runs the inputs that wait.
+
+Without `workClaims`, a host cannot see the owner. When a second host writes to the thread, the first host's session stops, and its running turn fails with a clear error.
 
 The lease decides when a stopped host counts as gone. Change its timing on the host:
 
@@ -373,6 +407,7 @@ For the full contract, see [`LeaseStore`](../persistence/store-reference#leasest
 ## What you have now
 
 - Turns that continue after a crash or a deploy, also on threads that nobody opens again.
+- Inputs that reach any instance and run on the instance that owns the thread.
 - A turn that stops after `maxAttempts` or `timeoutMs`, with a clear error code.
 - Tools that run again safely or tell the model to check first, in your own words.
 - Retries of short model errors, with a count that survives a crash.

docs/harness/plugins.md

@@ -46,6 +46,8 @@ For agents and users:
 
 If two plugins add the same tool, command, setting, or prompt id, `host.open` fails and names both plugins.
 
+Return these from `setup`, not from the top level of `definePlugin`. A key like `adapter` or `tools` at the top level is a type error, and `definePlugin` throws for it. Other keys at the top level stay on the plugin object, so a plugin can give your code its own functions.
+
 ## Add commands and settings
 
 Commands are actions for the user, not the model: slash commands and buttons in your app. Settings are typed, saved per session, and apply at the next turn.

docs/harness/sessions.md

@@ -144,7 +144,7 @@ await session.unrevert()
 - The next prompt drops the hidden messages for good.
 - The session saves the revert in the log, or in `stores.metadata` on a host without a log. A restart keeps it.
 - Each change sends a `harness.revert` event. Read the transcript again when it arrives.
-- `revert` refuses with `busy` when the session is not idle. It refuses with `unknown_message` for an id that is not in the transcript.
+- `revert` and `unrevert` refuse with `busy` when the session is not idle, or when another host runs the thread. `revert` refuses with `unknown_message` for an id that is not in the transcript.
 
 ## Rename and remove a session
 

docs/persistence/store-reference.md

@@ -627,9 +627,10 @@ interface InboxEntry {
 interface InboxStore {
   append(entry: Omit<InboxEntry, 'status'>): Promise<InboxEntry>
   listPending(threadId: string): Promise<Array<InboxEntry>>
-  markApplied(entry: { inputId: string; operationId: string }): Promise<void>
-  markRejected(entry: { inputId: string; reason: string }): Promise<void>
-  get(inputId: string): Promise<InboxEntry | null>
+  markApplied(entry: { threadId: string; inputId: string; operationId: string }): Promise<void>
+  markRejected(entry: { threadId: string; inputId: string; reason: string }): Promise<void>
+  get(entry: { threadId: string; inputId: string }): Promise<InboxEntry | null>
+  delete(entry: { threadId: string; inputId: string }): Promise<void>
 }
 ```
 
@@ -640,7 +641,10 @@ Rules for an implementation:
   stored, it returns the stored entry and ignores the new payload.
 - `listPending` returns the pending entries of one thread, oldest `createdAt` first.
 - `markApplied` and `markRejected` do nothing for an unknown `inputId`.
-- `get` returns `null` for an unknown `inputId`.
+- `get` returns `null` for an unknown `inputId`, and for an `inputId` of another
+  thread.
+- `delete` removes the entry. It does nothing for an unknown `inputId`, or for
+  an `inputId` of another thread.
 
 In SQL, make `input_id` the primary key and write with `INSERT ... ON CONFLICT DO
 NOTHING`, then read the row back.
@@ -855,6 +859,7 @@ interface WorkClaimStore {
     until: number // epoch ms
   }): Promise<boolean>
   release(claim: { threadId: string; ownerId: string }): Promise<void>
+  get(threadId: string): Promise<{ ownerId: string; until: number } | null>
   listExpired(options: {
     now: number // epoch ms
     limit?: number
@@ -870,6 +875,8 @@ Rules for an implementation:
 - `claim` returns `false` while another owner holds a claim that has not expired.
 - `release` removes the claim only for its owner. For another `ownerId`, it does
   nothing.
+- `get` returns the owner and the `until` of the claim, or `null` when no one holds
+  the thread. It can return a claim that expired.
 - `listExpired` returns the claims that expired before `now`, oldest first, at most
   `limit`. Without `limit`, it returns all of them.
 
@@ -996,9 +1003,9 @@ Rules for `getVersioned` and `upsertIf`:
 
 - `getVersioned` returns the entry and its revision, or `undefined` for an unknown
   thread. The format of the revision is yours.
-- Each write gives the entry a new revision: `upsert` and `upsertIf` both. After
-  `delete`, the next write gets a new revision. The deleted revision is never used
-  again.
+- A write that changes the entry gives it a new revision: `upsert` and `upsertIf`
+  both. A revision can be a hash of the content (an ETag of S3 or Netlify Blobs).
+  Then the same entry, written again after a `delete`, gets the same revision back.
 - `upsertIf` writes only when the current revision is `expectedRevision`, and returns
   `{ ok: true, revision }`. Pass `null` to write only when the thread has no entry.
 - On any other revision, `upsertIf` writes nothing and returns

packages/ai-harness/src/host.ts

@@ -55,7 +55,10 @@ type DurableLeases =
  *   not yet applied.
  * - Durable mode: a `log`, and `runs` or `leases`. The log holds the events,
  *   transcript, inputs, and tool steps of each thread, so this shape has no
- *   `messages` and no `inbox`.
+ *   `messages`. With several hosts, add `workClaims`: an input that reaches a
+ *   host while another host owns the thread is refused (`owned_elsewhere`),
+ *   so it never stops the owner's turn. Add `inbox` too, and that input waits
+ *   in the inbox for the owner instead (receipt `queued`).
  *
  * Without `artifacts`, `blobs`, and `generationRuns`, media is kept in memory.
  * With `sessions`, the host keeps one index entry per thread (see
@@ -75,7 +78,8 @@ export type HarnessPersistence =
         interrupts?: InterruptStore
         metadata?: MetadataStore
         messages?: never
-        inbox?: never
+        /** Inputs for the host that owns a thread. Needs `workClaims`. */
+        inbox?: InboxStore
       } & DurableLeases &
         SharedStores
     >
@@ -538,9 +542,14 @@ function checkDurableStores(persistence: HarnessPersistence) {
       'A durable host (stores.log) needs stores.runs or stores.leases: a lease finds a crashed host.',
     )
   }
-  if (stores.messages !== undefined || stores.inbox !== undefined) {
+  if (stores.messages !== undefined) {
+    throw new Error(
+      'A durable host (stores.log) keeps the transcript and the inputs in the log. Remove stores.messages. A reader outside a session can use logMessageStore({ store }).',
+    )
+  }
+  if (stores.inbox !== undefined && stores.workClaims === undefined) {
     throw new Error(
-      'A durable host (stores.log) keeps the transcript and the inputs in the log. Remove stores.messages and stores.inbox. A reader outside a session can use logMessageStore({ store }).',
+      'A durable host (stores.log) uses stores.inbox to pass inputs to the host that owns a thread. That needs stores.workClaims.',
     )
   }
 }

packages/ai-harness/src/index.ts

@@ -25,7 +25,7 @@ export type {
   ToolStep,
 } from './durable-tool'
 
-export { logMessageStore } from './log'
+export { LogVersionError, logMessageStore } from './log'
 export type { ProjectOptions, ProjectRecord, ReduceOptions } from './log'
 export type { LeaseOptions } from './resume'
 export type {

packages/ai-harness/src/log.ts

@@ -119,6 +119,28 @@ const HARNESS_RECORD_TYPES = new Set<string>([
   'harness.note',
 ])
 
+/**
+ * The error a host gets when it writes a log that has a `harness.*` record
+ * type it does not know: a newer harness wrote it, for example during a
+ * rolling deploy. The host can read the log, but a write on a fold that
+ * skipped a record could be wrong, so it stops. Deploy the newer version.
+ *
+ * So a change to what a record means takes a new record type, never a new
+ * shape of an old type.
+ */
+export class LogVersionError extends Error {
+  readonly logId: string
+  readonly recordType: string
+  constructor(logId: string, recordType: string) {
+    super(
+      `Log ${logId} has a "${recordType}" record from a newer harness. This host does not write it.`,
+    )
+    this.name = 'LogVersionError'
+    this.logId = logId
+    this.recordType = recordType
+  }
+}
+
 // The harness wrote every record of these types, so the fields are its own.
 function isHarnessRecord(record: LogRecord): record is HarnessRecord {
   return HARNESS_RECORD_TYPES.has(record.type)
@@ -478,6 +500,8 @@ export interface SharedLogState<TState = unknown> {
   /** By thread id. */
   sessions: Map<string, LogState>
   reduced: TState | undefined
+  /** The first `harness.*` record type of the log that this version does not know. */
+  newerRecord?: string
 }
 
 export function emptySharedLogState<TState = unknown>(
@@ -520,6 +544,9 @@ export function foldLogEntry<TState>(
   state.seq = entry.seq
   const { record } = entry
   const { reduce } = options
+  if (record.type.startsWith('harness.') && !isHarnessRecord(record)) {
+    state.newerRecord ??= record.type
+  }
   if (reduce && !record.type.startsWith('harness.')) {
     const current = state.reduced ?? reduce.initial
     state.reduced = hostFold(
@@ -553,6 +580,8 @@ function serialize(state: SharedLogState, versions: CheckpointVersions) {
     v: 2,
     seq: state.seq,
     ...versions,
+    // A host that does not know one of these types folds from the start.
+    recordTypes: [...HARNESS_RECORD_TYPES],
     sessions: [...state.sessions].map(([thread, session]) => [
       thread,
       {
@@ -639,6 +668,13 @@ function parseCheckpoint<TState>(
   ) {
     return undefined
   }
+  // A newer harness folded record types this one would skip.
+  if (
+    Array.isArray(value.recordTypes) &&
+    value.recordTypes.some((type) => !HARNESS_RECORD_TYPES.has(String(type)))
+  ) {
+    return undefined
+  }
   const sessions = new Map<string, LogState>()
   for (const item of value.sessions) {
     if (!Array.isArray(item) || typeof item[0] !== 'string') return undefined
@@ -1066,6 +1102,11 @@ export class SharedLog<TState = unknown> {
   private async write(batch: ReadonlyArray<LogRecord>) {
     if (this.failure !== undefined) throw this.failure
     if (batch.length === 0) return
+    if (this.state.newerRecord !== undefined) {
+      const error = new LogVersionError(this.logId, this.state.newerRecord)
+      this.fail(error)
+      throw error
+    }
     const seq = this.state.seq + 1
     this.writing = true
     try {
@@ -1466,6 +1507,9 @@ export function logMessageStore(options: {
             : { ...record, thread: threadId },
         )
         if (records.length === 0) return
+        if (state.newerRecord !== undefined) {
+          throw new LogVersionError(logOf(threadId), state.newerRecord)
+        }
         try {
           await store.append({
             threadId: logOf(threadId),

packages/ai-harness/src/plugins.ts

@@ -325,6 +325,32 @@ export interface PluginDefinition<TMetadata = Record<string, unknown>> {
 
 const PLUGIN_BRAND = Symbol.for('tanstack.ai.harnessPlugin')
 
+/**
+ * What `setup()` returns. A key of it at the top level of a definition would
+ * be ignored. Other extra keys are the plugin's own API, such as the `diff`
+ * of `snapshots()`.
+ */
+const SETUP_KEYS: ReadonlySet<string> = new Set<keyof PluginContributions>([
+  'adapter',
+  'tools',
+  'prompts',
+  'middleware',
+  'agentMiddleware',
+  'generationMiddleware',
+  'agents',
+  'subagents',
+  'commands',
+  'config',
+  'contribute',
+  'discoverTools',
+  'prepareTools',
+])
+
+/** No `setup()` contribution at the top level of a definition. */
+type NoSetupKeys = {
+  [K in Exclude<keyof PluginContributions, keyof PluginDefinition>]?: never
+}
+
 /**
  * A plugin made with {@link definePlugin}. `TMetadata` is the session
  * metadata type it reads, from its `metadataSchema`.
@@ -361,11 +387,22 @@ export function definePlugin<
   const TDefinition extends PluginDefinition<MetadataOf<TMetadataSchema>> =
     PluginDefinition<MetadataOf<TMetadataSchema>>,
 >(
-  definition: TDefinition & { metadataSchema?: TMetadataSchema },
+  definition: TDefinition & {
+    metadataSchema?: TMetadataSchema
+  } & NoSetupKeys,
 ): HarnessPlugin<TDefinition, MetadataOf<TMetadataSchema>> {
   if (definition.name.trim() === '') {
     throw new Error('definePlugin requires a non-empty name')
   }
+  // `const TDefinition` turns off the excess property check, so a `setup()`
+  // contribution such as `adapter` at the top level would be kept and never
+  // used. Refuse it.
+  for (const key of Object.keys(definition)) {
+    if (!SETUP_KEYS.has(key)) continue
+    throw new Error(
+      `definePlugin "${definition.name}": \`${key}\` is not a plugin option. Return it from \`setup()\` instead.`,
+    )
+  }
   assertMetadataSchema(
     `definePlugin "${definition.name}"`,
     definition.metadataSchema,