From a73716dd0657c60011ac4db549112c4fdaac705d Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:02:45 -0700 Subject: [PATCH 01/15] fix(ai-harness): definePlugin refuses a setup() contribution at the top level `const TDefinition` turns off the excess property check, so `definePlugin({ name, adapter })` type-checked and the adapter was never used. A key of the setup() return (adapter, tools, prompts, middleware, ...) at the top level is now a type error and throws. Other extra keys stay the plugin's own API, such as the diff() of snapshots(). Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-define-plugin-keys.md | 5 +++ packages/ai-harness/src/plugins.ts | 39 ++++++++++++++++++- .../ai-harness/tests/define-plugin.test.ts | 37 ++++++++++++++++++ 3 files changed, 80 insertions(+), 1 deletion(-) create mode 100644 .changeset/harness-define-plugin-keys.md create mode 100644 packages/ai-harness/tests/define-plugin.test.ts diff --git a/.changeset/harness-define-plugin-keys.md b/.changeset/harness-define-plugin-keys.md new file mode 100644 index 0000000000..91b6ef15d7 --- /dev/null +++ b/.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()`. diff --git a/packages/ai-harness/src/plugins.ts b/packages/ai-harness/src/plugins.ts index ddae900be5..b80f43f746 100644 --- a/packages/ai-harness/src/plugins.ts +++ b/packages/ai-harness/src/plugins.ts @@ -325,6 +325,32 @@ export interface PluginDefinition> { 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 = new Set([ + '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]?: 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> = PluginDefinition>, >( - definition: TDefinition & { metadataSchema?: TMetadataSchema }, + definition: TDefinition & { + metadataSchema?: TMetadataSchema + } & NoSetupKeys, ): HarnessPlugin> { 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, diff --git a/packages/ai-harness/tests/define-plugin.test.ts b/packages/ai-harness/tests/define-plugin.test.ts new file mode 100644 index 0000000000..0c966cb7cb --- /dev/null +++ b/packages/ai-harness/tests/define-plugin.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from 'vitest' +import { definePlugin } from '../src' +import { mockAdapter } from './helpers' + +describe('definePlugin', () => { + it('refuses a setup() contribution at the top level', () => { + const { adapter } = mockAdapter([]) + expect(() => + // @ts-expect-error `adapter` is returned from `setup()` + definePlugin({ name: 'acme/model', adapter: () => adapter }), + ).toThrow( + 'definePlugin "acme/model": `adapter` is not a plugin option. Return it from `setup()` instead.', + ) + }) + + it('keeps an extra key as the plugin API', () => { + const plugin = definePlugin({ + name: 'acme/api', + diff: (from: string) => `diff ${from}`, + }) + expect(plugin.diff('a')).toBe('diff a') + }) + + it('keeps every known option', () => { + const plugin = definePlugin({ + name: 'acme/ok', + lifetime: 'turn', + requires: [], + provides: [], + optionalRequires: [], + needs: {}, + setup: () => ({ prompts: ['hi'] }), + }) + expect(plugin.name).toBe('acme/ok') + expect(plugin.lifetime).toBe('turn') + }) +}) From 5aed172a91a6ac1aff0501b13e5e426ab8e0cfff Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 09:49:14 -0700 Subject: [PATCH 02/15] fix(ai-harness): refuse to write a log that has a record type from a newer harness During a rolling deploy an old host could fold a log with records it skips, then write on top of that wrong state. It now reads the log but fails the write with LogVersionError. Checkpoints list the record types their writer knew, so an old host never starts from a newer fold. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-log-newer-records.md | 5 +++ packages/ai-harness/src/index.ts | 2 +- packages/ai-harness/src/log.ts | 44 +++++++++++++++++++++++++ packages/ai-harness/tests/log.test.ts | 33 +++++++++++++++++++ 4 files changed, 83 insertions(+), 1 deletion(-) create mode 100644 .changeset/harness-log-newer-records.md diff --git a/.changeset/harness-log-newer-records.md b/.changeset/harness-log-newer-records.md new file mode 100644 index 0000000000..f614a2ed2e --- /dev/null +++ b/.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. diff --git a/packages/ai-harness/src/index.ts b/packages/ai-harness/src/index.ts index 89e16dcda4..ff3b7d2b95 100644 --- a/packages/ai-harness/src/index.ts +++ b/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 { diff --git a/packages/ai-harness/src/log.ts b/packages/ai-harness/src/log.ts index 4c86196282..9997a39451 100644 --- a/packages/ai-harness/src/log.ts +++ b/packages/ai-harness/src/log.ts @@ -119,6 +119,28 @@ const HARNESS_RECORD_TYPES = new Set([ '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 { /** By thread id. */ sessions: Map reduced: TState | undefined + /** The first `harness.*` record type of the log that this version does not know. */ + newerRecord?: string } export function emptySharedLogState( @@ -520,6 +544,9 @@ export function foldLogEntry( 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( ) { 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() for (const item of value.sessions) { if (!Array.isArray(item) || typeof item[0] !== 'string') return undefined @@ -1066,6 +1102,11 @@ export class SharedLog { private async write(batch: ReadonlyArray) { 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), diff --git a/packages/ai-harness/tests/log.test.ts b/packages/ai-harness/tests/log.test.ts index b6c2b4f3fb..85cbe8a7ea 100644 --- a/packages/ai-harness/tests/log.test.ts +++ b/packages/ai-harness/tests/log.test.ts @@ -6,6 +6,7 @@ import { memoryPersistence, } from '@tanstack/ai-persistence' import { + LogVersionError, SharedLog, emptySharedLogState, engineMessageStore, @@ -534,6 +535,34 @@ describe('write failures', () => { ]) }) + it('reads a log with a record type from a newer harness, and writes nothing to it', async () => { + const store = memoryLogStore() + const old = newWriter(store) + await old.commit({ messages: [user('u1', 'one')] }) + // A newer host appends a harness record type that this version does not know. + await store.append(THREAD, 2, [{ type: 'harness.later', value: 1 }]) + const state = await loadLogState({ store, logId: THREAD }) + expect(state.newerRecord).toBe('harness.later') + expect(sessionOf(state, THREAD).messages).toEqual([user('u1', 'one')]) + + const failures: Array = [] + const writer = newWriter(store, { + state, + onFailure: (e) => failures.push(e), + }) + await expect(writer.append([{ type: 'app.mine' }])).rejects.toBeInstanceOf( + LogVersionError, + ) + expect(failures[0]).toBeInstanceOf(LogVersionError) + await expect( + logMessageStore({ store }).saveThread(THREAD, [user('u2', 'two')]), + ).rejects.toBeInstanceOf(LogVersionError) + expect((await records(store)).map((record) => record.type)).toEqual([ + 'harness.transcript', + 'harness.later', + ]) + }) + it('calls onFailure on a conflict and rejects every later write', async () => { const store = memoryLogStore() const failures: Array = [] @@ -707,6 +736,10 @@ describe('fold checkpoints', () => { ['another projection version', checkpoint({ version: 'v0' })], ['a position past the end of the log', checkpoint({ seq: 99 })], ['a state that is not shaped', checkpoint({ sessions: 'x' })], + [ + 'a record type from a newer harness', + checkpoint({ recordTypes: ['harness.transcript', 'harness.later'] }), + ], ])('ignores a checkpoint with %s', async (_name, value) => { const store = memoryLogStore() await seedLog(store) From c17c820dcf511e24a0d54d5e0494fb226209dc72 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 09:53:53 -0700 Subject: [PATCH 03/15] fix(ai-persistence): store shapes that third-party backends can implement - InboxStore.get/markApplied/markRejected take threadId, so a store does not need a global index of input ids. - WorkClaimStore.get(threadId) tells a host who owns a thread. - The session revision contract allows content-hash ETags (S3, Netlify Blobs), which give the same revision back for the same content. None of these interfaces is released yet. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/persistence-store-shapes.md | 5 +++ packages/ai-harness/src/session/inputs.ts | 20 ++++++++-- packages/ai-harness/src/session/requests.ts | 6 ++- packages/ai-harness/tests/log.test.ts | 6 ++- packages/ai-harness/tests/session.test.ts | 5 ++- packages/ai-harness/tests/work-claims.test.ts | 11 +++-- packages/ai-persistence/src/memory.ts | 28 +++++++++++-- .../ai-persistence/src/testkit/conformance.ts | 40 +++++++++++++++---- packages/ai-persistence/src/types.ts | 26 +++++++++--- packages/ai-persistence/tests/inbox.test.ts | 4 +- 10 files changed, 124 insertions(+), 27 deletions(-) create mode 100644 .changeset/persistence-store-shapes.md diff --git a/.changeset/persistence-store-shapes.md b/.changeset/persistence-store-shapes.md new file mode 100644 index 0000000000..60b49e7950 --- /dev/null +++ b/.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. `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. diff --git a/packages/ai-harness/src/session/inputs.ts b/packages/ai-harness/src/session/inputs.ts index d0b3546a8d..293e0d17c8 100644 --- a/packages/ai-harness/src/session/inputs.ts +++ b/packages/ai-harness/src/session/inputs.ts @@ -217,7 +217,11 @@ export abstract class SessionInputs< } if (!this.writer) { this.inbox - .markApplied({ inputId: steer.inputId, operationId }) + .markApplied({ + threadId: this.threadId, + inputId: steer.inputId, + operationId, + }) .catch((error: unknown) => this.warnSession('harness:inbox', error)) } this.feed.publish( @@ -282,7 +286,11 @@ export abstract class SessionInputs< */ protected async applyControl(inputId: string, change: HarnessRecord) { if (!this.writer) { - await this.inbox.markApplied({ inputId, operationId: 'session' }) + await this.inbox.markApplied({ + threadId: this.threadId, + inputId, + operationId: 'session', + }) return } await this.writer.append([ @@ -381,7 +389,11 @@ export abstract class SessionInputs< operationId: string, ): Promise { if (!this.writer) { - await this.inbox.markApplied({ inputId, operationId }) + await this.inbox.markApplied({ + threadId: this.threadId, + inputId, + operationId, + }) return } const logged = this.writer.state.inputs.get(inputId) @@ -417,7 +429,7 @@ export abstract class SessionInputs< .catch(() => {}) } else { this.inbox - .markRejected({ inputId, reason }) + .markRejected({ threadId: this.threadId, inputId, reason }) .catch((error: unknown) => this.warnSession('harness:inbox', error)) } const receipt: Receipt = { inputId, status: 'rejected', reason } diff --git a/packages/ai-harness/src/session/requests.ts b/packages/ai-harness/src/session/requests.ts index 800b800263..b84c697287 100644 --- a/packages/ai-harness/src/session/requests.ts +++ b/packages/ai-harness/src/session/requests.ts @@ -590,7 +590,11 @@ export abstract class SessionRequests< // Without a log, the inbox entry must not stay pending: a restart runs // pending entries. if (!this.writer) - await this.inbox.markRejected({ inputId: target, reason: 'cancelled' }) + await this.inbox.markRejected({ + threadId: this.threadId, + inputId: target, + reason: 'cancelled', + }) const { operation } = taken await this.settle({ inputId: target, diff --git a/packages/ai-harness/tests/log.test.ts b/packages/ai-harness/tests/log.test.ts index 85cbe8a7ea..791b93beec 100644 --- a/packages/ai-harness/tests/log.test.ts +++ b/packages/ai-harness/tests/log.test.ts @@ -540,7 +540,11 @@ describe('write failures', () => { const old = newWriter(store) await old.commit({ messages: [user('u1', 'one')] }) // A newer host appends a harness record type that this version does not know. - await store.append(THREAD, 2, [{ type: 'harness.later', value: 1 }]) + await store.append({ + threadId: THREAD, + seq: 2, + records: [{ type: 'harness.later', value: 1 }], + }) const state = await loadLogState({ store, logId: THREAD }) expect(state.newerRecord).toBe('harness.later') expect(sessionOf(state, THREAD).messages).toEqual([user('u1', 'one')]) diff --git a/packages/ai-harness/tests/session.test.ts b/packages/ai-harness/tests/session.test.ts index dc8dbbc9ff..660d5fc781 100644 --- a/packages/ai-harness/tests/session.test.ts +++ b/packages/ai-harness/tests/session.test.ts @@ -438,7 +438,10 @@ describe('inbox', () => { ) await vi.waitFor(() => expect(session.snapshot().status).toBe('idle')) - expect((await persistence.stores.inbox.get('in-1'))?.status).toBe('applied') + expect( + (await persistence.stores.inbox.get({ threadId: 't1', inputId: 'in-1' })) + ?.status, + ).toBe('applied') expect(messageTexts(calls[0])).toEqual(['left over']) await host.close() }) diff --git a/packages/ai-harness/tests/work-claims.test.ts b/packages/ai-harness/tests/work-claims.test.ts index 063949238c..735fbfbe09 100644 --- a/packages/ai-harness/tests/work-claims.test.ts +++ b/packages/ai-harness/tests/work-claims.test.ts @@ -387,9 +387,14 @@ describe('host.resumePending', () => { }) await vi.waitFor(() => expect(calls).toHaveLength(1)) await vi.waitFor(async () => - expect((await persistence.stores.inbox.get('in-1'))?.status).toBe( - 'applied', - ), + expect( + ( + await persistence.stores.inbox.get({ + threadId: THREAD, + inputId: 'in-1', + }) + )?.status, + ).toBe('applied'), ) await host.close() }) diff --git a/packages/ai-persistence/src/memory.ts b/packages/ai-persistence/src/memory.ts index f57a2a6238..12347aab9e 100644 --- a/packages/ai-persistence/src/memory.ts +++ b/packages/ai-persistence/src/memory.ts @@ -625,35 +625,49 @@ class MemoryInboxStore implements InboxStore { return Promise.resolve(pending) } markApplied({ + threadId, inputId, operationId, }: { + threadId: string inputId: string operationId: string }): Promise { - const existing = this.entries.get(inputId) + const existing = this.of(threadId, inputId) if (existing) { this.entries.set(inputId, { ...existing, status: 'applied', operationId }) } return Promise.resolve() } markRejected({ + threadId, inputId, reason, }: { + threadId: string inputId: string reason: string }): Promise { - const existing = this.entries.get(inputId) + const existing = this.of(threadId, inputId) if (existing) { this.entries.set(inputId, { ...existing, status: 'rejected', reason }) } return Promise.resolve() } - get(inputId: string): Promise { - const existing = this.entries.get(inputId) + get({ + threadId, + inputId, + }: { + threadId: string + inputId: string + }): Promise { + const existing = this.of(threadId, inputId) return Promise.resolve(existing ? structuredClone(existing) : null) } + private of(threadId: string, inputId: string) { + const existing = this.entries.get(inputId) + return existing?.threadId === threadId ? existing : undefined + } } class MemoryCredentialStore implements CredentialStore { @@ -831,6 +845,12 @@ class MemoryWorkClaimStore implements WorkClaimStore { } return Promise.resolve() } + get(threadId: string) { + const held = this.claims.get(threadId) + return Promise.resolve( + held ? { ownerId: held.ownerId, until: held.until } : null, + ) + } listExpired(options: { now: number limit?: number diff --git a/packages/ai-persistence/src/testkit/conformance.ts b/packages/ai-persistence/src/testkit/conformance.ts index 8912745176..24e451a0a5 100644 --- a/packages/ai-persistence/src/testkit/conformance.ts +++ b/packages/ai-persistence/src/testkit/conformance.ts @@ -1674,23 +1674,47 @@ export function runPersistenceConformance( (await store.listPending(thread)).map((entry) => entry.inputId), ).toEqual([in1, in2]) - await store.markApplied({ inputId: in1, operationId: 'op-1' }) - await store.markRejected({ inputId: in2, reason: 'busy' }) + await store.markApplied({ + threadId: thread, + inputId: in1, + operationId: 'op-1', + }) + await store.markRejected({ + threadId: thread, + inputId: in2, + reason: 'busy', + }) expect(await store.listPending(thread)).toEqual([]) - expect(await store.get(in1)).toMatchObject({ + expect( + await store.get({ threadId: thread, inputId: in1 }), + ).toMatchObject({ status: 'applied', operationId: 'op-1', }) - expect(await store.get(in2)).toMatchObject({ + expect( + await store.get({ threadId: thread, inputId: in2 }), + ).toMatchObject({ status: 'rejected', reason: 'busy', principal: { id: 'user-1' }, }) // Unknown ids are no-ops. - await store.markApplied({ inputId: missing, operationId: 'op-x' }) - await store.markRejected({ inputId: missing, reason: 'nope' }) - expect(await store.get(missing)).toBeNull() + await store.markApplied({ + threadId: thread, + inputId: missing, + operationId: 'op-x', + }) + await store.markRejected({ + threadId: thread, + inputId: missing, + reason: 'nope', + }) + expect( + await store.get({ threadId: thread, inputId: missing }), + ).toBeNull() + // An id is found only with its own thread. + expect(await store.get({ threadId: thread, inputId: other })).toBeNull() }) }) @@ -1711,10 +1735,12 @@ export function runPersistenceConformance( expect(await claim('a')).toBe(true) // Another owner is refused while the claim is live. expect(await claim('b')).toBe(false) + expect(await store.get(threadId)).toMatchObject({ ownerId: 'a' }) // Only the owner releases. await store.release({ threadId, ownerId: 'b' }) expect(await claim('b')).toBe(false) await store.release({ threadId, ownerId: 'a' }) + expect(await store.get(threadId)).toBeNull() expect(await claim('b')).toBe(true) await store.release({ threadId, ownerId: 'b' }) }) diff --git a/packages/ai-persistence/src/types.ts b/packages/ai-persistence/src/types.ts index e46e69405f..0a2bdf1192 100644 --- a/packages/ai-persistence/src/types.ts +++ b/packages/ai-persistence/src/types.ts @@ -685,13 +685,21 @@ export interface InboxStore { listPending: (threadId: string) => Promise> /** Mark an entry applied by `operationId`. A no-op for an unknown id. */ markApplied: (entry: { + threadId: string inputId: string operationId: string }) => Promise /** Mark an entry rejected with `reason`. A no-op for an unknown id. */ - markRejected: (entry: { inputId: string; reason: string }) => Promise - /** The entry for `inputId`, or `null`. */ - get: (inputId: string) => Promise + markRejected: (entry: { + threadId: string + inputId: string + reason: string + }) => Promise + /** The entry for `inputId` of `threadId`, or `null`. */ + get: (entry: { + threadId: string + inputId: string + }) => Promise } /** Type an {@link InboxStore} implementation inline. */ @@ -891,6 +899,12 @@ export interface WorkClaimStore { }) => Promise /** The thread is idle: remove the claim. Only its owner can. */ release: (claim: { threadId: string; ownerId: string }) => Promise + /** + * The owner of the claim on `threadId` and when it expires, or `null` when + * no one holds it. A claim that expired can still come back: compare + * `until` with the time. + */ + get: (threadId: string) => Promise<{ ownerId: string; until: number } | null> /** Claims that expired before `now`: their host stopped. Oldest first. */ listExpired: (options: { now: number @@ -1004,8 +1018,10 @@ export interface SessionIndexStore { /** * Optional. The entry with a revision, or `undefined` when it is absent. * Pair it with `upsertIf` for compare-and-set writes: implement both or - * neither. Each write (`upsert` or `upsertIf`) gives the entry a new - * revision, and a revision does not come back after a `delete`. + * neither. A write that changes the entry gives it a new revision. A + * revision may name the content (a content-hash ETag, as in S3 or Netlify + * Blobs), so the same entry written again, also after a `delete`, can get + * the same revision back. */ getVersioned?: ( threadId: string, diff --git a/packages/ai-persistence/tests/inbox.test.ts b/packages/ai-persistence/tests/inbox.test.ts index b88414483e..2bfe2773a5 100644 --- a/packages/ai-persistence/tests/inbox.test.ts +++ b/packages/ai-persistence/tests/inbox.test.ts @@ -20,7 +20,9 @@ describe('inbox store', () => { }) entry.status = 'applied' - expect((await inbox.get('in-1'))?.status).toBe('pending') + expect((await inbox.get({ threadId: 't', inputId: 'in-1' }))?.status).toBe( + 'pending', + ) }) it('accepts a user-implemented inbox in defineAIPersistence', () => { From 02d9163c20b72c55a85314fcb9160d1c84037b29 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 09:59:51 -0700 Subject: [PATCH 04/15] fix(ai-harness): pass an input to the host that owns the thread, not over its turn On a durable host, an input on a second host appended to the log and made the running owner fail with LogConflictError, which aborted its turn. Two actors per cold start make this routine on serverless platforms. With stores.workClaims, accept() now checks who owns the thread: - With stores.inbox (now allowed in durable mode), the input waits in the inbox and the receipt is 'queued'. The owner drains the inbox every second while it holds the claim, once more when it goes idle, and when a session opens (the crash sweep). - Without an inbox, the input is refused with 'owned_elsewhere'. A question from ctx.session.ask claims the thread and keeps it busy, so its answer is forwarded to the host that holds the question. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-forward-inputs.md | 5 + packages/ai-harness/src/host.ts | 17 +- packages/ai-harness/src/session/base.ts | 166 ++++++++++++++++- .../ai-harness/src/session/harness-session.ts | 2 + packages/ai-harness/src/session/helpers.ts | 6 + packages/ai-harness/src/session/inputs.ts | 64 ++++++- packages/ai-harness/src/session/requests.ts | 20 ++- packages/ai-harness/src/session/state.ts | 13 +- packages/ai-harness/src/session/turn.ts | 19 +- .../ai-harness/tests/durable-session.test.ts | 2 +- packages/ai-harness/tests/forwarding.test.ts | 168 ++++++++++++++++++ 11 files changed, 463 insertions(+), 19 deletions(-) create mode 100644 .changeset/harness-forward-inputs.md create mode 100644 packages/ai-harness/tests/forwarding.test.ts diff --git a/.changeset/harness-forward-inputs.md b/.changeset/harness-forward-inputs.md new file mode 100644 index 0000000000..27ae95bb0b --- /dev/null +++ b/.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. diff --git a/packages/ai-harness/src/host.ts b/packages/ai-harness/src/host.ts index f0b1c201f7..6dcd3b042e 100644 --- a/packages/ai-harness/src/host.ts +++ b/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 > @@ -505,9 +509,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.', ) } } diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index f00f8e8350..2aa1040d42 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -34,6 +34,7 @@ import type { import type { AIPersistence, ChatTranscriptStores, + InboxEntry, InboxStore, MessageStore, SessionIndexStore, @@ -60,6 +61,7 @@ import type { import type { ChatTurnResult, Cursor, + HarnessInput, InputSettlement, OnEvent, Operation, @@ -99,6 +101,9 @@ const KEPT_SETTLED = 500 * session class is split into layers, one per file: base, inputs, state, * requests, turn-parts, agents, turn, harness-session. */ +/** How often the owner of a thread reads `stores.inbox`. */ +const INBOX_POLL_MS = 1_000 + export abstract class SessionBase { readonly threadId: string /** The log of this session. Sessions with the same log id share one log. */ @@ -180,6 +185,11 @@ export abstract class SessionBase { protected claimTimer: ReturnType | undefined /** The claim write of the last `markBusy`. An input waits for it. */ protected claimWrite: Promise = Promise.resolve() + /** The last claim write found another host's claim. */ + protected claimLost = false + /** Drains `stores.inbox` while this host owns the thread. */ + protected inboxTimer: ReturnType | undefined + private draining: Promise | undefined /** Called each time the thread goes idle. See `onIdle`. */ protected readonly idleListeners = new Set<() => void>() /** Work inputs that are being stored and are not queued yet. */ @@ -647,12 +657,13 @@ export abstract class SessionBase { // Another host's claim that has not expired is left alone: the call // returns false. The timer tries again while this thread has work, // so the work gets a claim once that one expires. - await claims.claim({ + const isHeld = await claims.claim({ threadId: this.threadId, harness: this.harness.name, ownerId: this.hostId, until: Date.now() + ttlMs, }) + this.claimLost = !isHeld } catch (error) { // The work goes on without a claim. this.warnClaim(error) @@ -666,11 +677,157 @@ export abstract class SessionBase { if (typeof timer === 'object' && 'unref' in timer) timer.unref() this.claimTimer = timer this.claimWrite = claim() + if (this.writer && this.persistence.stores.inbox) { + // ponytail: a 1s poll. A store that can push could wake the owner instead. + const poll = setInterval(() => void this.drainInbox(), INBOX_POLL_MS) + if (typeof poll === 'object' && 'unref' in poll) poll.unref() + this.inboxTimer = poll + } } protected stopClaimTimer(): void { if (this.claimTimer) clearInterval(this.claimTimer) this.claimTimer = undefined + if (this.inboxTimer) clearInterval(this.inboxTimer) + this.inboxTimer = undefined + } + + /** + * Does another host hold a live claim on this thread? Only a durable host + * with `stores.workClaims` asks. Its inputs then go to that host, through + * `stores.inbox`: a write here would take the log from the running owner. + */ + protected async isOwnedElsewhere(): Promise { + const claims = this.persistence.stores.workClaims + if (!this.writer || !claims) return false + const held = await claims.get(this.threadId) + return ( + held !== null && held.ownerId !== this.hostId && held.until > Date.now() + ) + } + + /** + * Apply the inputs that other hosts left in `stores.inbox` for this + * thread, in order. Only the host that holds the claim does: it claims the + * thread first. An input that ran before (a drain cut by a crash) is a + * duplicate in the log. + */ + protected drainInbox(): Promise { + this.draining ??= this.drainOnce().finally(() => { + this.draining = undefined + }) + return this.draining + } + + private async drainOnce(): Promise { + const inbox = this.persistence.stores.inbox + const claims = this.persistence.stores.workClaims + if (!this.writer || !inbox || !claims || this.closing) return + try { + const pending = await inbox.listPending(this.threadId) + if (pending.length === 0) return + const isHeld = await claims.claim({ + threadId: this.threadId, + harness: this.harness.name, + ownerId: this.hostId, + until: Date.now() + (this.lease?.ttlMs ?? LEASE.ttlMs), + }) + if (!isHeld) return + for (const entry of pending) { + await this.applyForwarded(entry) + await inbox.markApplied({ + threadId: this.threadId, + inputId: entry.inputId, + operationId: 'forwarded', + }) + } + // A drain of controls only leaves the thread idle: give the claim back. + this.checkIdle() + } catch (error) { + this.warnClaim(error) + } + } + + /** Apply an input that another host stored, as if it arrived here. */ + private async applyForwarded(entry: InboxEntry): Promise { + const session = this.self() + // The host that stored it checked the input against `HarnessInput`. + const input = entry.input as HarnessInput + const { inputId } = entry + const principal = entry.principal + const sent = { + inputId, + ...(principal ? { principal } : {}), + ...('context' in input && input.context !== undefined + ? { context: input.context } + : {}), + } + const settle = (operation: PromiseLike) => { + // Nobody awaits a forwarded turn here. Its client reads the log. + operation.then( + () => {}, + () => {}, + ) + } + switch (input.op) { + case 'prompt': + return settle( + session.prompt(input.message, { + ...sent, + ...(input.busy ? { busy: input.busy } : {}), + }), + ) + case 'continue': + return settle(session.continue(sent)) + case 'steer': + return session.steer(input.message, sent) + case 'followUp': + return session.followUp(input.message, sent) + case 'resolve': + return session.resolve(input.resume, sent) + case 'reset': + return session.reset(input.note, sent) + case 'configure': + return session.configure(input.settings, sent) + case 'agentMessage': + return session.sendToAgent({ + ...sent, + operationId: input.operationId, + message: input.message, + mode: input.mode, + }) + case 'cancel': + return session.cancel(input.operationId) + case 'cancelInput': + return session.cancelInput(input.inputId) + case 'setDelivery': + return session.setDelivery({ + inputId: input.inputId, + delivery: input.delivery, + }) + case 'command': + return settle( + session.command( + input.name, + input.input, + principal ? { principal } : undefined, + ), + ) + case 'answer': + return session.answer({ + questionId: input.questionId, + value: input.value, + }) + case 'config': + return session.setConfig({ key: input.key, value: input.value }) + case 'revert': + return session.revert(input.messageId) + case 'unrevert': + return session.unrevert() + default: + // `agent` and `background` start on the owner, never on another host. + return undefined + } } /** Give the thread's claim back. A failed write is a warning. */ @@ -717,6 +874,8 @@ export abstract class SessionBase { ) if (hasFollowUps) return false return ( + // A question waits in this host's memory: its answer must come here. + this.questions.size === 0 && this.inputsInFlight === 0 && this.activeTurn === undefined && this.queue.length === 0 && @@ -753,7 +912,10 @@ export abstract class SessionBase { // ponytail: one release write per idle thread, also when this host // holds no claim (a sweep may hold one for it). Track the claim if the // writes matter. - void this.releaseClaim() + // An input that another host stored after the last drain runs now. + void this.releaseClaim().then(() => + this.draining ? undefined : this.drainInbox(), + ) for (const listener of [...this.idleListeners]) listener() } diff --git a/packages/ai-harness/src/session/harness-session.ts b/packages/ai-harness/src/session/harness-session.ts index 222d929100..debf9b23a5 100644 --- a/packages/ai-harness/src/session/harness-session.ts +++ b/packages/ai-harness/src/session/harness-session.ts @@ -63,6 +63,8 @@ export class HarnessSession< await this.openEntry() if (this.writer) { await this.recoverFromLog(this.writer) + // Inputs that waited for an owner that stopped. + void this.drainInbox() return } await this.loadUsage() diff --git a/packages/ai-harness/src/session/helpers.ts b/packages/ai-harness/src/session/helpers.ts index ac59a7415a..e5e64cbea1 100644 --- a/packages/ai-harness/src/session/helpers.ts +++ b/packages/ai-harness/src/session/helpers.ts @@ -693,3 +693,9 @@ export function tapFeed( close: () => feed.close(), } } + +/** `accept` passed the input to the host that owns the thread, or refused it for that. */ +export const isElsewhere = ( + admission: string, +): admission is 'forwarded' | 'owned_elsewhere' => + admission === 'forwarded' || admission === 'owned_elsewhere' diff --git a/packages/ai-harness/src/session/inputs.ts b/packages/ai-harness/src/session/inputs.ts index 293e0d17c8..479bcbf0cf 100644 --- a/packages/ai-harness/src/session/inputs.ts +++ b/packages/ai-harness/src/session/inputs.ts @@ -27,6 +27,19 @@ import type { } from '../types' import type { QueuedTurn, WaitingSteer } from './types' +/** + * How `accept` took an input. `forwarded`: another host owns the thread, and + * the input waits for it in `stores.inbox`. `owned_elsewhere`: another host + * owns the thread, and there is no inbox to pass the input on. + */ +export type Admission = + | 'new' + | 'duplicate' + | 'conflict' + | 'closed' + | 'forwarded' + | 'owned_elsewhere' + /** * A layer of {@link HarnessSession}: admitting inputs, receipts, and * settlements. @@ -34,6 +47,9 @@ import type { QueuedTurn, WaitingSteer } from './types' export abstract class SessionInputs< THarness extends AnyHarness = AnyHarness, > extends SessionBase { + /** Chat inputs that this host passed to the owner of the thread. */ + private readonly forwarded = new Set() + /** * Append host records to the session log, in one batch after the events * that wait. Only a durable host (with `stores.log`) has a log. A `type` @@ -317,7 +333,7 @@ export abstract class SessionInputs< inputId: string, input: HarnessInput, principal = this.principal, - ): Promise<'new' | 'duplicate' | 'conflict' | 'closed'> { + ): Promise { const sender = storedPrincipal(principal) const payload = inputKey(input, sender.principal) const local = this.admitted.get(inputId) @@ -332,11 +348,19 @@ export abstract class SessionInputs< if (this.closing && WORK_OPS.has(input.op)) return 'closed' this.admitted.set(inputId, payload) try { + // An agent run starts from the running turn, on the owner. + if (input.op !== 'agent' && (await this.isOwnedElsewhere())) { + return await this.forward(inputId, input, sender) + } // Before the input is stored, so a crash after the write leaves a claim // that a sweep finds. if (WORK_OPS.has(input.op)) { this.markBusy() await this.claimWrite + // Another host claimed the thread first, for example on a cold start. + if (this.writer && this.claimLost) { + return await this.forward(inputId, input, sender) + } } const at = Date.now() if (this.writer) { @@ -370,6 +394,33 @@ export abstract class SessionInputs< return 'new' } + /** + * Another host owns the thread: store the input in `stores.inbox` for it. + * Without an inbox the input is refused, so the owner's turn goes on. + */ + private async forward( + inputId: string, + input: HarnessInput, + sender: ReturnType, + ): Promise { + this.admitted.delete(inputId) + const inbox = this.persistence.stores.inbox + if (!inbox) return 'owned_elsewhere' + // `settled` waits for the owner to end it. + if (CHAT_OPS.has(input.op)) this.forwarded.add(inputId) + await inbox.append({ + inputId, + threadId: this.threadId, + input, + createdAt: Date.now(), + ...sender, + }) + // The owner can go idle before it reads the inbox again. Then no host + // owns the thread, and this one runs the input. + if (!(await this.isOwnedElsewhere())) void this.drainInbox() + return 'forwarded' + } + /** * Does `stores.runs` have a run with this id, on any thread? A client run * id must not reuse it. A store that fails counts as yes, so nothing runs. @@ -488,7 +539,11 @@ export abstract class SessionInputs< input.op === 'agent' || input.op === 'agentMessage' ) - return this.turnOperations.has(inputId) || this.receipts.has(inputId) + return ( + this.turnOperations.has(inputId) || + this.receipts.has(inputId) || + this.forwarded.has(inputId) + ) } /** Link a turn operation and its input. */ @@ -522,11 +577,12 @@ export abstract class SessionInputs< /** What a duplicate or a conflicting chat input gets back. */ protected duplicateReceipt( inputId: string, - admission: 'duplicate' | 'conflict' | 'closed', + admission: Exclude, ): Receipt { - if (admission !== 'duplicate') { + if (admission !== 'duplicate' && admission !== 'forwarded') { return { inputId, status: 'rejected', reason: admission } } + if (admission === 'forwarded') return { inputId, status: 'queued' } const kept = this.receipts.get(inputId) if (kept) return kept const operationId = this.writer?.state.inputs.get(inputId)?.operationId diff --git a/packages/ai-harness/src/session/requests.ts b/packages/ai-harness/src/session/requests.ts index b84c697287..c5ab759586 100644 --- a/packages/ai-harness/src/session/requests.ts +++ b/packages/ai-harness/src/session/requests.ts @@ -9,6 +9,7 @@ import { errorText, inputKey, SHUTDOWN_REASON, + isElsewhere, } from './helpers' import { SessionState } from './state' import type { @@ -31,6 +32,7 @@ import type { WaitingInput, } from '../types' import type { QueuedTurn, WaitingSteer } from './types' +import type { Admission } from './inputs' /** * A layer of {@link HarnessSession}: prompt, steer, follow-up, continue, reset, @@ -482,7 +484,8 @@ export abstract class SessionRequests< async cancel(operationId?: string): Promise { const inputId = createInputId() - await this.accept(inputId, { op: 'cancel', operationId }) + const admission = await this.accept(inputId, { op: 'cancel', operationId }) + if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) const target = operationId ? this.operations.get(operationId) : this.activeTurn @@ -580,7 +583,11 @@ export abstract class SessionRequests< */ async cancelInput(target: string): Promise { const inputId = createInputId() - await this.accept(inputId, { op: 'cancelInput', inputId: target }) + const admission = await this.accept(inputId, { + op: 'cancelInput', + inputId: target, + }) + if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) const taken = this.takeWaiting(target) if (!taken) return this.reject(inputId, 'not_waiting') await this.applyControl(inputId, { @@ -629,7 +636,12 @@ export abstract class SessionRequests< delivery: WaitingInput['delivery'] }): Promise { const inputId = createInputId() - await this.accept(inputId, { op: 'setDelivery', inputId: target, delivery }) + const admission = await this.accept(inputId, { + op: 'setDelivery', + inputId: target, + delivery, + }) + if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) const current = this.inputs().find((input) => input.inputId === target) if (current?.delivery !== delivery) { const taken = this.takeWaiting(target) @@ -906,7 +918,7 @@ export abstract class SessionRequests< protected answerDuplicate( operation: OperationImpl, inputId: string, - admission: 'duplicate' | 'conflict' | 'closed', + admission: Exclude, ) { const receipt = this.duplicateReceipt(inputId, admission) if (receipt.status === 'rejected') { diff --git a/packages/ai-harness/src/session/state.ts b/packages/ai-harness/src/session/state.ts index fdd3eb118f..624052e38d 100644 --- a/packages/ai-harness/src/session/state.ts +++ b/packages/ai-harness/src/session/state.ts @@ -29,6 +29,7 @@ import { storedPrincipal, SETTINGS, changeSettings, + isElsewhere, } from './helpers' import { SessionInputs } from './inputs' import type { OperationImpl } from '../operation' @@ -289,7 +290,8 @@ export abstract class SessionState< value: unknown }): Promise { const inputId = createInputId() - await this.accept(inputId, { op: 'config', key, value }) + const admission = await this.accept(inputId, { op: 'config', key, value }) + if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) const entry = this.sessionPlugins?.config.get(key) if (!entry) return this.reject(inputId, 'unknown_config') let checked: unknown @@ -618,11 +620,13 @@ export abstract class SessionState< // A secret (a key or a password) stays out of the inbox. A restart does // not replay an answer, so the inbox never needs it. const isSecret = this.questions.get(questionId)?.secret === true - await this.accept(inputId, { + // A host that does not have the question passes the value on to its owner. + const admission = await this.accept(inputId, { op: 'answer', questionId, value: isSecret ? '[secret]' : value, }) + if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) const question = this.questions.get(questionId) if (!question) return this.reject(inputId, 'unknown_question') let checked: unknown = value @@ -653,6 +657,9 @@ export abstract class SessionState< customEvent(HARNESS_EVENTS.questionAnswered, { questionId }), ) question.resolve(checked) + // The question no longer holds the thread. The resolved work, if any, + // marks it busy again first. + queueMicrotask(() => this.checkIdle()) return { inputId, status: 'accepted' } } @@ -989,6 +996,8 @@ export abstract class SessionState< // so a client that streams that operation sees it, and a cancel of it // rejects the question. The session stream has every operation's events. const operationId = owner ?? this.activeTurn?.id ?? 'session' + // The claim sends an answer that reaches another host here. + this.markBusy() return new Promise((resolve, reject) => { this.questions.set(questionId, { message: question.message, diff --git a/packages/ai-harness/src/session/turn.ts b/packages/ai-harness/src/session/turn.ts index 5b536a45d2..d6002c0dd2 100644 --- a/packages/ai-harness/src/session/turn.ts +++ b/packages/ai-harness/src/session/turn.ts @@ -13,7 +13,7 @@ import { mediaMiddleware } from '../media' import { OperationImpl } from '../operation' import { mountPlugins } from '../plugins' import { holdRunLease, truncationMiddleware } from '../resume' -import { HARNESS_EVENTS } from '../types' +import { HARNESS_EVENTS, InputRejectedError } from '../types' import { createInputId, customEvent, @@ -38,6 +38,7 @@ import { lastAssistantText, isRoutedBag, keepTurnTools, + isElsewhere, } from './helpers' import { SessionAgents } from './agents' import type { @@ -99,7 +100,21 @@ export abstract class SessionTurn< const inputId = createInputId() // The log keeps the input for replay, so a command must not take a secret // as its input. Ask a `secret` question instead. - await this.accept(inputId, { op: 'command', name, input }, principal) + const admission = await this.accept( + inputId, + { op: 'command', name, input }, + principal, + ) + // The owner runs it. Its result is in the owner's events. + if (isElsewhere(admission)) { + const receipt = this.duplicateReceipt(inputId, admission) + if (receipt.status === 'rejected') { + operation.fail('failed', new InputRejectedError(receipt)) + } else { + operation.finish('completed', undefined) + } + return + } const entry = this.sessionPlugins?.commands.get(name) if (!entry) { this.reject(inputId, 'unknown_command') diff --git a/packages/ai-harness/tests/durable-session.test.ts b/packages/ai-harness/tests/durable-session.test.ts index 4a65ac2d09..b432422232 100644 --- a/packages/ai-harness/tests/durable-session.test.ts +++ b/packages/ai-harness/tests/durable-session.test.ts @@ -209,7 +209,7 @@ describe('durable host stores', () => { stores: { log: memoryLogStore(), runs, messages }, } as never, }), - ).toThrow('Remove stores.messages and stores.inbox') + ).toThrow('Remove stores.messages.') }) it('refuses host records without a log, and harness record types', async () => { diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts new file mode 100644 index 0000000000..405ef4494a --- /dev/null +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -0,0 +1,168 @@ +import { describe, expect, it, vi } from 'vitest' +import { z } from 'zod' +import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' +import { + createHarnessHost, + defineCommand, + defineHarness, + definePlugin, +} from '../src' +import { after, gate, mockAdapter, text } from './helpers' +import type { AnyTextAdapter } from '@tanstack/ai' + +const THREAD = 't1' +const NAME = 'test/forwarding' + +/** Stores that several hosts share, as on a serverless platform. */ +function sharedStores(options: { inbox?: boolean } = {}) { + const { runs, metadata, workClaims, inbox } = memoryPersistence().stores + return { + stores: { + log: memoryLogStore(), + runs, + metadata, + workClaims, + ...(options.inbox === false ? {} : { inbox }), + }, + } +} + +const harnessOf = (adapter: AnyTextAdapter) => + defineHarness({ name: NAME, adapter }) + +describe('an input that reaches a host that does not own the thread', () => { + it('waits in the inbox for the owner, and the owner turn goes on', async () => { + const persistence = sharedStores() + const release = gate() + const { adapter, calls } = mockAdapter([ + after(release.opened, 'One.'), + () => text('Two.'), + ]) + const harness = harnessOf(adapter) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const turn = owner.prompt('Go.') + await turn.receipt + const receipt = await other.followUp('And then?') + expect(receipt.status).toBe('queued') + + release.open() + expect((await turn).text).toBe('One.') + // The owner applied the follow-up, and both hosts read how it ended. + // The owner reads the inbox within a second. + const settled = await vi.waitFor(() => owner.settled(receipt.inputId), { + timeout: 3_000, + }) + expect(settled.outcome).toBe('completed') + expect((await other.settled(receipt.inputId)).outcome).toBe('completed') + expect(calls).toHaveLength(2) + expect(await owner.transcript()).toMatchObject([ + { role: 'user', content: 'Go.' }, + { role: 'assistant', content: 'One.' }, + { role: 'user', content: 'And then?' }, + { role: 'assistant', content: 'Two.' }, + ]) + }) + + it('is refused without an inbox, and the owner turn goes on', async () => { + const persistence = sharedStores({ inbox: false }) + const release = gate() + const { adapter } = mockAdapter([after(release.opened, 'One.')]) + const harness = harnessOf(adapter) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const turn = owner.prompt('Go.') + await turn.receipt + expect(await other.steer('Faster.')).toMatchObject({ + status: 'rejected', + reason: 'owned_elsewhere', + }) + + release.open() + expect((await turn).text).toBe('One.') + }) + + it('runs on the next host that opens the thread after the owner stopped', async () => { + const persistence = sharedStores() + const { adapter, calls } = mockAdapter([() => text('Picked up.')]) + const harness = harnessOf(adapter) + // A host that stopped holds a claim that is about to expire. + await persistence.stores.workClaims.claim({ + threadId: THREAD, + harness: NAME, + ownerId: 'stopped-host', + until: Date.now() + 50, + }) + const first = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const receipt = await first.followUp('Still there?') + expect(receipt.status).toBe('queued') + expect(calls).toHaveLength(0) + + await new Promise((resolve) => setTimeout(resolve, 80)) + const host = createHarnessHost({ persistence }) + await host.resumePending({ harnesses: [harness] }) + const next = await host.open(harness, { threadId: THREAD }) + const settled = await vi.waitFor(() => next.settled(receipt.inputId), { + timeout: 3_000, + }) + expect(settled.outcome).toBe('completed') + expect(calls).toHaveLength(1) + }) + + it('answers a question that waits on another host', async () => { + const persistence = sharedStores() + const deploy = definePlugin({ + name: 'test/deploy', + setup: () => ({ + commands: { + deploy: defineCommand({ + description: 'Deploy after a confirmation', + run: async (_input, ctx) => { + const sure = await ctx.session.ask({ + message: 'Deploy to production?', + schema: z.boolean(), + }) + return sure ? 'deployed' : 'skipped' + }, + }), + }, + }), + }) + const { adapter } = mockAdapter([]) + const harness = defineHarness({ + name: NAME, + adapter, + plugins: () => [deploy], + }) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const running = owner.command('deploy') + await vi.waitFor(() => + expect(owner.snapshot().pendingQuestions).toHaveLength(1), + ) + const [question] = owner.snapshot().pendingQuestions + expect( + await other.answer({ questionId: question!.questionId, value: true }), + ).toMatchObject({ + status: 'queued', + }) + expect(await running).toBe('deployed') + }) +}) From 56a94b216d8e777795e33817ec7378e0e34791f4 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:06:53 -0700 Subject: [PATCH 05/15] fix(ai-harness): a host that opens a thread leaves a live owner's inputs alone Between storing an input and writing its run lease, the owner's input looked crashed to another host. Any request that opened the thread there (events, hydrate, a join) recovered and ran it, and the owner then failed with a log conflict. Recovery now skips a thread whose work claim another live host holds. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-recover-live-owner.md | 5 +++ .../ai-harness/src/session/harness-session.ts | 3 ++ packages/ai-harness/tests/forwarding.test.ts | 35 +++++++++++++++++++ 3 files changed, 43 insertions(+) create mode 100644 .changeset/harness-recover-live-owner.md diff --git a/.changeset/harness-recover-live-owner.md b/.changeset/harness-recover-live-owner.md new file mode 100644 index 0000000000..7cdc56db50 --- /dev/null +++ b/.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. diff --git a/packages/ai-harness/src/session/harness-session.ts b/packages/ai-harness/src/session/harness-session.ts index debf9b23a5..0ba1a3fa1a 100644 --- a/packages/ai-harness/src/session/harness-session.ts +++ b/packages/ai-harness/src/session/harness-session.ts @@ -466,6 +466,9 @@ export class HarnessSession< * again. */ protected async recoverFromLog(writer: LogWriter): Promise { + // A live owner runs its own inputs, also one that has no run lease yet. + // A sweep recovers them after its claim expires. + if (await this.isOwnedElsewhere()) return const runs = this.persistence.stores.runs const maxAttempts = this.harness.durability?.maxAttempts ?? 10 // True for an input that this session stored or took. Else the session diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 405ef4494a..401b712fb1 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -121,6 +121,41 @@ describe('an input that reaches a host that does not own the thread', () => { expect(calls).toHaveLength(1) }) + it('leaves the input of a live owner alone when another host opens the thread', async () => { + const persistence = sharedStores() + const { adapter, calls } = mockAdapter([() => text('One.')]) + const harness = harnessOf(adapter) + // The owner stores the input, then waits before it writes its run lease. + const leaseWrite = gate() + const shared = persistence.stores.runs + const owner = await createHarnessHost({ + persistence: { + stores: { + ...persistence.stores, + runs: Object.assign(Object.create(persistence.stores.runs), { + createOrResume: async ( + ...args: Parameters + ) => { + await leaseWrite.opened + return shared.createOrResume(...args) + }, + }), + }, + }, + }).open(harness, { threadId: THREAD }) + const turn = owner.prompt('Go.') + await turn.receipt + + // In that window, another host opens the thread and finds the input. + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + leaseWrite.open() + expect((await turn).text).toBe('One.') + expect(calls).toHaveLength(1) + await other.close() + }) + it('answers a question that waits on another host', async () => { const persistence = sharedStores() const deploy = definePlugin({ From d61fc79522743860acbed6df4d0f798ed2d8832d Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:07:35 -0700 Subject: [PATCH 06/15] test(e2e): an input on a second host waits for the owner of the thread Two hosts over one set of durable stores. A follow-up on the second host is queued and runs on the owner after its turn; without an inbox it is refused. In both, the owner's turn ends normally. Co-Authored-By: Claude Opus 5.5 (1M context) --- testing/e2e/src/routeTree.gen.ts | 105 ++++++++------ .../e2e/src/routes/api.harness-forwarding.ts | 136 ++++++++++++++++++ testing/e2e/tests/harness-forwarding.spec.ts | 48 +++++++ 3 files changed, 247 insertions(+), 42 deletions(-) create mode 100644 testing/e2e/src/routes/api.harness-forwarding.ts create mode 100644 testing/e2e/tests/harness-forwarding.spec.ts diff --git a/testing/e2e/src/routeTree.gen.ts b/testing/e2e/src/routeTree.gen.ts index d7fe95be71..37480762d9 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/testing/e2e/src/routeTree.gen.ts @@ -84,8 +84,6 @@ import { Route as ApiDurableDeliveryRouteImport } from './routes/api.durable-del import { Route as ApiDurableTakeoverRouteImport } from './routes/api.durable-takeover' import { Route as ApiEmbeddingRouteImport } from './routes/api.embedding' import { Route as ApiFakeTextRouteImport } from './routes/api.fake-text' -import { Route as ApiHarnessContinueRouteImport } from './routes/api.harness-continue' -import { Route as ApiHarnessRetryAfterRouteImport } from './routes/api.harness-retry-after' import { Route as ApiFileSourceWireRouteImport } from './routes/api.file-source-wire' import { Route as ApiForeignChunkEventsRouteImport } from './routes/api.foreign-chunk-events' import { Route as ApiForeignClientToolRouteImport } from './routes/api.foreign-client-tool' @@ -94,6 +92,9 @@ import { Route as ApiGeminiImageGaModelsRouteImport } from './routes/api.gemini- import { Route as ApiGeminiNativeImageWireRouteImport } from './routes/api.gemini-native-image-wire' import { Route as ApiGenerationPersistenceResumeRouteImport } from './routes/api.generation-persistence-resume' import { Route as ApiGenerationPersistenceServerRouteImport } from './routes/api.generation-persistence-server' +import { Route as ApiHarnessContinueRouteImport } from './routes/api.harness-continue' +import { Route as ApiHarnessForwardingRouteImport } from './routes/api.harness-forwarding' +import { Route as ApiHarnessRetryAfterRouteImport } from './routes/api.harness-retry-after' import { Route as ApiImageRouteImport } from './routes/api.image' import { Route as ApiInterruptsTestRouteImport } from './routes/api.interrupts-test' import { Route as ApiJoinRunClientToolRouteImport } from './routes/api.join-run-client-tool' @@ -572,16 +573,6 @@ const ApiFakeTextRoute = ApiFakeTextRouteImport.update({ path: '/api/fake-text', getParentRoute: () => rootRouteImport, } as any) -const ApiHarnessContinueRoute = ApiHarnessContinueRouteImport.update({ - id: '/api/harness-continue', - path: '/api/harness-continue', - getParentRoute: () => rootRouteImport, -} as any) -const ApiHarnessRetryAfterRoute = ApiHarnessRetryAfterRouteImport.update({ - id: '/api/harness-retry-after', - path: '/api/harness-retry-after', - getParentRoute: () => rootRouteImport, -} as any) const ApiFileSourceWireRoute = ApiFileSourceWireRouteImport.update({ id: '/api/file-source-wire', path: '/api/file-source-wire', @@ -625,6 +616,21 @@ const ApiGenerationPersistenceServerRoute = path: '/api/generation-persistence-server', getParentRoute: () => rootRouteImport, } as any) +const ApiHarnessContinueRoute = ApiHarnessContinueRouteImport.update({ + id: '/api/harness-continue', + path: '/api/harness-continue', + getParentRoute: () => rootRouteImport, +} as any) +const ApiHarnessForwardingRoute = ApiHarnessForwardingRouteImport.update({ + id: '/api/harness-forwarding', + path: '/api/harness-forwarding', + getParentRoute: () => rootRouteImport, +} as any) +const ApiHarnessRetryAfterRoute = ApiHarnessRetryAfterRouteImport.update({ + id: '/api/harness-retry-after', + path: '/api/harness-retry-after', + getParentRoute: () => rootRouteImport, +} as any) const ApiImageRoute = ApiImageRouteImport.update({ id: '/api/image', path: '/api/image', @@ -1147,8 +1153,6 @@ export interface FileRoutesByFullPath { '/api/durable-takeover': typeof ApiDurableTakeoverRoute '/api/embedding': typeof ApiEmbeddingRoute '/api/fake-text': typeof ApiFakeTextRoute - '/api/harness-continue': typeof ApiHarnessContinueRoute - '/api/harness-retry-after': typeof ApiHarnessRetryAfterRoute '/api/file-source-wire': typeof ApiFileSourceWireRoute '/api/foreign-chunk-events': typeof ApiForeignChunkEventsRoute '/api/foreign-client-tool': typeof ApiForeignClientToolRoute @@ -1157,6 +1161,9 @@ export interface FileRoutesByFullPath { '/api/gemini-native-image-wire': typeof ApiGeminiNativeImageWireRoute '/api/generation-persistence-resume': typeof ApiGenerationPersistenceResumeRoute '/api/generation-persistence-server': typeof ApiGenerationPersistenceServerRoute + '/api/harness-continue': typeof ApiHarnessContinueRoute + '/api/harness-forwarding': typeof ApiHarnessForwardingRoute + '/api/harness-retry-after': typeof ApiHarnessRetryAfterRoute '/api/image': typeof ApiImageRouteWithChildren '/api/interrupts-test': typeof ApiInterruptsTestRoute '/api/join-run-client-tool': typeof ApiJoinRunClientToolRoute @@ -1319,8 +1326,6 @@ export interface FileRoutesByTo { '/api/durable-takeover': typeof ApiDurableTakeoverRoute '/api/embedding': typeof ApiEmbeddingRoute '/api/fake-text': typeof ApiFakeTextRoute - '/api/harness-continue': typeof ApiHarnessContinueRoute - '/api/harness-retry-after': typeof ApiHarnessRetryAfterRoute '/api/file-source-wire': typeof ApiFileSourceWireRoute '/api/foreign-chunk-events': typeof ApiForeignChunkEventsRoute '/api/foreign-client-tool': typeof ApiForeignClientToolRoute @@ -1329,6 +1334,9 @@ export interface FileRoutesByTo { '/api/gemini-native-image-wire': typeof ApiGeminiNativeImageWireRoute '/api/generation-persistence-resume': typeof ApiGenerationPersistenceResumeRoute '/api/generation-persistence-server': typeof ApiGenerationPersistenceServerRoute + '/api/harness-continue': typeof ApiHarnessContinueRoute + '/api/harness-forwarding': typeof ApiHarnessForwardingRoute + '/api/harness-retry-after': typeof ApiHarnessRetryAfterRoute '/api/image': typeof ApiImageRouteWithChildren '/api/interrupts-test': typeof ApiInterruptsTestRoute '/api/join-run-client-tool': typeof ApiJoinRunClientToolRoute @@ -1492,8 +1500,6 @@ export interface FileRoutesById { '/api/durable-takeover': typeof ApiDurableTakeoverRoute '/api/embedding': typeof ApiEmbeddingRoute '/api/fake-text': typeof ApiFakeTextRoute - '/api/harness-continue': typeof ApiHarnessContinueRoute - '/api/harness-retry-after': typeof ApiHarnessRetryAfterRoute '/api/file-source-wire': typeof ApiFileSourceWireRoute '/api/foreign-chunk-events': typeof ApiForeignChunkEventsRoute '/api/foreign-client-tool': typeof ApiForeignClientToolRoute @@ -1502,6 +1508,9 @@ export interface FileRoutesById { '/api/gemini-native-image-wire': typeof ApiGeminiNativeImageWireRoute '/api/generation-persistence-resume': typeof ApiGenerationPersistenceResumeRoute '/api/generation-persistence-server': typeof ApiGenerationPersistenceServerRoute + '/api/harness-continue': typeof ApiHarnessContinueRoute + '/api/harness-forwarding': typeof ApiHarnessForwardingRoute + '/api/harness-retry-after': typeof ApiHarnessRetryAfterRoute '/api/image': typeof ApiImageRouteWithChildren '/api/interrupts-test': typeof ApiInterruptsTestRoute '/api/join-run-client-tool': typeof ApiJoinRunClientToolRoute @@ -1666,8 +1675,6 @@ export interface FileRouteTypes { | '/api/durable-takeover' | '/api/embedding' | '/api/fake-text' - | '/api/harness-continue' - | '/api/harness-retry-after' | '/api/file-source-wire' | '/api/foreign-chunk-events' | '/api/foreign-client-tool' @@ -1676,6 +1683,9 @@ export interface FileRouteTypes { | '/api/gemini-native-image-wire' | '/api/generation-persistence-resume' | '/api/generation-persistence-server' + | '/api/harness-continue' + | '/api/harness-forwarding' + | '/api/harness-retry-after' | '/api/image' | '/api/interrupts-test' | '/api/join-run-client-tool' @@ -1838,8 +1848,6 @@ export interface FileRouteTypes { | '/api/durable-takeover' | '/api/embedding' | '/api/fake-text' - | '/api/harness-continue' - | '/api/harness-retry-after' | '/api/file-source-wire' | '/api/foreign-chunk-events' | '/api/foreign-client-tool' @@ -1848,6 +1856,9 @@ export interface FileRouteTypes { | '/api/gemini-native-image-wire' | '/api/generation-persistence-resume' | '/api/generation-persistence-server' + | '/api/harness-continue' + | '/api/harness-forwarding' + | '/api/harness-retry-after' | '/api/image' | '/api/interrupts-test' | '/api/join-run-client-tool' @@ -2010,8 +2021,6 @@ export interface FileRouteTypes { | '/api/durable-takeover' | '/api/embedding' | '/api/fake-text' - | '/api/harness-continue' - | '/api/harness-retry-after' | '/api/file-source-wire' | '/api/foreign-chunk-events' | '/api/foreign-client-tool' @@ -2020,6 +2029,9 @@ export interface FileRouteTypes { | '/api/gemini-native-image-wire' | '/api/generation-persistence-resume' | '/api/generation-persistence-server' + | '/api/harness-continue' + | '/api/harness-forwarding' + | '/api/harness-retry-after' | '/api/image' | '/api/interrupts-test' | '/api/join-run-client-tool' @@ -2183,8 +2195,6 @@ export interface RootRouteChildren { ApiDurableTakeoverRoute: typeof ApiDurableTakeoverRoute ApiEmbeddingRoute: typeof ApiEmbeddingRoute ApiFakeTextRoute: typeof ApiFakeTextRoute - ApiHarnessContinueRoute: typeof ApiHarnessContinueRoute - ApiHarnessRetryAfterRoute: typeof ApiHarnessRetryAfterRoute ApiFileSourceWireRoute: typeof ApiFileSourceWireRoute ApiForeignChunkEventsRoute: typeof ApiForeignChunkEventsRoute ApiForeignClientToolRoute: typeof ApiForeignClientToolRoute @@ -2193,6 +2203,9 @@ export interface RootRouteChildren { ApiGeminiNativeImageWireRoute: typeof ApiGeminiNativeImageWireRoute ApiGenerationPersistenceResumeRoute: typeof ApiGenerationPersistenceResumeRoute ApiGenerationPersistenceServerRoute: typeof ApiGenerationPersistenceServerRoute + ApiHarnessContinueRoute: typeof ApiHarnessContinueRoute + ApiHarnessForwardingRoute: typeof ApiHarnessForwardingRoute + ApiHarnessRetryAfterRoute: typeof ApiHarnessRetryAfterRoute ApiImageRoute: typeof ApiImageRouteWithChildren ApiInterruptsTestRoute: typeof ApiInterruptsTestRoute ApiJoinRunClientToolRoute: typeof ApiJoinRunClientToolRoute @@ -2803,20 +2816,6 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ApiFakeTextRouteImport parentRoute: typeof rootRouteImport } - '/api/harness-continue': { - id: '/api/harness-continue' - path: '/api/harness-continue' - fullPath: '/api/harness-continue' - preLoaderRoute: typeof ApiHarnessContinueRouteImport - parentRoute: typeof rootRouteImport - } - '/api/harness-retry-after': { - id: '/api/harness-retry-after' - path: '/api/harness-retry-after' - fullPath: '/api/harness-retry-after' - preLoaderRoute: typeof ApiHarnessRetryAfterRouteImport - parentRoute: typeof rootRouteImport - } '/api/file-source-wire': { id: '/api/file-source-wire' path: '/api/file-source-wire' @@ -2873,6 +2872,27 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ApiGenerationPersistenceServerRouteImport parentRoute: typeof rootRouteImport } + '/api/harness-continue': { + id: '/api/harness-continue' + path: '/api/harness-continue' + fullPath: '/api/harness-continue' + preLoaderRoute: typeof ApiHarnessContinueRouteImport + parentRoute: typeof rootRouteImport + } + '/api/harness-forwarding': { + id: '/api/harness-forwarding' + path: '/api/harness-forwarding' + fullPath: '/api/harness-forwarding' + preLoaderRoute: typeof ApiHarnessForwardingRouteImport + parentRoute: typeof rootRouteImport + } + '/api/harness-retry-after': { + id: '/api/harness-retry-after' + path: '/api/harness-retry-after' + fullPath: '/api/harness-retry-after' + preLoaderRoute: typeof ApiHarnessRetryAfterRouteImport + parentRoute: typeof rootRouteImport + } '/api/image': { id: '/api/image' path: '/api/image' @@ -3609,8 +3629,6 @@ const rootRouteChildren: RootRouteChildren = { ApiDurableTakeoverRoute: ApiDurableTakeoverRoute, ApiEmbeddingRoute: ApiEmbeddingRoute, ApiFakeTextRoute: ApiFakeTextRoute, - ApiHarnessContinueRoute: ApiHarnessContinueRoute, - ApiHarnessRetryAfterRoute: ApiHarnessRetryAfterRoute, ApiFileSourceWireRoute: ApiFileSourceWireRoute, ApiForeignChunkEventsRoute: ApiForeignChunkEventsRoute, ApiForeignClientToolRoute: ApiForeignClientToolRoute, @@ -3619,6 +3637,9 @@ const rootRouteChildren: RootRouteChildren = { ApiGeminiNativeImageWireRoute: ApiGeminiNativeImageWireRoute, ApiGenerationPersistenceResumeRoute: ApiGenerationPersistenceResumeRoute, ApiGenerationPersistenceServerRoute: ApiGenerationPersistenceServerRoute, + ApiHarnessContinueRoute: ApiHarnessContinueRoute, + ApiHarnessForwardingRoute: ApiHarnessForwardingRoute, + ApiHarnessRetryAfterRoute: ApiHarnessRetryAfterRoute, ApiImageRoute: ApiImageRouteWithChildren, ApiInterruptsTestRoute: ApiInterruptsTestRoute, ApiJoinRunClientToolRoute: ApiJoinRunClientToolRoute, diff --git a/testing/e2e/src/routes/api.harness-forwarding.ts b/testing/e2e/src/routes/api.harness-forwarding.ts new file mode 100644 index 0000000000..21c30f036a --- /dev/null +++ b/testing/e2e/src/routes/api.harness-forwarding.ts @@ -0,0 +1,136 @@ +import { createFileRoute } from '@tanstack/react-router' +import { createAnthropicChat } from '@tanstack/ai-anthropic' +import { createHarnessHost, defineHarness } from '@tanstack/ai-harness' +import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' + +/** + * Two hosts share one set of durable stores, as two instances of a + * serverless app do. The first host runs a turn. While it runs, the second + * host gets a follow-up for the same thread. + * + * - `?inbox=1`: the follow-up waits in `stores.inbox`, and the owner runs it + * after its turn. + * - Without it: the second host refuses the follow-up. + * + * In both, the owner's turn ends normally. A custom `fetch` answers with + * canned Anthropic streams and holds the first answer until the follow-up + * is in, so no aimock fixture is needed. + */ + +const DUMMY_KEY = 'sk-ant-e2e-test-dummy-key' +const MODEL = 'claude-sonnet-5-5' + +const reply = (id: string, text: string) => + [ + { + type: 'message_start', + message: { + id, + type: 'message', + role: 'assistant', + content: [], + model: MODEL, + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 5, output_tokens: 0 }, + }, + }, + { + type: 'content_block_start', + index: 0, + content_block: { type: 'text', text: '' }, + }, + { + type: 'content_block_delta', + index: 0, + delta: { type: 'text_delta', text }, + }, + { type: 'content_block_stop', index: 0 }, + { + type: 'message_delta', + delta: { stop_reason: 'end_turn', stop_sequence: null }, + usage: { output_tokens: 1 }, + }, + { type: 'message_stop' }, + ] + .map((event) => `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`) + .join('') + +/** The text of each message of the transcript. */ +const textsOf = (messages: Array<{ role: string; content: unknown }>) => + messages.map((message) => ({ + role: message.role, + text: typeof message.content === 'string' ? message.content : '', + })) + +export const Route = createFileRoute('/api/harness-forwarding')({ + server: { + handlers: { + POST: async ({ request }) => { + const withInbox = new URL(request.url).searchParams.get('inbox') === '1' + let release!: () => void + const released = new Promise((resolve) => { + release = resolve + }) + let calls = 0 + const scriptedFetch: typeof fetch = async () => { + calls += 1 + if (calls === 1) await released + const text = calls === 1 ? 'One.' : 'Two.' + return new Response(reply(`msg_forwarding_${calls}`, text), { + headers: { 'Content-Type': 'text/event-stream' }, + }) + } + + const { runs, metadata, workClaims, inbox } = memoryPersistence().stores + const persistence = { + stores: { + log: memoryLogStore(), + runs, + metadata, + workClaims, + ...(withInbox ? { inbox } : {}), + }, + } + const harness = defineHarness({ + name: 'e2e/harness-forwarding', + adapter: createAnthropicChat(MODEL, DUMMY_KEY, { + fetch: scriptedFetch, + }), + }) + const ownerHost = createHarnessHost({ persistence }) + const otherHost = createHarnessHost({ persistence }) + try { + const threadId = 'e2e-harness-forwarding' + const owner = await ownerHost.open(harness, { threadId }) + const other = await otherHost.open(harness, { threadId }) + + const turn = owner.prompt('[harness-forwarding] go') + await turn.receipt + const receipt = await other.followUp('[harness-forwarding] and then') + release() + const first = await turn + + let followUp: string | undefined + if (receipt.status === 'queued') { + followUp = (await other.settled(receipt.inputId)).outcome + } + return Response.json({ + receipt: { status: receipt.status, reason: receipt.reason }, + first: first.text, + followUp, + calls, + transcript: textsOf(await owner.transcript()), + }) + } catch (error) { + return Response.json({ + error: error instanceof Error ? error.message : String(error), + }) + } finally { + await ownerHost.close() + await otherHost.close() + } + }, + }, + }, +}) diff --git a/testing/e2e/tests/harness-forwarding.spec.ts b/testing/e2e/tests/harness-forwarding.spec.ts new file mode 100644 index 0000000000..a6260857ba --- /dev/null +++ b/testing/e2e/tests/harness-forwarding.spec.ts @@ -0,0 +1,48 @@ +import { test, expect } from './fixtures' + +interface Result { + error?: string + receipt: { status: string; reason?: string } + first: string + followUp?: string + calls: number + transcript: Array<{ role: string; text: string }> +} + +test.describe('harness: an input on a host that does not own the thread', () => { + test('waits in the inbox for the owner, and the owner turn ends normally', async ({ + request, + }) => { + const res = await request.post('/api/harness-forwarding?inbox=1') + expect(res.ok()).toBe(true) + const result: Result = await res.json() + expect(result.error ?? null).toBeNull() + + expect(result.receipt.status).toBe('queued') + expect(result.first).toBe('One.') + expect(result.followUp).toBe('completed') + expect(result.calls).toBe(2) + expect(result.transcript).toEqual([ + { role: 'user', text: '[harness-forwarding] go' }, + { role: 'assistant', text: 'One.' }, + { role: 'user', text: '[harness-forwarding] and then' }, + { role: 'assistant', text: 'Two.' }, + ]) + }) + + test('is refused without an inbox, and the owner turn ends normally', async ({ + request, + }) => { + const res = await request.post('/api/harness-forwarding') + expect(res.ok()).toBe(true) + const result: Result = await res.json() + expect(result.error ?? null).toBeNull() + + expect(result.receipt).toEqual({ + status: 'rejected', + reason: 'owned_elsewhere', + }) + expect(result.first).toBe('One.') + expect(result.calls).toBe(1) + }) +}) From be8eba188bb774d0ba382fa65f62d20559fab834 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:16:07 -0700 Subject: [PATCH 07/15] fix(ai-harness): a forwarded turn runs under the run id of the host that took it A POST run on a host that does not own the thread forwards the prompt. The owner now runs it as the request's AG-UI run id, so the operation on the forwarding host streams the owner's events from the shared log, and useChat sees the turn on whichever instance it reached. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/ai-harness/src/session/base.ts | 5 +++- packages/ai-harness/src/session/inputs.ts | 10 +++++-- packages/ai-harness/src/session/requests.ts | 2 +- packages/ai-harness/tests/forwarding.test.ts | 30 ++++++++++++++++++++ 4 files changed, 42 insertions(+), 5 deletions(-) diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index 2aa1040d42..b8e5a312e0 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -770,13 +770,16 @@ export abstract class SessionBase { ) } switch (input.op) { - case 'prompt': + case 'prompt': { + const { runId } = entry.input as { runId?: unknown } return settle( session.prompt(input.message, { ...sent, ...(input.busy ? { busy: input.busy } : {}), + ...(typeof runId === 'string' ? { runId } : {}), }), ) + } case 'continue': return settle(session.continue(sent)) case 'steer': diff --git a/packages/ai-harness/src/session/inputs.ts b/packages/ai-harness/src/session/inputs.ts index 479bcbf0cf..e8875377e9 100644 --- a/packages/ai-harness/src/session/inputs.ts +++ b/packages/ai-harness/src/session/inputs.ts @@ -333,6 +333,8 @@ export abstract class SessionInputs< inputId: string, input: HarnessInput, principal = this.principal, + /** The AG-UI run id of a turn, so a forwarded turn runs as it. */ + runId?: string, ): Promise { const sender = storedPrincipal(principal) const payload = inputKey(input, sender.principal) @@ -350,7 +352,7 @@ export abstract class SessionInputs< try { // An agent run starts from the running turn, on the owner. if (input.op !== 'agent' && (await this.isOwnedElsewhere())) { - return await this.forward(inputId, input, sender) + return await this.forward(inputId, input, sender, runId) } // Before the input is stored, so a crash after the write leaves a claim // that a sweep finds. @@ -359,7 +361,7 @@ export abstract class SessionInputs< await this.claimWrite // Another host claimed the thread first, for example on a cold start. if (this.writer && this.claimLost) { - return await this.forward(inputId, input, sender) + return await this.forward(inputId, input, sender, runId) } } const at = Date.now() @@ -402,6 +404,7 @@ export abstract class SessionInputs< inputId: string, input: HarnessInput, sender: ReturnType, + runId: string | undefined, ): Promise { this.admitted.delete(inputId) const inbox = this.persistence.stores.inbox @@ -411,7 +414,8 @@ export abstract class SessionInputs< await inbox.append({ inputId, threadId: this.threadId, - input, + // The owner runs the turn as `runId`, so this host streams it from the log. + input: runId === undefined ? input : { ...input, runId }, createdAt: Date.now(), ...sender, }) diff --git a/packages/ai-harness/src/session/requests.ts b/packages/ai-harness/src/session/requests.ts index c5ab759586..d2abf91f18 100644 --- a/packages/ai-harness/src/session/requests.ts +++ b/packages/ai-harness/src/session/requests.ts @@ -114,7 +114,7 @@ export abstract class SessionRequests< const operation = this.createTurnOperation(runId) this.bindTurn(inputId, operation) void this.admitting(() => - this.accept(inputId, input, principal).then( + this.accept(inputId, input, principal, runId).then( async (admission) => { if (admission !== 'new') { this.answerDuplicate(operation, inputId, admission) diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 401b712fb1..5c6e8115de 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import { z } from 'zod' +import { EventType } from '@tanstack/ai' import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' import { createHarnessHost, @@ -69,6 +70,35 @@ describe('an input that reaches a host that does not own the thread', () => { ]) }) + it('streams a forwarded turn under its run id on the host that took it', async () => { + const persistence = sharedStores() + const release = gate() + const { adapter } = mockAdapter([ + after(release.opened, 'One.'), + () => text('Two.'), + ]) + const harness = harnessOf(adapter) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const first = owner.prompt('Go.') + await first.receipt + // A `POST run` on the other host: the turn runs on the owner as `run-2`. + const second = other.prompt('And then?', { runId: 'run-2' }) + expect((await second.receipt).status).toBe('queued') + release.open() + const texts: Array = [] + for await (const event of second.stream()) { + if (event.type === EventType.TEXT_MESSAGE_CONTENT) texts.push(event.delta) + } + expect(texts.join('')).toBe('Two.') + expect((await second).text).toBe('Two.') + }) + it('is refused without an inbox, and the owner turn goes on', async () => { const persistence = sharedStores({ inbox: false }) const release = gate() From 858080102e8ae917f7a9299b913307a89485988b Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:17:23 -0700 Subject: [PATCH 08/15] docs(ai-persistence): the inbox and work claim store shapes, and content-hash revisions Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/persistence/store-reference.md | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/docs/persistence/store-reference.md b/docs/persistence/store-reference.md index afb34c9fa5..9ed9ac977f 100644 --- a/docs/persistence/store-reference.md +++ b/docs/persistence/store-reference.md @@ -627,9 +627,9 @@ interface InboxEntry { interface InboxStore { append(entry: Omit): Promise listPending(threadId: string): Promise> - markApplied(entry: { inputId: string; operationId: string }): Promise - markRejected(entry: { inputId: string; reason: string }): Promise - get(inputId: string): Promise + markApplied(entry: { threadId: string; inputId: string; operationId: string }): Promise + markRejected(entry: { threadId: string; inputId: string; reason: string }): Promise + get(entry: { threadId: string; inputId: string }): Promise } ``` @@ -640,7 +640,8 @@ 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. In SQL, make `input_id` the primary key and write with `INSERT ... ON CONFLICT DO NOTHING`, then read the row back. @@ -855,6 +856,7 @@ interface WorkClaimStore { until: number // epoch ms }): Promise release(claim: { threadId: string; ownerId: string }): Promise + get(threadId: string): Promise<{ ownerId: string; until: number } | null> listExpired(options: { now: number // epoch ms limit?: number @@ -870,6 +872,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 +1000,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 From f4e73f4b70dcdc11f714d2e7fbdec637077f01e1 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:39:29 -0700 Subject: [PATCH 09/15] docs(ai-harness): inputs on another instance, rolling deploys, and setup() keys Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/config.json | 3 ++- docs/harness/durable-sessions.md | 33 +++++++++++++++++++++++++++++++- docs/harness/plugins.md | 2 ++ 3 files changed, 36 insertions(+), 2 deletions(-) diff --git a/docs/config.json b/docs/config.json index 111ac54010..33995f8714 100644 --- a/docs/config.json +++ b/docs/config.json @@ -938,7 +938,8 @@ { "label": "Durable sessions", "to": "harness/durable-sessions", - "addedAt": "2026-10-13" + "addedAt": "2026-10-13", + "updatedAt": "2026-10-14" }, { "label": "List, rename, and fork sessions", diff --git a/docs/harness/durable-sessions.md b/docs/harness/durable-sessions.md index 1434480a07..0b34171874 100644 --- a/docs/harness/durable-sessions.md +++ b/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,8 @@ 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). + ### What a sweep does `host.resumePending({ harnesses, close, limit })`: @@ -305,7 +309,33 @@ 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. + +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 +403,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. diff --git a/docs/harness/plugins.md b/docs/harness/plugins.md index 01724f30ee..c5b951761c 100644 --- a/docs/harness/plugins.md +++ b/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. From 98e8d100085ecc0afd28300ca18acb2c57c492bd Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 13:21:07 -0700 Subject: [PATCH 10/15] fix(ai-harness): a host that passed an input on runs it if the owner stops The forwarding host reads the inbox every second until its inputs ran. When the owner stops, its claim expires, and the forwarding host claims the thread and runs them, so an input never waits for a sweep. The claim test from #1779 now covers both setups: refused without an inbox, and run once the other claim expires with one. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/ai-harness/src/session/base.ts | 24 +++++++++++ packages/ai-harness/src/session/inputs.ts | 5 ++- packages/ai-harness/tests/work-claims.test.ts | 42 +++++++++++++------ 3 files changed, 57 insertions(+), 14 deletions(-) diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index b8e5a312e0..fb7f2a791f 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -189,6 +189,8 @@ export abstract class SessionBase { protected claimLost = false /** Drains `stores.inbox` while this host owns the thread. */ protected inboxTimer: ReturnType | undefined + /** Reads the inbox until the inputs this host left there have run. */ + protected inboxWatch: ReturnType | undefined private draining: Promise | undefined /** Called each time the thread goes idle. See `onIdle`. */ protected readonly idleListeners = new Set<() => void>() @@ -706,6 +708,28 @@ export abstract class SessionBase { ) } + /** + * This host left an input for the owner: read the inbox every second until + * it is empty. If the owner stops, its claim expires, and this host claims + * the thread and runs the input. + */ + protected watchInbox(): void { + const inbox = this.persistence.stores.inbox + if (this.inboxWatch || !inbox) return + const stop = () => { + clearInterval(timer) + this.inboxWatch = undefined + } + const timer = setInterval(async () => { + if (this.closing) return stop() + const pending = await inbox.listPending(this.threadId).catch(() => null) + if (pending?.length === 0) return stop() + void this.drainInbox() + }, INBOX_POLL_MS) + if (typeof timer === 'object' && 'unref' in timer) timer.unref() + this.inboxWatch = timer + } + /** * Apply the inputs that other hosts left in `stores.inbox` for this * thread, in order. Only the host that holds the claim does: it claims the diff --git a/packages/ai-harness/src/session/inputs.ts b/packages/ai-harness/src/session/inputs.ts index e8875377e9..bdf73e83c7 100644 --- a/packages/ai-harness/src/session/inputs.ts +++ b/packages/ai-harness/src/session/inputs.ts @@ -419,9 +419,10 @@ export abstract class SessionInputs< createdAt: Date.now(), ...sender, }) - // The owner can go idle before it reads the inbox again. Then no host - // owns the thread, and this one runs the input. + // The owner can go idle before it reads the inbox again, or stop. Then no + // host owns the thread, and this one runs the input. if (!(await this.isOwnedElsewhere())) void this.drainInbox() + this.watchInbox() return 'forwarded' } diff --git a/packages/ai-harness/tests/work-claims.test.ts b/packages/ai-harness/tests/work-claims.test.ts index 735fbfbe09..21b025c941 100644 --- a/packages/ai-harness/tests/work-claims.test.ts +++ b/packages/ai-harness/tests/work-claims.test.ts @@ -127,32 +127,50 @@ describe('work claims while a session runs', () => { await host.close() }) - it('claims the thread once another host claim expires, while the turn runs', async () => { + it('refuses an input while another host claim is live, without an inbox', async () => { const persistence = durablePersistence() + await persistence.stores.workClaims.claim({ + threadId: THREAD, + harness: NAME, + ownerId: 'other-host', + until: Date.now() + 60_000, + }) + const { adapter, calls } = mockAdapter([() => text('Done.')]) + const host = createHarnessHost({ persistence }) + const session = await host.open(defineHarness({ name: NAME, adapter }), { + threadId: THREAD, + }) + + expect(await session.prompt('Go.').receipt).toMatchObject({ + status: 'rejected', + reason: 'owned_elsewhere', + }) + expect(calls).toHaveLength(0) + await host.close() + }) + + it('runs a queued input once another host claim expires', async () => { + const persistence = durablePersistence() + const { inbox } = memoryPersistence().stores const claims = persistence.stores.workClaims await claims.claim({ threadId: THREAD, harness: NAME, ownerId: 'other-host', - until: Date.now() + 100, + until: Date.now() + 300, }) - const release = gate() - const { adapter } = mockAdapter([after(release.opened, 'Done.')]) + const { adapter } = mockAdapter([() => text('Done.')]) const host = createHarnessHost({ - persistence, - lease: { ttlMs: 60_000, renewMs: 20 }, + persistence: { stores: { ...persistence.stores, inbox } }, }) const session = await host.open(defineHarness({ name: NAME, adapter }), { threadId: THREAD, }) const turn = session.prompt('Go.') - await turn.receipt - await new Promise((resolve) => setTimeout(resolve, 200)) - expect(await otherHostCanClaim(claims)).toBe(false) - - release.open() - await turn + expect((await turn.receipt).status).toBe('queued') + // The other host stops, so its claim expires, and this host runs the input. + expect((await turn).text).toBe('Done.') await host.close() }) From d62f10ff031f410bfa302c90164bc21e57b72728 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 15 Oct 2026 13:00:08 +0200 Subject: [PATCH 11/15] fix(ai-harness): run a forwarded input safely when the owner stops - A host that passed a POST run on runs it itself, on the operation its caller holds, when the owner stops. Before, its own run id refused it as a conflict. - An inbox entry is applied only after the owner stored the turn. Before, a failed log write lost a forwarded prompt. - A host that takes a thread from the inbox recovers the stopped owner's inputs first, as an open does. Before, they were never run. --- packages/ai-harness/src/session/base.ts | 20 +++- .../ai-harness/src/session/harness-session.ts | 5 +- packages/ai-harness/src/session/inputs.ts | 2 +- packages/ai-harness/src/session/requests.ts | 8 +- packages/ai-harness/tests/forwarding.test.ts | 103 +++++++++++++++++- 5 files changed, 132 insertions(+), 6 deletions(-) diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index fb7f2a791f..30848bd3e0 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -191,6 +191,8 @@ export abstract class SessionBase { protected inboxTimer: ReturnType | undefined /** Reads the inbox until the inputs this host left there have run. */ protected inboxWatch: ReturnType | undefined + /** The open skipped recovery: another host owned the thread then. */ + protected recoveryDeferred = false private draining: Promise | undefined /** Called each time the thread goes idle. See `onIdle`. */ protected readonly idleListeners = new Set<() => void>() @@ -757,8 +759,16 @@ export abstract class SessionBase { until: Date.now() + (this.lease?.ttlMs ?? LEASE.ttlMs), }) if (!isHeld) return + // This host takes the thread from an owner that stopped. Its inputs + // run first, as on an open. + if (this.recoveryDeferred) { + await this.recoverFromLog(this.writer) + this.recoveryDeferred = false + } for (const entry of pending) { await this.applyForwarded(entry) + // The input was not stored. It waits for the next host. + if (this.closing || this.logFailure) return await inbox.markApplied({ threadId: this.threadId, inputId: entry.inputId, @@ -793,10 +803,15 @@ export abstract class SessionBase { () => {}, ) } + // A turn is stored when its receipt comes. Only then is the entry applied. + const stored = (operation: Operation) => { + settle(operation) + return operation.receipt + } switch (input.op) { case 'prompt': { const { runId } = entry.input as { runId?: unknown } - return settle( + return stored( session.prompt(input.message, { ...sent, ...(input.busy ? { busy: input.busy } : {}), @@ -805,7 +820,7 @@ export abstract class SessionBase { ) } case 'continue': - return settle(session.continue(sent)) + return stored(session.continue(sent)) case 'steer': return session.steer(input.message, sent) case 'followUp': @@ -1055,6 +1070,7 @@ export abstract class SessionBase { operationId?: string, ): PluginSessionApi protected abstract resumeSignIns(): Promise + protected abstract recoverFromLog(writer: LogWriter): Promise protected abstract runAgent( target: string | AnyAgent, input: unknown, diff --git a/packages/ai-harness/src/session/harness-session.ts b/packages/ai-harness/src/session/harness-session.ts index 0ba1a3fa1a..2d73a3f607 100644 --- a/packages/ai-harness/src/session/harness-session.ts +++ b/packages/ai-harness/src/session/harness-session.ts @@ -468,7 +468,10 @@ export class HarnessSession< protected async recoverFromLog(writer: LogWriter): Promise { // A live owner runs its own inputs, also one that has no run lease yet. // A sweep recovers them after its claim expires. - if (await this.isOwnedElsewhere()) return + if (await this.isOwnedElsewhere()) { + this.recoveryDeferred = true + return + } const runs = this.persistence.stores.runs const maxAttempts = this.harness.durability?.maxAttempts ?? 10 // True for an input that this session stored or took. Else the session diff --git a/packages/ai-harness/src/session/inputs.ts b/packages/ai-harness/src/session/inputs.ts index bdf73e83c7..fb568862f3 100644 --- a/packages/ai-harness/src/session/inputs.ts +++ b/packages/ai-harness/src/session/inputs.ts @@ -48,7 +48,7 @@ export abstract class SessionInputs< THarness extends AnyHarness = AnyHarness, > extends SessionBase { /** Chat inputs that this host passed to the owner of the thread. */ - private readonly forwarded = new Set() + protected readonly forwarded = new Set() /** * Append host records to the session log, in one batch after the events diff --git a/packages/ai-harness/src/session/requests.ts b/packages/ai-harness/src/session/requests.ts index d2abf91f18..1e8273e8b0 100644 --- a/packages/ai-harness/src/session/requests.ts +++ b/packages/ai-harness/src/session/requests.ts @@ -102,8 +102,14 @@ export abstract class SessionRequests< const known = this.knownTurn(inputId, input, principal) if (known) return known const runId = options?.runId + // A prompt that this host passed on comes back from the inbox when its + // owner stopped. It runs on the operation that its caller holds. + const passed = this.forwarded.has(inputId) + ? this.turnOperations.get(inputId) + : undefined // A run id must not take the id of another operation. if ( + !passed && runId !== undefined && (this.operations.has(runId) || this.retiredIds.has(runId)) ) { @@ -111,7 +117,7 @@ export abstract class SessionRequests< this.refuse(refused, { inputId, status: 'rejected', reason: 'conflict' }) return refused } - const operation = this.createTurnOperation(runId) + const operation = passed ?? this.createTurnOperation(runId) this.bindTurn(inputId, operation) void this.admitting(() => this.accept(inputId, input, principal, runId).then( diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 5c6e8115de..09f87431dc 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -8,7 +8,7 @@ import { defineHarness, definePlugin, } from '../src' -import { after, gate, mockAdapter, text } from './helpers' +import { after, gate, messageTexts, mockAdapter, text } from './helpers' import type { AnyTextAdapter } from '@tanstack/ai' const THREAD = 't1' @@ -230,4 +230,105 @@ describe('an input that reaches a host that does not own the thread', () => { }) expect(await running).toBe('deployed') }) + + it('runs a forwarded POST run on the host that took it, under its run id, after the owner stopped', async () => { + const persistence = sharedStores() + const { adapter, calls } = mockAdapter([() => text('Picked up.')]) + // A host that stopped holds a claim that is about to expire. + await persistence.stores.workClaims.claim({ + threadId: THREAD, + harness: NAME, + ownerId: 'stopped-host', + until: Date.now() + 50, + }) + const host = await createHarnessHost({ persistence }).open( + harnessOf(adapter), + { threadId: THREAD }, + ) + const turn = host.prompt('Still there?', { runId: 'run-9' }) + expect((await turn.receipt).status).toBe('queued') + // The claim expires. This host reads the inbox and runs the input. + await vi.waitFor(() => expect(calls).toHaveLength(1), { timeout: 3_000 }) + expect(turn.id).toBe('run-9') + expect((await turn).text).toBe('Picked up.') + }) + + it('keeps an inbox entry pending when the owner fails to store the input', async () => { + const persistence = sharedStores() + const release = gate() + const { adapter } = mockAdapter([after(release.opened, 'One.')]) + const harness = harnessOf(adapter) + const log = persistence.stores.log + let failing: string | undefined + const owner = await createHarnessHost({ + persistence: { + stores: { + ...persistence.stores, + log: Object.assign(Object.create(log), { + append: (batch: Parameters[0]) => + batch.records.some( + (record) => + record.type === 'harness.input' && + record.inputId === failing, + ) + ? Promise.reject(new Error('disk full')) + : log.append(batch), + }), + }, + }, + }).open(harness, { threadId: THREAD }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const turn = owner.prompt('Go.') + await turn.receipt + const receipt = await other.prompt('And then?').receipt + expect(receipt.status).toBe('queued') + failing = receipt.inputId + // The owner reads the inbox within a second, and the write fails. + await new Promise((resolve) => setTimeout(resolve, 1_500)) + // The input waits for the next host. + expect( + (await persistence.stores.inbox.listPending(THREAD)).map( + (entry) => entry.inputId, + ), + ).toEqual([receipt.inputId]) + release.open() + await Promise.resolve(turn).catch(() => {}) + }) + + it('recovers the input of the stopped owner when a host takes the thread from the inbox', async () => { + const persistence = sharedStores() + const { adapter, calls } = mockAdapter([ + () => text('One.'), + () => text('Two.'), + ]) + const harness = harnessOf(adapter) + // The owner stores its input, then stops before it writes its run lease. + const runs = persistence.stores.runs + const owner = await createHarnessHost({ + lease: { ttlMs: 200, renewMs: 50 }, + persistence: { + stores: { + ...persistence.stores, + runs: Object.assign(Object.create(runs), { + createOrResume: () => new Promise(() => {}), + }), + }, + }, + }).open(harness, { threadId: THREAD }) + await owner.prompt('First.').receipt + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const receipt = await other.followUp('Second.') + expect(receipt.status).toBe('queued') + + // The owner stops renewing its claim, as after a crash. + ;(owner as unknown as { stopClaimTimer: () => void }).stopClaimTimer() + await vi.waitFor(() => expect(calls).toHaveLength(2), { timeout: 4_000 }) + // The owner's input runs first, then the forwarded one. + expect(messageTexts(calls[1])).toContain('Second.') + }) }) From 8c38b16d8a9cfc823d079e17c651f0d42eb82c0d Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 15 Oct 2026 13:10:50 +0200 Subject: [PATCH 12/15] fix(ai-harness): a forwarded resume runs as its run id, and two lint errors - A POST run with a resume that reaches a host that does not own the thread got HTTP 409: the queued receipt had no operationId, and the owner ran the resume under a new id. The inbox entry now keeps the run id, the owner runs the resume as it, and the receipt names it. - watchInbox passed an async function to setInterval, and applyForwarded had a default case where the switch can be exhaustive. Both failed test:oxlint in CI. --- packages/ai-harness/src/session/base.ts | 20 +++---- packages/ai-harness/src/session/requests.ts | 11 +++- packages/ai-harness/tests/forwarding.test.ts | 56 ++++++++++++++++++-- 3 files changed, 71 insertions(+), 16 deletions(-) diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index 30848bd3e0..c21eb49574 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -722,12 +722,13 @@ export abstract class SessionBase { clearInterval(timer) this.inboxWatch = undefined } - const timer = setInterval(async () => { + const check = async () => { if (this.closing) return stop() const pending = await inbox.listPending(this.threadId).catch(() => null) if (pending?.length === 0) return stop() void this.drainInbox() - }, INBOX_POLL_MS) + } + const timer = setInterval(() => void check(), INBOX_POLL_MS) if (typeof timer === 'object' && 'unref' in timer) timer.unref() this.inboxWatch = timer } @@ -808,17 +809,17 @@ export abstract class SessionBase { settle(operation) return operation.receipt } + const { runId } = entry.input as { runId?: unknown } + const run = typeof runId === 'string' ? { runId } : {} switch (input.op) { - case 'prompt': { - const { runId } = entry.input as { runId?: unknown } + case 'prompt': return stored( session.prompt(input.message, { ...sent, ...(input.busy ? { busy: input.busy } : {}), - ...(typeof runId === 'string' ? { runId } : {}), + ...run, }), ) - } case 'continue': return stored(session.continue(sent)) case 'steer': @@ -826,7 +827,7 @@ export abstract class SessionBase { case 'followUp': return session.followUp(input.message, sent) case 'resolve': - return session.resolve(input.resume, sent) + return session.resolve(input.resume, { ...sent, ...run }) case 'reset': return session.reset(input.note, sent) case 'configure': @@ -866,8 +867,9 @@ export abstract class SessionBase { return session.revert(input.messageId) case 'unrevert': return session.unrevert() - default: - // `agent` and `background` start on the owner, never on another host. + case 'agent': + case 'background': + // These start on the owner, never on another host. return undefined } } diff --git a/packages/ai-harness/src/session/requests.ts b/packages/ai-harness/src/session/requests.ts index 1e8273e8b0..d657c9bd87 100644 --- a/packages/ai-harness/src/session/requests.ts +++ b/packages/ai-harness/src/session/requests.ts @@ -427,13 +427,20 @@ export abstract class SessionRequests< return this.admitting(async () => { const inputId = options?.inputId ?? createInputId() const principal = options?.principal ?? this.principal + const runId = options?.runId const admission = await this.accept( inputId, { op: 'resolve', resume }, principal, + runId, ) - if (admission !== 'new') return this.duplicateReceipt(inputId, admission) - const runId = options?.runId + if (admission !== 'new') { + const receipt = this.duplicateReceipt(inputId, admission) + // The owner runs a forwarded resume as `runId`. + return admission === 'forwarded' && runId !== undefined + ? { ...receipt, operationId: runId } + : receipt + } // A run id must not take the id of another operation or stored run. const isTaken = runId !== undefined && diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 09f87431dc..52903e5085 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it, vi } from 'vitest' import { z } from 'zod' -import { EventType } from '@tanstack/ai' +import { EventType, toolDefinition } from '@tanstack/ai' import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' import { createHarnessHost, @@ -8,7 +8,14 @@ import { defineHarness, definePlugin, } from '../src' -import { after, gate, messageTexts, mockAdapter, text } from './helpers' +import { + after, + gate, + messageTexts, + mockAdapter, + text, + toolCall, +} from './helpers' import type { AnyTextAdapter } from '@tanstack/ai' const THREAD = 't1' @@ -268,8 +275,7 @@ describe('an input that reaches a host that does not own the thread', () => { append: (batch: Parameters[0]) => batch.records.some( (record) => - record.type === 'harness.input' && - record.inputId === failing, + record.type === 'harness.input' && record.inputId === failing, ) ? Promise.reject(new Error('disk full')) : log.append(batch), @@ -290,7 +296,7 @@ describe('an input that reaches a host that does not own the thread', () => { await new Promise((resolve) => setTimeout(resolve, 1_500)) // The input waits for the next host. expect( - (await persistence.stores.inbox.listPending(THREAD)).map( + (await persistence.stores.inbox!.listPending(THREAD)).map( (entry) => entry.inputId, ), ).toEqual([receipt.inputId]) @@ -331,4 +337,44 @@ describe('an input that reaches a host that does not own the thread', () => { // The owner's input runs first, then the forwarded one. expect(messageTexts(calls[1])).toContain('Second.') }) + + it('runs a forwarded resume as its run id, and its receipt names that run', async () => { + const persistence = sharedStores() + const openScreen = toolDefinition({ + name: 'openScreen', + description: 'Open a screen. Runs on the client.', + inputSchema: z.object({ route: z.string() }), + outputSchema: z.object({ opened: z.boolean() }), + }) + const { adapter, calls } = mockAdapter([ + () => toolCall('openScreen', { route: '/settings' }), + () => text('Opened it.'), + ]) + const harness = defineHarness({ name: NAME, adapter, tools: [openScreen] }) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const turn = await owner.prompt('Open settings.') + const interruptId = turn.interrupts?.[0]?.id + if (!interruptId) throw new Error('The turn did not stop for the tool.') + // Another host holds the thread when the resume comes (a turn that + // still ends, for example), then stops. + await persistence.stores.workClaims.claim({ + threadId: THREAD, + harness: NAME, + ownerId: 'busy-host', + until: Date.now() + 100, + }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + // A `POST run` with a resume: the run id is the input id too. + const receipt = await other.resolve( + [{ interruptId, status: 'resolved', payload: { opened: true } }], + { inputId: 'run-r', runId: 'run-r' }, + ) + expect(receipt).toMatchObject({ status: 'queued', operationId: 'run-r' }) + await vi.waitFor(() => expect(calls).toHaveLength(2), { timeout: 3_000 }) + expect((await other.settled('run-r')).outcome).toBe('completed') + }) }) From bf983b0f2d2618df121ba0736217b5ee0bd396ec Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 15 Oct 2026 13:20:30 +0200 Subject: [PATCH 13/15] fix(ai-harness, ai-persistence): keep a forwarded secret answer out of the inbox A host that does not have a question cannot tell if it is a secret, so it stored the answer in stores.inbox as typed, and the entry stayed there. InboxStore.delete removes an entry. The owner deletes a forwarded answer after it applies it, and logs a secret answer as [secret]. --- .changeset/persistence-store-shapes.md | 2 +- docs/config.json | 4 +- docs/harness/durable-sessions.md | 2 + docs/persistence/store-reference.md | 3 + packages/ai-harness/src/session/base.ts | 9 ++- packages/ai-harness/src/session/state.ts | 6 +- packages/ai-harness/tests/forwarding.test.ts | 56 +++++++++++++++++++ packages/ai-persistence/src/memory.ts | 10 ++++ .../ai-persistence/src/testkit/conformance.ts | 13 ++++- packages/ai-persistence/src/types.ts | 5 ++ packages/ai-persistence/tests/inbox.test.ts | 1 + 11 files changed, 104 insertions(+), 7 deletions(-) diff --git a/.changeset/persistence-store-shapes.md b/.changeset/persistence-store-shapes.md index 60b49e7950..c448f320e4 100644 --- a/.changeset/persistence-store-shapes.md +++ b/.changeset/persistence-store-shapes.md @@ -2,4 +2,4 @@ '@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. `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. +`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. diff --git a/docs/config.json b/docs/config.json index 4c8247c5f5..1f2da30b86 100644 --- a/docs/config.json +++ b/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", @@ -951,7 +951,7 @@ "label": "Durable sessions", "to": "harness/durable-sessions", "addedAt": "2026-10-13", - "updatedAt": "2026-10-14" + "updatedAt": "2026-10-15" }, { "label": "List, rename, and fork sessions", diff --git a/docs/harness/durable-sessions.md b/docs/harness/durable-sessions.md index 21e21d96e0..4391846153 100644 --- a/docs/harness/durable-sessions.md +++ b/docs/harness/durable-sessions.md @@ -149,6 +149,8 @@ The session gives the claim back when the thread is idle: no queued or running t 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 })`: diff --git a/docs/persistence/store-reference.md b/docs/persistence/store-reference.md index 9ed9ac977f..7d68aa5f16 100644 --- a/docs/persistence/store-reference.md +++ b/docs/persistence/store-reference.md @@ -630,6 +630,7 @@ interface InboxStore { markApplied(entry: { threadId: string; inputId: string; operationId: string }): Promise markRejected(entry: { threadId: string; inputId: string; reason: string }): Promise get(entry: { threadId: string; inputId: string }): Promise + delete(entry: { threadId: string; inputId: string }): Promise } ``` @@ -642,6 +643,8 @@ Rules for an implementation: - `markApplied` and `markRejected` do nothing 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. diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index c21eb49574..5b19b6442b 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -770,9 +770,16 @@ export abstract class SessionBase { await this.applyForwarded(entry) // The input was not stored. It waits for the next host. if (this.closing || this.logFailure) return + const { inputId } = entry + // The host that stored an answer did not know if it is a secret. + // This host does, and logs it hidden. The entry goes. + if ((entry.input as HarnessInput).op === 'answer') { + await inbox.delete({ threadId: this.threadId, inputId }) + continue + } await inbox.markApplied({ threadId: this.threadId, - inputId: entry.inputId, + inputId, operationId: 'forwarded', }) } diff --git a/packages/ai-harness/src/session/state.ts b/packages/ai-harness/src/session/state.ts index 624052e38d..4a9b0fe756 100644 --- a/packages/ai-harness/src/session/state.ts +++ b/packages/ai-harness/src/session/state.ts @@ -617,8 +617,10 @@ export abstract class SessionState< value: unknown }): Promise { const inputId = createInputId() - // A secret (a key or a password) stays out of the inbox. A restart does - // not replay an answer, so the inbox never needs it. + // A secret (a key or a password) stays out of the inbox and the log. A + // restart does not replay an answer, so they never need it. A host that + // does not have the question cannot tell: the owner deletes the inbox + // entry of the answer after it applies it. const isSecret = this.questions.get(questionId)?.secret === true // A host that does not have the question passes the value on to its owner. const admission = await this.accept(inputId, { diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 52903e5085..8e2c02e8e4 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -238,6 +238,62 @@ describe('an input that reaches a host that does not own the thread', () => { expect(await running).toBe('deployed') }) + it('keeps a secret answer from another host out of the inbox and the log', async () => { + const persistence = sharedStores() + const login = definePlugin({ + name: 'test/login', + setup: () => ({ + commands: { + login: defineCommand({ + description: 'Sign in with a password', + run: async (_input, ctx) => { + const password = await ctx.session.ask({ + message: 'Password?', + schema: z.string(), + secret: true, + }) + return password === 'hunter2' ? 'signed in' : 'wrong' + }, + }), + }, + }), + }) + const harness = defineHarness({ + name: NAME, + adapter: mockAdapter([]).adapter, + plugins: () => [login], + }) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const running = owner.command('login') + await vi.waitFor(() => + expect(owner.snapshot().pendingQuestions).toHaveLength(1), + ) + const [question] = owner.snapshot().pendingQuestions + const receipt = await other.answer({ + questionId: question!.questionId, + value: 'hunter2', + }) + expect(receipt.status).toBe('queued') + expect(await running).toBe('signed in') + // The owner deletes the entry after it applies the answer. + await vi.waitFor(async () => + expect( + await persistence.stores.inbox!.get({ + threadId: THREAD, + inputId: receipt.inputId, + }), + ).toBeNull(), + ) + const log = await persistence.stores.log.read(THREAD) + expect(JSON.stringify(log)).not.toContain('hunter2') + }) + it('runs a forwarded POST run on the host that took it, under its run id, after the owner stopped', async () => { const persistence = sharedStores() const { adapter, calls } = mockAdapter([() => text('Picked up.')]) diff --git a/packages/ai-persistence/src/memory.ts b/packages/ai-persistence/src/memory.ts index 12347aab9e..d4b4b3d1bb 100644 --- a/packages/ai-persistence/src/memory.ts +++ b/packages/ai-persistence/src/memory.ts @@ -664,6 +664,16 @@ class MemoryInboxStore implements InboxStore { const existing = this.of(threadId, inputId) return Promise.resolve(existing ? structuredClone(existing) : null) } + delete({ + threadId, + inputId, + }: { + threadId: string + inputId: string + }): Promise { + if (this.of(threadId, inputId)) this.entries.delete(inputId) + return Promise.resolve() + } private of(threadId: string, inputId: string) { const existing = this.entries.get(inputId) return existing?.threadId === threadId ? existing : undefined diff --git a/packages/ai-persistence/src/testkit/conformance.ts b/packages/ai-persistence/src/testkit/conformance.ts index 24e451a0a5..827f34df4f 100644 --- a/packages/ai-persistence/src/testkit/conformance.ts +++ b/packages/ai-persistence/src/testkit/conformance.ts @@ -1639,6 +1639,7 @@ export function runPersistenceConformance( const in2 = id('in-2') const other = id('in-other') const missing = id('missing') + const otherThread = id('other-thread') const first = await store.append({ inputId: in1, @@ -1665,7 +1666,7 @@ export function runPersistenceConformance( }) await store.append({ inputId: other, - threadId: id('other-thread'), + threadId: otherThread, input: { op: 'prompt', message: 'x' }, createdAt: 0, }) @@ -1715,6 +1716,16 @@ export function runPersistenceConformance( ).toBeNull() // An id is found only with its own thread. expect(await store.get({ threadId: thread, inputId: other })).toBeNull() + + // A delete removes the entry. With another thread, or an unknown + // id, it does nothing. + await store.delete({ threadId: thread, inputId: other }) + await store.delete({ threadId: thread, inputId: missing }) + expect( + await store.get({ threadId: otherThread, inputId: other }), + ).toMatchObject({ status: 'pending' }) + await store.delete({ threadId: thread, inputId: in1 }) + expect(await store.get({ threadId: thread, inputId: in1 })).toBeNull() }) }) diff --git a/packages/ai-persistence/src/types.ts b/packages/ai-persistence/src/types.ts index 0a2bdf1192..59b0585abb 100644 --- a/packages/ai-persistence/src/types.ts +++ b/packages/ai-persistence/src/types.ts @@ -700,6 +700,11 @@ export interface InboxStore { threadId: string inputId: string }) => Promise + /** + * Remove the entry. A no-op for an unknown id. The host removes an answer + * after it applies it, so a secret answer does not stay in the store. + */ + delete: (entry: { threadId: string; inputId: string }) => Promise } /** Type an {@link InboxStore} implementation inline. */ diff --git a/packages/ai-persistence/tests/inbox.test.ts b/packages/ai-persistence/tests/inbox.test.ts index 2bfe2773a5..e2a362a29e 100644 --- a/packages/ai-persistence/tests/inbox.test.ts +++ b/packages/ai-persistence/tests/inbox.test.ts @@ -32,6 +32,7 @@ describe('inbox store', () => { markApplied: () => Promise.resolve(), markRejected: () => Promise.resolve(), get: () => Promise.resolve(null), + delete: () => Promise.resolve(), }) expect(defineAIPersistence({ stores: { inbox } }).stores.inbox).toBe(inbox) From 38851a6694a55ed78578c1bd8ee17a7660a95063 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 15 Oct 2026 13:29:54 +0200 Subject: [PATCH 14/15] fix(ai-harness): a forwarded input runs once, keeps its wake mark, and ids stay bounded - The owner passes the inbox input id to cancel, command, answer, and setConfig (an @internal inputId option). An entry that the owner applies again, after a crash or a failed markApplied, is a duplicate. Before, a forwarded command ran again on each read of the inbox. - A forwarded follow-up that wakes the session keeps its wake mark. - The forwarded id set keeps the newest 500 ids, like settled operations. - Docs: a forwarded prompt runs without overrides and ephemeral messages. --- docs/harness/durable-sessions.md | 2 + packages/ai-harness/src/session/base.ts | 35 ++++-- packages/ai-harness/src/session/inputs.ts | 13 ++- packages/ai-harness/src/session/requests.ts | 10 +- packages/ai-harness/src/session/state.ts | 13 ++- packages/ai-harness/src/session/turn.ts | 16 ++- packages/ai-harness/tests/forwarding.test.ts | 110 +++++++++++++++++++ 7 files changed, 173 insertions(+), 26 deletions(-) diff --git a/docs/harness/durable-sessions.md b/docs/harness/durable-sessions.md index 4391846153..d18ea03936 100644 --- a/docs/harness/durable-sessions.md +++ b/docs/harness/durable-sessions.md @@ -335,6 +335,8 @@ The host that runs the thread holds its work claim. An input that reaches anothe 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. diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index 5b19b6442b..dc1b8694d6 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -69,6 +69,7 @@ import type { Receipt, SessionEvent, ThreadSettings, + UserInput, } from '../types' import type { AgentRunOptions, @@ -94,7 +95,7 @@ import type { HarnessSession } from './harness-session' // the inbox. The duplicate check of an input with no operation of its own (a // command, a cancel) keeps its id forever: one string each. Make the window // an option if a host needs more. -const KEPT_SETTLED = 500 +export const KEPT_SETTLED = 500 /** * The fields of a {@link HarnessSession}, and the helpers every layer uses. The @@ -832,7 +833,10 @@ export abstract class SessionBase { case 'steer': return session.steer(input.message, sent) case 'followUp': - return session.followUp(input.message, sent) + return this.queueFollowUp(input.message, { + ...sent, + ...(input.wake ? { wake: true } : {}), + }) case 'resolve': return session.resolve(input.resume, { ...sent, ...run }) case 'reset': @@ -847,7 +851,7 @@ export abstract class SessionBase { mode: input.mode, }) case 'cancel': - return session.cancel(input.operationId) + return session.cancel(input.operationId, { inputId }) case 'cancelInput': return session.cancelInput(input.inputId) case 'setDelivery': @@ -857,19 +861,23 @@ export abstract class SessionBase { }) case 'command': return settle( - session.command( - input.name, - input.input, - principal ? { principal } : undefined, - ), + session.command(input.name, input.input, { + inputId, + ...(principal ? { principal } : {}), + }), ) case 'answer': return session.answer({ questionId: input.questionId, value: input.value, + inputId, }) case 'config': - return session.setConfig({ key: input.key, value: input.value }) + return session.setConfig({ + key: input.key, + value: input.value, + inputId, + }) case 'revert': return session.revert(input.messageId) case 'unrevert': @@ -1080,6 +1088,15 @@ export abstract class SessionBase { ): PluginSessionApi protected abstract resumeSignIns(): Promise protected abstract recoverFromLog(writer: LogWriter): Promise + protected abstract queueFollowUp( + message: UserInput, + options?: { + inputId?: string + principal?: Principal + context?: unknown + wake?: boolean + }, + ): Promise protected abstract runAgent( target: string | AnyAgent, input: unknown, diff --git a/packages/ai-harness/src/session/inputs.ts b/packages/ai-harness/src/session/inputs.ts index fb568862f3..004687bc38 100644 --- a/packages/ai-harness/src/session/inputs.ts +++ b/packages/ai-harness/src/session/inputs.ts @@ -12,7 +12,7 @@ import { lastAssistantText, checkHostRecords, } from './helpers' -import { SessionBase } from './base' +import { KEPT_SETTLED, SessionBase } from './base' import type { OperationImpl } from '../operation' import type { LogRecord } from '@tanstack/ai-persistence' import type { HarnessRecord } from '../log' @@ -409,8 +409,15 @@ export abstract class SessionInputs< this.admitted.delete(inputId) const inbox = this.persistence.stores.inbox if (!inbox) return 'owned_elsewhere' - // `settled` waits for the owner to end it. - if (CHAT_OPS.has(input.op)) this.forwarded.add(inputId) + // `settled` waits for the owner to end it. The oldest ids go, as in the + // window of settled operations. + if (CHAT_OPS.has(input.op)) { + this.forwarded.add(inputId) + for (const old of this.forwarded) { + if (this.forwarded.size <= KEPT_SETTLED) break + this.forwarded.delete(old) + } + } await inbox.append({ inputId, threadId: this.threadId, diff --git a/packages/ai-harness/src/session/requests.ts b/packages/ai-harness/src/session/requests.ts index d657c9bd87..64a7ca7454 100644 --- a/packages/ai-harness/src/session/requests.ts +++ b/packages/ai-harness/src/session/requests.ts @@ -495,10 +495,14 @@ export abstract class SessionRequests< }) } - async cancel(operationId?: string): Promise { - const inputId = createInputId() + async cancel( + operationId?: string, + /** @internal The id of a forwarded input, so a replay is a duplicate. */ + options?: { inputId?: string }, + ): Promise { + const inputId = options?.inputId ?? createInputId() const admission = await this.accept(inputId, { op: 'cancel', operationId }) - if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) + if (admission !== 'new') return this.duplicateReceipt(inputId, admission) const target = operationId ? this.operations.get(operationId) : this.activeTurn diff --git a/packages/ai-harness/src/session/state.ts b/packages/ai-harness/src/session/state.ts index 4a9b0fe756..54a0a1e31f 100644 --- a/packages/ai-harness/src/session/state.ts +++ b/packages/ai-harness/src/session/state.ts @@ -29,7 +29,6 @@ import { storedPrincipal, SETTINGS, changeSettings, - isElsewhere, } from './helpers' import { SessionInputs } from './inputs' import type { OperationImpl } from '../operation' @@ -285,13 +284,15 @@ export abstract class SessionState< async setConfig({ key, value, + inputId = createInputId(), }: { key: string value: unknown + /** @internal The id of a forwarded input, so a replay is a duplicate. */ + inputId?: string }): Promise { - const inputId = createInputId() const admission = await this.accept(inputId, { op: 'config', key, value }) - if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) + if (admission !== 'new') return this.duplicateReceipt(inputId, admission) const entry = this.sessionPlugins?.config.get(key) if (!entry) return this.reject(inputId, 'unknown_config') let checked: unknown @@ -612,11 +613,13 @@ export abstract class SessionState< async answer({ questionId, value, + inputId = createInputId(), }: { questionId: string value: unknown + /** @internal The id of a forwarded input, so a replay is a duplicate. */ + inputId?: string }): Promise { - const inputId = createInputId() // A secret (a key or a password) stays out of the inbox and the log. A // restart does not replay an answer, so they never need it. A host that // does not have the question cannot tell: the owner deletes the inbox @@ -628,7 +631,7 @@ export abstract class SessionState< questionId, value: isSecret ? '[secret]' : value, }) - if (isElsewhere(admission)) return this.duplicateReceipt(inputId, admission) + if (admission !== 'new') return this.duplicateReceipt(inputId, admission) const question = this.questions.get(questionId) if (!question) return this.reject(inputId, 'unknown_question') let checked: unknown = value diff --git a/packages/ai-harness/src/session/turn.ts b/packages/ai-harness/src/session/turn.ts index d6002c0dd2..56b1c8600c 100644 --- a/packages/ai-harness/src/session/turn.ts +++ b/packages/ai-harness/src/session/turn.ts @@ -38,7 +38,6 @@ import { lastAssistantText, isRoutedBag, keepTurnTools, - isElsewhere, } from './helpers' import { SessionAgents } from './agents' import type { @@ -75,7 +74,11 @@ export abstract class SessionTurn< command( name: string, input?: unknown, - options?: { principal?: Principal }, + options?: { + principal?: Principal + /** @internal The id of a forwarded input, so a replay is a duplicate. */ + inputId?: string + }, ): Operation { const operation = new OperationImpl( 'command', @@ -85,7 +88,8 @@ export abstract class SessionTurn< this.track(operation) const principal = options?.principal ?? this.principal // A store write that throws (the inbox, the log) still ends the command. - this.executeCommand(operation, name, input, principal).catch( + const inputId = options?.inputId ?? createInputId() + this.executeCommand(operation, name, input, principal, inputId).catch( (error: unknown) => this.failOperation(operation, error), ) return operation @@ -96,8 +100,8 @@ export abstract class SessionTurn< name: string, input: unknown, principal: Principal | undefined, + inputId: string, ): Promise { - const inputId = createInputId() // The log keeps the input for replay, so a command must not take a secret // as its input. Ask a `secret` question instead. const admission = await this.accept( @@ -105,8 +109,8 @@ export abstract class SessionTurn< { op: 'command', name, input }, principal, ) - // The owner runs it. Its result is in the owner's events. - if (isElsewhere(admission)) { + // The owner runs it, or it ran before. Its result is in the events. + if (admission !== 'new') { const receipt = this.duplicateReceipt(inputId, admission) if (receipt.status === 'rejected') { operation.fail('failed', new InputRejectedError(receipt)) diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 8e2c02e8e4..cacdfd160a 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -15,6 +15,7 @@ import { mockAdapter, text, toolCall, + wakeMessages, } from './helpers' import type { AnyTextAdapter } from '@tanstack/ai' @@ -294,6 +295,115 @@ describe('an input that reaches a host that does not own the thread', () => { expect(JSON.stringify(log)).not.toContain('hunter2') }) + it('runs a forwarded command once when the owner applies it again', async () => { + const persistence = sharedStores() + const release = gate() + const { adapter } = mockAdapter([after(release.opened, 'One.')]) + let runs = 0 + const counter = definePlugin({ + name: 'test/counter', + setup: () => ({ + commands: { + count: defineCommand({ + description: 'Count the runs', + run: () => (runs += 1), + }), + }, + }), + }) + const harness = defineHarness({ + name: NAME, + adapter, + plugins: () => [counter], + }) + // The owner's first `markApplied` fails, so the next read applies the + // entry again. + const inbox = persistence.stores.inbox! + let failed = false + const owner = await createHarnessHost({ + persistence: { + stores: { + ...persistence.stores, + inbox: Object.assign(Object.create(inbox), { + markApplied: (entry: Parameters[0]) => { + if (failed) return inbox.markApplied(entry) + failed = true + return Promise.reject(new Error('store down')) + }, + }), + }, + }, + }).open(harness, { threadId: THREAD }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const turn = owner.prompt('Go.') + await turn.receipt + await other.command('count') + await vi.waitFor( + async () => expect(await inbox.listPending(THREAD)).toEqual([]), + { timeout: 4_000 }, + ) + expect(runs).toBe(1) + release.open() + await turn + }) + + it('keeps the wake mark of a forwarded follow-up', async () => { + const persistence = sharedStores() + const release = gate() + const { adapter, calls } = mockAdapter([ + after(release.opened, 'One.'), + () => text('Two.'), + ]) + const harness = harnessOf(adapter) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + + const turn = owner.prompt('Go.') + await turn.receipt + // A background job ended on the other host and wakes the session. + const wake = other as unknown as { + queueFollowUp: ( + message: string, + options: { wake: boolean }, + ) => Promise<{ status: string }> + } + expect((await wake.queueFollowUp('Job done.', { wake: true })).status).toBe( + 'queued', + ) + release.open() + await vi.waitFor(() => expect(calls).toHaveLength(2), { timeout: 3_000 }) + expect(wakeMessages(calls[1]!.messages)).toEqual(['Job done.']) + }) + + it('keeps a bounded number of forwarded ids', async () => { + const persistence = sharedStores() + await persistence.stores.workClaims.claim({ + threadId: THREAD, + harness: NAME, + ownerId: 'busy-host', + until: Date.now() + 60_000, + }) + const other = await createHarnessHost({ persistence }).open( + harnessOf(mockAdapter([]).adapter), + { threadId: THREAD }, + ) + let last = '' + for (let index = 0; index < 510; index++) { + last = (await other.followUp(`Message ${index}.`)).inputId + } + const { forwarded } = other as unknown as { forwarded: Set } + expect(forwarded.size).toBe(500) + expect(forwarded.has(last)).toBe(true) + await other.close() + }) + it('runs a forwarded POST run on the host that took it, under its run id, after the owner stopped', async () => { const persistence = sharedStores() const { adapter, calls } = mockAdapter([() => text('Picked up.')]) From a092215c2c8a232699bbecaac9a567e48b5f3433 Mon Sep 17 00:00:00 2001 From: Alem Tuzlak Date: Thu, 15 Oct 2026 13:35:51 +0200 Subject: [PATCH 15/15] fix(ai-harness): refuse a revert on a host that does not own the thread revert and unrevert write a harness.revert record and never went through accept, so a host that did not own the thread wrote to the log during the owner's turn, and the owner's turn failed with a log conflict. They now refuse with busy while another host holds the claim. The revert and unrevert cases of applyForwarded never ran, so they go with agent and background. --- docs/config.json | 2 +- docs/harness/sessions.md | 2 +- packages/ai-harness/src/session/base.ts | 4 +-- packages/ai-harness/src/session/state.ts | 11 ++++--- packages/ai-harness/tests/forwarding.test.ts | 31 ++++++++++++++++++++ 5 files changed, 41 insertions(+), 9 deletions(-) diff --git a/docs/config.json b/docs/config.json index 1f2da30b86..21d55cdb1d 100644 --- a/docs/config.json +++ b/docs/config.json @@ -957,7 +957,7 @@ "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", diff --git a/docs/harness/sessions.md b/docs/harness/sessions.md index d4341500bc..b7ef1d620a 100644 --- a/docs/harness/sessions.md +++ b/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 diff --git a/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index dc1b8694d6..105efb5793 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -879,12 +879,10 @@ export abstract class SessionBase { inputId, }) case 'revert': - return session.revert(input.messageId) case 'unrevert': - return session.unrevert() case 'agent': case 'background': - // These start on the owner, never on another host. + // These run only on the owner. Another host never stores them. return undefined } } diff --git a/packages/ai-harness/src/session/state.ts b/packages/ai-harness/src/session/state.ts index 54a0a1e31f..ac34c59ac2 100644 --- a/packages/ai-harness/src/session/state.ts +++ b/packages/ai-harness/src/session/state.ts @@ -476,12 +476,14 @@ export abstract class SessionState< * after it. With the snapshots plugin, the files that their tool calls * changed go back too. `unrevert()` undoes it. The next turn drops the * hidden messages for good. The revert state is saved, so it survives a - * restart. Refused while the session is not idle (`'busy'`), and for a - * message that is not in the transcript (`'unknown_message'`). + * restart. Refused while the session is not idle, or another host runs + * the thread (`'busy'`), and for a message that is not in the transcript + * (`'unknown_message'`). */ async revert(messageId: string): Promise { const inputId = createInputId() - if (this.snapshot().status !== 'idle') { + // Another host runs the thread: its log write would stop that host. + if (this.snapshot().status !== 'idle' || (await this.isOwnedElsewhere())) { return { inputId, status: 'rejected', reason: 'busy' } } return this.holding(async () => { @@ -506,7 +508,8 @@ export abstract class SessionState< /** End the revert that stands: its files and messages come back. */ async unrevert(): Promise { const inputId = createInputId() - if (this.snapshot().status !== 'idle') { + // Another host runs the thread: its log write would stop that host. + if (this.snapshot().status !== 'idle' || (await this.isOwnedElsewhere())) { return { inputId, status: 'rejected', reason: 'busy' } } return this.holding(async () => { diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index cacdfd160a..1bbb396cc6 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -404,6 +404,37 @@ describe('an input that reaches a host that does not own the thread', () => { await other.close() }) + it('refuses a revert on a host that does not own the thread, and the owner turn goes on', async () => { + const persistence = sharedStores() + const release = gate() + const { adapter } = mockAdapter([ + () => text('One.'), + after(release.opened, 'Two.'), + ]) + const harness = harnessOf(adapter) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + await owner.prompt('First.') + const turn = owner.prompt('Second.') + await turn.receipt + // Another instance opens the thread while the owner runs the turn. + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const [first] = await other.transcript() + expect(await other.revert(first?.id ?? '')).toMatchObject({ + status: 'rejected', + reason: 'busy', + }) + expect(await other.unrevert()).toMatchObject({ + status: 'rejected', + reason: 'busy', + }) + release.open() + expect((await turn).text).toBe('Two.') + }) + it('runs a forwarded POST run on the host that took it, under its run id, after the owner stopped', async () => { const persistence = sharedStores() const { adapter, calls } = mockAdapter([() => text('Picked up.')])