From b65d8de260cc7c2f304fc9c4b4a7ff0f4aa982d0 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Thu, 15 Oct 2026 10:30:40 -0700 Subject: [PATCH 1/3] fix(ai-harness): a host that does not own a thread keeps its session events off the log A durable host that opened a thread while another host ran a turn wrote its plugin state (for example the MCP plugin's connection status) to the shared log. The owner's next append conflicted, and its turn stopped. The session now reads the claim before its plugins mount, and drops session events while another host owns the thread. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-session-events-owner.md | 5 ++ packages/ai-harness/src/session/base.ts | 12 ++++- .../ai-harness/src/session/harness-session.ts | 20 ++++++- packages/ai-harness/tests/forwarding.test.ts | 53 +++++++++++++++++++ .../e2e/src/routes/api.harness-forwarding.ts | 29 ++++++++-- testing/e2e/tests/harness-forwarding.spec.ts | 14 +++++ 6 files changed, 126 insertions(+), 7 deletions(-) create mode 100644 .changeset/harness-session-events-owner.md 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/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index c9ec860e05..a876479f27 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>() @@ -701,6 +707,7 @@ export abstract class SessionBase { until: Date.now() + ttlMs, }) this.claimLost = !isHeld + this.ownerElsewhere = !isHeld } catch (error) { // The work goes on without a claim. this.warnClaim(error) @@ -738,9 +745,9 @@ export abstract class SessionBase { const claims = this.persistence.stores.workClaims if (!this.writer || !claims) return false const held = await claims.get(this.threadId) - return ( + this.ownerElsewhere = held !== null && held.ownerId !== this.hostId && held.until > Date.now() - ) + return this.ownerElsewhere } /** @@ -792,6 +799,7 @@ export abstract class SessionBase { ownerId: this.hostId, until: Date.now() + (this.lease?.ttlMs ?? LEASE.ttlMs), }) + this.ownerElsewhere = !isHeld if (!isHeld) return // This host takes the thread from an owner that stopped. Its inputs // run first, as on an open. 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..72e57a4584 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -201,6 +201,59 @@ 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('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..5cc6e49758 100644 --- a/testing/e2e/src/routes/api.harness-forwarding.ts +++ b/testing/e2e/src/routes/api.harness-forwarding.ts @@ -1,6 +1,10 @@ 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' /** @@ -11,6 +15,9 @@ 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. * * 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 +74,9 @@ 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' let release!: () => void const released = new Promise((resolve) => { release = resolve @@ -92,21 +101,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 diff --git a/testing/e2e/tests/harness-forwarding.spec.ts b/testing/e2e/tests/harness-forwarding.spec.ts index a6260857ba..82981d190f 100644 --- a/testing/e2e/tests/harness-forwarding.spec.ts +++ b/testing/e2e/tests/harness-forwarding.spec.ts @@ -30,6 +30,20 @@ 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('is refused without an inbox, and the owner turn ends normally', async ({ request, }) => { From f83885b73a28dc1abf625a6618440b5390a0e7ef Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Fri, 16 Oct 2026 11:25:15 -0700 Subject: [PATCH 2/3] fix(ai-harness): a live run lease of another host means the thread is owned elsewhere On Netlify Play, Blobs requests stalled for ~10 s (the client retries for up to 25 s). The owner's claim renewals did not land and its claim expired, while its run lease renewed and its turn ran a tool. Another host opened the thread, took the claim, skipped the live turn in recovery, and ran the queued follow-ups. The transcript got a user message between a tool call and its result, and the model refused it. `isOwnedElsewhere` also reads the thread's active run: a live lease of another host counts. The inbox drain checks it before it claims. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-live-lease-owner.md | 5 + packages/ai-harness/src/session/base.ts | 17 ++- packages/ai-harness/tests/forwarding.test.ts | 66 +++++++++++ .../e2e/src/routes/api.harness-forwarding.ts | 104 ++++++++++++++++++ testing/e2e/tests/harness-forwarding.spec.ts | 19 ++++ 5 files changed, 210 insertions(+), 1 deletion(-) create mode 100644 .changeset/harness-live-lease-owner.md 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/packages/ai-harness/src/session/base.ts b/packages/ai-harness/src/session/base.ts index a876479f27..be2b85404f 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -745,8 +745,21 @@ export abstract class SessionBase { const claims = this.persistence.stores.workClaims if (!this.writer || !claims) return false const held = await claims.get(this.threadId) + 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 = - held !== null && held.ownerId !== this.hostId && held.until > Date.now() + run !== null && + run !== undefined && + run.leaseOwner !== undefined && + run.leaseOwner !== this.hostId && + (run.leaseExpiresAt ?? 0) > now return this.ownerElsewhere } @@ -793,6 +806,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/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index 72e57a4584..d5ab807bf0 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -254,6 +254,72 @@ describe('an input that reaches a host that does not own the thread', () => { 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.'), + ]) + 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 }) + await otherHost.open(harness, { threadId: THREAD }) + await otherHost.recover() + await new Promise((resolve) => setTimeout(resolve, 200)) + expect(calls).toHaveLength(1) + + isStalled = false + claimWrites.open() + toolDone.open() + expect((await turn).text).toBe('One.') + await vi.waitFor(() => expect(calls).toHaveLength(3), { timeout: 3_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.' }, + ]) + 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 5cc6e49758..c508ef3169 100644 --- a/testing/e2e/src/routes/api.harness-forwarding.ts +++ b/testing/e2e/src/routes/api.harness-forwarding.ts @@ -6,6 +6,13 @@ import { 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 @@ -18,6 +25,9 @@ import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' * - `?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 @@ -77,6 +87,7 @@ export const Route = createFileRoute('/api/harness-forwarding')({ 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 @@ -92,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(), @@ -161,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 82981d190f..6be8199af8 100644 --- a/testing/e2e/tests/harness-forwarding.spec.ts +++ b/testing/e2e/tests/harness-forwarding.spec.ts @@ -44,6 +44,25 @@ test.describe('harness: an input on a host that does not own the thread', () => 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, }) => { From d6d2400a33fb2dd96b4a0f42ea14d0e430eaba0e Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Fri, 16 Oct 2026 14:42:37 -0700 Subject: [PATCH 3/3] fix(ai-harness): cover the inbox drain in the live-lease test, document both rules The inbox drain guard is now proven by the test: an input that waits in the inbox for an owner whose claim lapsed stays there until the owner's turn ends. Drop the ownerElsewhere resets after a claim: every accept() reads ownership again. Docs: a live run lease makes the owner, and a host that does not own a thread writes nothing to its log. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/config.json | 2 +- docs/harness/durable-sessions.md | 6 ++++-- packages/ai-harness/src/session/base.ts | 2 -- packages/ai-harness/tests/forwarding.test.ts | 13 ++++++++++--- 4 files changed, 15 insertions(+), 8 deletions(-) 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 be2b85404f..c0d84d4eea 100644 --- a/packages/ai-harness/src/session/base.ts +++ b/packages/ai-harness/src/session/base.ts @@ -707,7 +707,6 @@ export abstract class SessionBase { until: Date.now() + ttlMs, }) this.claimLost = !isHeld - this.ownerElsewhere = !isHeld } catch (error) { // The work goes on without a claim. this.warnClaim(error) @@ -814,7 +813,6 @@ export abstract class SessionBase { ownerId: this.hostId, until: Date.now() + (this.lease?.ttlMs ?? LEASE.ttlMs), }) - this.ownerElsewhere = !isHeld if (!isHeld) return // This host takes the thread from an owner that stopped. Its inputs // run first, as on an open. diff --git a/packages/ai-harness/tests/forwarding.test.ts b/packages/ai-harness/tests/forwarding.test.ts index d5ab807bf0..116d146fbd 100644 --- a/packages/ai-harness/tests/forwarding.test.ts +++ b/packages/ai-harness/tests/forwarding.test.ts @@ -280,6 +280,7 @@ describe('an input that reaches a host that does not own the thread', () => { () => toolCall('wait', {}), () => text('One.'), () => text('Two.'), + () => text('Three.'), ]) const harness = defineHarness({ name: NAME, adapter, tools: [wait] }) const lease = { renewMs: 50, ttlMs: 300 } @@ -298,16 +299,20 @@ describe('an input that reaches a host that does not own the thread', () => { // Another host opens the thread after the claim expired. const otherHost = createHarnessHost({ persistence, lease }) - await otherHost.open(harness, { threadId: THREAD }) + const other = await otherHost.open(harness, { threadId: THREAD }) await otherHost.recover() - await new Promise((resolve) => setTimeout(resolve, 200)) + // 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(3), { timeout: 3_000 }) + await vi.waitFor(() => expect(calls).toHaveLength(4), { timeout: 5_000 }) expect(await owner.transcript()).toMatchObject([ { role: 'user', content: 'Go.' }, { role: 'assistant' }, @@ -315,6 +320,8 @@ describe('an input that reaches a host that does not own the thread', () => { { 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()