fix(ai-harness): a second instance must not stop or interleave the owner's turn

.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.

.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.

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",

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.
 

packages/ai-harness/src/session/base.ts

@@ -201,6 +201,12 @@ export abstract class SessionBase<THarness extends AnyHarness = AnyHarness> {
   protected inboxWatch: ReturnType<typeof setInterval> | 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<void> | undefined
   /** Called each time the thread goes idle. See `onIdle`. */
   protected readonly idleListeners = new Set<() => void>()
@@ -738,9 +744,22 @@ export abstract class SessionBase<THarness extends AnyHarness = AnyHarness> {
     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<THarness extends AnyHarness = AnyHarness> {
     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,

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,

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<typeof workClaims.claim>) => {
+        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({