diff --git a/.changeset/harness-live-lease-owner.md b/.changeset/harness-live-lease-owner.md new file mode 100644 index 0000000000..0ff768b029 --- /dev/null +++ b/.changeset/harness-live-lease-owner.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-harness': patch +--- + +A host that finds a live run lease of another host on a thread treats the thread as owned elsewhere, also when that host's work claim lapsed. Before, a claim whose writes stalled (a store that retries) expired while its turn still ran, and another host ran the queued inputs in the middle of that turn: a user message landed between a tool call and its result. diff --git a/.changeset/harness-session-events-owner.md b/.changeset/harness-session-events-owner.md new file mode 100644 index 0000000000..912619ba5d --- /dev/null +++ b/.changeset/harness-session-events-owner.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-harness': patch +--- + +A host that opens a thread that another host owns keeps its session events (plugin state, plugin events, warnings) off the log. Before, the MCP plugin's connection status on a second instance took the log from the owner, and the owner's turn stopped. diff --git a/docs/config.json b/docs/config.json index 5eded52077..3e85940b86 100644 --- a/docs/config.json +++ b/docs/config.json @@ -966,7 +966,7 @@ "label": "Durable sessions", "to": "harness/durable-sessions", "addedAt": "2026-10-13", - "updatedAt": "2026-10-15" + "updatedAt": "2026-10-16" }, { "label": "List, rename, and fork sessions", diff --git a/docs/harness/durable-sessions.md b/docs/harness/durable-sessions.md index d18ea03936..9050c4fa4a 100644 --- a/docs/harness/durable-sessions.md +++ b/docs/harness/durable-sessions.md @@ -328,16 +328,18 @@ const replica = createHarnessHost({ }) ``` -The host that runs the thread holds its work claim. An input that reaches another host: +The host that runs the thread holds its work claim. A claim can expire while its host still runs a turn, for example when the claim writes are slow. A live run lease of that host still makes it the owner. 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. +A host that does not own the thread writes nothing to its log. This includes plugin state and plugin events, for example the status of an MCP connection on that host. Clients of that host do not see these events until it owns the thread. + 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. +When the owner stops, its claim and its run lease expire. 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 c9ec860e05..c0d84d4eea 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -201,6 +201,12 @@ export abstract class SessionBase { protected inboxWatch: ReturnType | undefined /** The open skipped recovery: another host owned the thread then. */ protected recoveryDeferred = false + /** + * Another host held a live claim on this thread when this host last read + * it. Session events of this host then stay off the log: an append here + * would take the log from the owner. + */ + protected ownerElsewhere = false private draining: Promise | undefined /** Called each time the thread goes idle. See `onIdle`. */ protected readonly idleListeners = new Set<() => void>() @@ -738,9 +744,22 @@ export abstract class SessionBase { 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() - ) + const now = Date.now() + if (held !== null && held.ownerId !== this.hostId && held.until > now) { + this.ownerElsewhere = true + return true + } + // A claim can lapse while its host still runs a turn: claim writes can + // stall (a store that retries) while the run lease renews. A live run + // lease of another host means that host still drives the thread. + const run = await this.persistence.stores.runs?.findActiveRun(this.threadId) + this.ownerElsewhere = + run !== null && + run !== undefined && + run.leaseOwner !== undefined && + run.leaseOwner !== this.hostId && + (run.leaseExpiresAt ?? 0) > now + return this.ownerElsewhere } /** @@ -786,6 +805,8 @@ export abstract class SessionBase { try { const pending = await inbox.listPending(this.threadId) if (pending.length === 0) return + // Its claim lapsed, but the owner still runs a turn: leave its inputs. + if (await this.isOwnedElsewhere()) return const isHeld = await claims.claim({ threadId: this.threadId, harness: this.harness.name, diff --git a/packages/ai-harness/src/session/harness-session.ts b/packages/ai-harness/src/session/harness-session.ts index 05eb724123..28408a80bf 100644 --- a/packages/ai-harness/src/session/harness-session.ts +++ b/packages/ai-harness/src/session/harness-session.ts @@ -26,6 +26,7 @@ import { SessionTurn } from './turn' import type { ModelMessage } from '@tanstack/ai' import type { InboxEntry } from '@tanstack/ai-persistence' import type { InputState, LogWriter } from '../log' +import type { EventFeed } from '../feed' import type { AnyHarness } from '../define' import type { RecoverContext, RecoverDecision } from '../turn' import type { HarnessInput, SettlementOutcome } from '../types' @@ -48,6 +49,9 @@ export class HarnessSession< // First: plugins publish events while they mount. await this.openLog() try { + // Before the plugins mount: their events stay off the log of a thread + // that another host owns. + await this.isOwnedElsewhere() await this.mountSessionPlugins() await this.loadSettings() this.reverted = this.writer @@ -212,7 +216,21 @@ export class HarnessSession< this.writer = await this.log.open(this.threadId, (error) => this.stopOnLogFailure(error), ) - this.feed = tapFeed(this.writer, this.notifyHost, this.eventListeners) + const writer = this.writer + // ponytail: session events (plugin state, warnings) of a host that does + // not own the thread are dropped, not passed to the owner. Clients of + // this host miss them until it owns the thread. + const feed: EventFeed = { + publish: (operationId, event) => { + if (operationId === 'session' && this.ownerElsewhere) return + writer.publish(operationId, event) + }, + head: () => writer.head(), + read: (options) => writer.read(options), + close: () => writer.close(), + wake: () => writer.wake(), + } + this.feed = tapFeed(feed, this.notifyHost, this.eventListeners) const view = { writer: this.writer, store, diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 55bcf128ac..116d146fbd 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -201,6 +201,132 @@ describe('an input that reaches a host that does not own the thread', () => { await other.close() }) + it('keeps the plugin state of another host off the log while the owner runs a tool', async () => { + const persistence = sharedStores() + const release = gate() + const wait = toolDefinition({ + name: 'wait', + description: 'Wait', + inputSchema: z.object({}), + }).server(async () => { + await release.opened + return 'waited' + }) + // Like the MCP plugin: each host reports its own connection on mount. + const status = definePlugin({ + name: 'test/status', + setup: (ctx) => { + void ctx.state({ connected: false }).update(() => ({ connected: true })) + return {} + }, + }) + const { adapter, calls } = mockAdapter([ + () => toolCall('wait', {}), + () => text('One.'), + () => text('Two.'), + ]) + const harness = defineHarness({ + name: NAME, + adapter, + tools: [wait], + plugins: () => [status], + }) + const owner = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const turn = owner.send('Go.') + await vi.waitFor(() => expect(calls).toHaveLength(1)) + + const other = await createHarnessHost({ persistence }).open(harness, { + threadId: THREAD, + }) + const receipt = await other.send('And then?', { mode: 'followUp' }).receipt + expect(receipt.status).toBe('queued') + // The other host's state write had time to land on the log. + await new Promise((resolve) => setTimeout(resolve, 50)) + release.open() + expect((await turn).text).toBe('One.') + const settled = await vi.waitFor(() => owner.settled(receipt.inputId), { + timeout: 3_000, + }) + expect(settled.outcome).toBe('completed') + expect(calls).toHaveLength(3) + await other.close() + }) + + it('leaves a turn alone when its host claim lapsed but its run lease is live', async () => { + const persistence = sharedStores() + // The owner's claim writes hang (a store that retries for a while), and + // its claim expires. Its run lease still renews. + const claimWrites = gate() + let isStalled = false + const { workClaims } = persistence.stores + const stalledClaims = Object.assign(Object.create(workClaims), { + claim: async (...args: Parameters) => { + if (isStalled) await claimWrites.opened + return workClaims.claim(...args) + }, + }) + const toolDone = gate() + const wait = toolDefinition({ + name: 'wait', + description: 'Wait', + inputSchema: z.object({}), + }).server(async () => { + await toolDone.opened + return 'waited' + }) + const { adapter, calls } = mockAdapter([ + () => toolCall('wait', {}), + () => text('One.'), + () => text('Two.'), + () => text('Three.'), + ]) + const harness = defineHarness({ name: NAME, adapter, tools: [wait] }) + const lease = { renewMs: 50, ttlMs: 300 } + const ownerHost = createHarnessHost({ + persistence: { + stores: { ...persistence.stores, workClaims: stalledClaims }, + }, + lease, + }) + const owner = await ownerHost.open(harness, { threadId: THREAD }) + const turn = owner.send('Go.') + await vi.waitFor(() => expect(calls).toHaveLength(1)) + await owner.send('And then?', { mode: 'followUp' }).receipt + isStalled = true + await new Promise((resolve) => setTimeout(resolve, 600)) + + // Another host opens the thread after the claim expired. + const otherHost = createHarnessHost({ persistence, lease }) + const other = await otherHost.open(harness, { threadId: THREAD }) + await otherHost.recover() + // An input on that host waits in the inbox for the owner, and its inbox + // watch (every second) leaves it there. + const queued = await other.send('Also?', { mode: 'followUp' }).receipt + expect(queued.status).toBe('queued') + await new Promise((resolve) => setTimeout(resolve, 1_500)) + expect(calls).toHaveLength(1) + + isStalled = false + claimWrites.open() + toolDone.open() + expect((await turn).text).toBe('One.') + await vi.waitFor(() => expect(calls).toHaveLength(4), { timeout: 5_000 }) + expect(await owner.transcript()).toMatchObject([ + { role: 'user', content: 'Go.' }, + { role: 'assistant' }, + { role: 'tool', content: 'waited' }, + { role: 'assistant', content: 'One.' }, + { role: 'user', content: 'And then?' }, + { role: 'assistant', content: 'Two.' }, + { role: 'user', content: 'Also?' }, + { role: 'assistant', content: 'Three.' }, + ]) + await otherHost.close() + await ownerHost.close() + }) + it('answers a question that waits on another host', async () => { const persistence = sharedStores() const deploy = definePlugin({ diff --git a/testing/e2e/src/routes/api.harness-forwarding.ts b/testing/e2e/src/routes/api.harness-forwarding.ts index 6d5bd48e1d..c508ef3169 100644 --- a/testing/e2e/src/routes/api.harness-forwarding.ts +++ b/testing/e2e/src/routes/api.harness-forwarding.ts @@ -1,7 +1,18 @@ import { createFileRoute } from '@tanstack/react-router' import { createAnthropicChat } from '@tanstack/ai-anthropic' -import { createHarnessHost, defineHarness } from '@tanstack/ai-harness' +import { + createHarnessHost, + defineHarness, + definePlugin, +} from '@tanstack/ai-harness' import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' +import type { RunStore } from '@tanstack/ai' +import type { + InboxStore, + LogStore, + MetadataStore, + WorkClaimStore, +} from '@tanstack/ai-persistence' /** * Two hosts share one set of durable stores, as two instances of a @@ -11,6 +22,12 @@ import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' * - `?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. + * - `?plugin=1` (with `?inbox=1`): the second host opens the thread while + * the turn runs, and a plugin writes its state as it mounts, as the MCP + * plugin does. That state stays off the log of the owner. + * - `?stall=1`: the owner's claim writes stall until its claim expires, + * while its run lease still renews. The second host opens the thread and + * recovers it, and must not run the owner's queued 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 @@ -67,7 +84,10 @@ export const Route = createFileRoute('/api/harness-forwarding')({ server: { handlers: { POST: async ({ request }) => { - const withInbox = new URL(request.url).searchParams.get('inbox') === '1' + const params = new URL(request.url).searchParams + const withInbox = params.get('inbox') === '1' + const withPlugin = params.get('plugin') === '1' + const withStall = params.get('stall') === '1' let release!: () => void const released = new Promise((resolve) => { release = resolve @@ -83,6 +103,22 @@ export const Route = createFileRoute('/api/harness-forwarding')({ } const { runs, metadata, workClaims, inbox } = memoryPersistence().stores + if (withStall) { + return Response.json( + await claimStall({ + stores: { + log: memoryLogStore(), + runs, + metadata, + workClaims, + inbox, + }, + fetch: scriptedFetch, + calls: () => calls, + release, + }), + ) + } const persistence = { stores: { log: memoryLogStore(), @@ -92,21 +128,33 @@ export const Route = createFileRoute('/api/harness-forwarding')({ ...(withInbox ? { inbox } : {}), }, } + const status = definePlugin({ + name: 'e2e/status', + setup: (ctx) => { + void ctx + .state({ connected: false }) + .update(() => ({ connected: true })) + return {} + }, + }) const harness = defineHarness({ name: 'e2e/harness-forwarding', adapter: createAnthropicChat(MODEL, DUMMY_KEY, { fetch: scriptedFetch, }), + plugins: () => (withPlugin ? [status] : []), }) 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.send('[harness-forwarding] go') await turn.receipt + const other = await otherHost.open(harness, { threadId }) + // The plugin state write of the second host had time to land. + if (withPlugin) + await new Promise((resolve) => setTimeout(resolve, 50)) const receipt = await other.send('[harness-forwarding] and then', { mode: 'followUp', }).receipt @@ -140,3 +188,80 @@ export const Route = createFileRoute('/api/harness-forwarding')({ }, }, }) + +/** The `?stall=1` case. Returns how many model calls ran during the stall. */ +async function claimStall({ + stores, + fetch: scriptedFetch, + calls, + release, +}: { + stores: { + log: LogStore + runs: RunStore + metadata: MetadataStore + workClaims: WorkClaimStore + inbox: InboxStore + } + fetch: typeof fetch + calls: () => number + release: () => void +}) { + const persistence = { stores } + const { workClaims } = stores + let isStalled = false + let resume!: () => void + const resumed = new Promise((resolve) => { + resume = resolve + }) + const stalledClaims = Object.assign(Object.create(workClaims), { + claim: async (...args: Parameters) => { + if (isStalled) await resumed + return workClaims.claim(...args) + }, + }) + const harness = defineHarness({ + name: 'e2e/harness-forwarding', + adapter: createAnthropicChat(MODEL, DUMMY_KEY, { fetch: scriptedFetch }), + }) + const lease = { renewMs: 50, ttlMs: 300 } + const ownerHost = createHarnessHost({ + persistence: { stores: { ...stores, workClaims: stalledClaims } }, + lease, + }) + const otherHost = createHarnessHost({ persistence, lease }) + const sleep = (ms: number) => + new Promise((resolve) => setTimeout(resolve, ms)) + try { + const threadId = 'e2e-harness-forwarding-stall' + const owner = await ownerHost.open(harness, { threadId }) + const turn = owner.send('[harness-forwarding] go') + await turn.receipt + const receipt = await owner.send('[harness-forwarding] and then', { + mode: 'followUp', + }).receipt + isStalled = true + await sleep(600) + await otherHost.open(harness, { threadId }) + await otherHost.recover() + await sleep(200) + const callsDuringStall = calls() + isStalled = false + resume() + release() + const first = await turn + const followUp = (await owner.settled(receipt.inputId)).outcome + return { + callsDuringStall, + first: first.text, + followUp, + calls: calls(), + transcript: textsOf(await owner.transcript()), + } + } catch (error) { + return { error: error instanceof Error ? error.message : String(error) } + } finally { + await otherHost.close() + await ownerHost.close() + } +} diff --git a/testing/e2e/tests/harness-forwarding.spec.ts b/testing/e2e/tests/harness-forwarding.spec.ts index a6260857ba..6be8199af8 100644 --- a/testing/e2e/tests/harness-forwarding.spec.ts +++ b/testing/e2e/tests/harness-forwarding.spec.ts @@ -30,6 +30,39 @@ test.describe('harness: an input on a host that does not own the thread', () => ]) }) + test('keeps the plugin state of the second host off the log, and the owner turn ends normally', async ({ + request, + }) => { + const res = await request.post('/api/harness-forwarding?inbox=1&plugin=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) + }) + + test('leaves a turn alone when its host claim lapsed but its run lease is live', async ({ + request, + }) => { + const res = await request.post('/api/harness-forwarding?stall=1') + expect(res.ok()).toBe(true) + const result: Result & { callsDuringStall: number } = await res.json() + expect(result.error ?? null).toBeNull() + + expect(result.callsDuringStall).toBe(1) + expect(result.first).toBe('One.') + expect(result.followUp).toBe('completed') + 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, }) => {