From d11a5985e27064c3ab61b5d3f08b6626f4bb0718 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 09:57:45 -0700 Subject: [PATCH 1/3] fix(ai-harness): no insecure or single-process transport defaults - `canAccess` is now required on `createHarnessHandler` and `handleHarnessSocket`. Forgetting it used to let every principal into every thread. Allow-all is now an explicit `canAccess: () => true`. - A durable host (`stores.log`) runs behind several instances, so a random per-handler `mediaSecret` makes signed media URLs fail on every other instance. The handler now throws at creation on a durable host without `mediaSecret`. A host without a log store keeps the random default. - `oauthConnector` no longer defaults to `login: 'loopback'`, which opened a node:http listener on 127.0.0.1 inside the server process, where the user's browser can never reach it. `login` is now required. A default of `'device'` was rejected: it needs `oauth.deviceUrl`, which many apps do not have, so a silent default would just move the failure to run time. Requiring the choice is the smallest change that is correct for both a CLI (`'loopback'`) and a server (`'device'`). Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-secure-defaults.md | 9 ++++ docs/harness/auth.md | 6 ++- docs/harness/connect.md | 12 ++++- packages/ai-harness/src/connectors.ts | 14 +++++- packages/ai-harness/src/host.ts | 15 ++++++ packages/ai-harness/src/http.ts | 47 +++++++++++-------- packages/ai-harness/src/index.ts | 1 + packages/ai-harness/tests/auth-wait.test.ts | 3 ++ packages/ai-harness/tests/client.test.ts | 6 +++ packages/ai-harness/tests/http-media.test.ts | 25 +++++++++- packages/ai-harness/tests/oauth-edges.test.ts | 1 + packages/ai-harness/tests/protocol.test.ts | 8 +++- packages/ai-harness/tests/run-route.test.ts | 5 +- .../ai-harness/tests/transport-edges.test.ts | 18 ++++++- packages/ai-harness/tests/types.test-d.ts | 24 +++++++++- .../tests/view-client-tools.test.ts | 1 + packages/ai-harness/tests/view.test.ts | 2 + .../e2e/src/routes/api.harness-protocol.$.ts | 2 + 18 files changed, 167 insertions(+), 32 deletions(-) create mode 100644 .changeset/harness-secure-defaults.md diff --git a/.changeset/harness-secure-defaults.md b/.changeset/harness-secure-defaults.md new file mode 100644 index 0000000000..d8f7ec8847 --- /dev/null +++ b/.changeset/harness-secure-defaults.md @@ -0,0 +1,9 @@ +--- +'@tanstack/ai-harness': patch +--- + +Make the transport defaults safe for a server with many instances. + +- `createHarnessHandler` and `handleHarnessSocket` need `canAccess`. To let every principal use every thread, pass `canAccess: () => true`. +- On a durable host (`stores.log`), `createHarnessHandler` throws without `mediaSecret`, so a media URL that one instance signs works on every instance. +- `oauthConnector` needs `login`. Use `'device'` on a server, and `'loopback'` only for a harness on the user's own machine, such as a CLI. diff --git a/docs/harness/auth.md b/docs/harness/auth.md index 0d17498f3d..d825439b85 100644 --- a/docs/harness/auth.md +++ b/docs/harness/auth.md @@ -23,6 +23,8 @@ import { openaiText } from '@tanstack/ai-openai' const github = oauthConnector({ id: 'github', label: 'GitHub', + // The harness runs on the user's machine (a CLI), so open a browser here. + login: 'loopback', oauth: { authorizationUrl: 'https://github.com/login/oauth/authorize', tokenUrl: 'https://github.com/login/oauth/access_token', @@ -54,10 +56,10 @@ export const assistant = defineHarness({ ## 2. Sign in - The user runs `/connect github`. -- The CLI opens the browser. The harness listens on `127.0.0.1` on a random port for one callback, with PKCE and a random `state`. +- With `login: 'loopback'`, the CLI opens the browser. The harness listens on `127.0.0.1` on a random port for one callback, with PKCE and a random `state`. - The token goes into the credential store. `/disconnect github` deletes it. -Set `login: 'device'` for SSH sessions, containers, and CI. The CLI then shows a code to enter on the service's page. +`login` is required. Use `login: 'device'` for a harness on a server, and for SSH sessions, containers, and CI. The user then gets a code to enter on the service's page. Device login needs `oauth.deviceUrl`. If a tool runs before sign-in, it stops with a `harness.auth_required` event. The CLI shows which service to connect. diff --git a/docs/harness/connect.md b/docs/harness/connect.md index a98a5904ef..3c9803a2b8 100644 --- a/docs/harness/connect.md +++ b/docs/harness/connect.md @@ -40,6 +40,10 @@ export const handler = createHarnessHandler({ }) ``` +`canAccess` is required. It decides which threads each principal can use. To let every principal use every thread, pass `canAccess: () => true`. + +On a durable host (`stores.log`), also set `mediaSecret`. It signs media URLs, so every instance must use the same value. Without a log store, the default is a random key, and signed URLs stop working after a restart. + The handler answers these paths under your route: - `GET capabilities`: the AG-UI capabilities, with the agents in `expose.agents`. @@ -302,7 +306,13 @@ import { handleHarnessSocket } from '@tanstack/ai-harness' import type { WebSocketLike } from '@tanstack/ai' export function onUpgrade(socket: WebSocketLike) { - handleHarnessSocket({ host, harness: assistant, socket, principal: { id: 'user-1' } }) + handleHarnessSocket({ + host, + harness: assistant, + socket, + principal: { id: 'user-1' }, + canAccess: (principal, threadId) => threadId.startsWith(`${principal.id}-`), + }) } ``` diff --git a/packages/ai-harness/src/connectors.ts b/packages/ai-harness/src/connectors.ts index c7f55d8c4f..bd5639d999 100644 --- a/packages/ai-harness/src/connectors.ts +++ b/packages/ai-harness/src/connectors.ts @@ -14,8 +14,17 @@ export interface OAuthConnectorOptions { id: string label: string oauth: OAuthConfig - /** `'loopback'` (default) opens a browser here. `'device'` shows a code to enter elsewhere. */ - login?: 'loopback' | 'device' + /** + * How the user signs in. Required, because only you know where the harness + * runs: + * + * - `'device'`: show a code to enter on another device. It works on a + * server, and needs `oauth.deviceUrl`. + * - `'loopback'`: open a browser on this machine and listen on + * `127.0.0.1` for the redirect. Only for a harness that runs on the + * user's own machine, such as a CLI. + */ + login: 'loopback' | 'device' /** * The tools of this service. Call `token()` inside a tool: it returns a * fresh access token, or fails with `auth_required` before sign-in. With @@ -39,6 +48,7 @@ export interface OAuthConnectorOptions { * const github = oauthConnector({ * id: 'github', * label: 'GitHub', + * login: 'device', * oauth: { authorizationUrl, tokenUrl, deviceUrl, clientId, scopes: ['repo'] }, * tools: (token) => [listIssues(token)], * }) diff --git a/packages/ai-harness/src/host.ts b/packages/ai-harness/src/host.ts index 65bcd5d109..cffbfdc675 100644 --- a/packages/ai-harness/src/host.ts +++ b/packages/ai-harness/src/host.ts @@ -518,6 +518,17 @@ export function sessionOfTurn( ) } +/** The log and run stores of each durable host. Not public API. */ +const durableStores = new WeakMap() + +/** + * @internal The log and run stores of `host` when it is durable + * (`stores.log`), else `undefined`. + */ +export function durableStoresOf(host: HarnessHost) { + return durableStores.get(host) +} + /** Throw a clear error for a durable store set that cannot work. */ function checkDurableStores(persistence: HarnessPersistence) { const { stores } = persistence @@ -944,6 +955,10 @@ export function createHarnessHost( } }, } + if (logStore) { + const { runs } = persistence.stores + durableStores.set(host, { log: logStore, ...(runs ? { runs } : {}) }) + } turnFinders.set(host, async (name, operationId) => { for (const [key, opening] of sessions) { if (!key.startsWith(`${name}\u0000`)) continue diff --git a/packages/ai-harness/src/http.ts b/packages/ai-harness/src/http.ts index 5cc8e03e9e..9fbc87c2fd 100644 --- a/packages/ai-harness/src/http.ts +++ b/packages/ai-harness/src/http.ts @@ -10,7 +10,7 @@ import { toServerSentEventsStream, } from '@tanstack/ai' import { parseRangeHeader, resolveBlobRange } from '@tanstack/ai-persistence' -import { sessionOfTurn } from './host' +import { durableStoresOf, sessionOfTurn } from './host' import { MediaError } from './media' import { base64url } from './oauth' import { @@ -45,22 +45,28 @@ export type Authorize = ( request: Request, ) => Principal | null | Promise +/** May this principal use this thread? */ +export type CanAccess = ( + principal: Principal, + threadId: string, +) => boolean | Promise + export interface HarnessHandlerOptions { host: HarnessHost harness: AnyHarness authorize: Authorize /** - * May this principal use this thread? Default: yes. Use it to keep users - * out of each other's threads. + * May this principal use this thread? Required: use it to keep users out + * of each other's threads. Pass `() => true` to let every principal use + * every thread. */ - canAccess?: ( - principal: Principal, - threadId: string, - ) => boolean | Promise + canAccess: CanAccess /** - * The secret that signs media URLs (HMAC SHA-256). Default: a random key - * per handler, so signed URLs stop working after a restart. Set it to keep - * them working, and use the same value on every server behind one URL. + * The secret that signs media URLs (HMAC SHA-256). Use the same value on + * every server behind one URL. Required on a durable host (`stores.log`), + * where another instance can serve the next request. Without a log store, + * the default is a random key per handler, so signed URLs stop working + * after a restart. */ mediaSecret?: string } @@ -309,16 +315,20 @@ async function mediaResponse( export function createHarnessHandler( options: HarnessHandlerOptions, ): (request: Request) => Promise { - const { host, harness, authorize } = options - const canAccess = options.canAccess ?? (() => true) + const { host, harness, authorize, canAccess } = options + if (options.mediaSecret === undefined && durableStoresOf(host)) { + throw new Error( + 'createHarnessHandler needs mediaSecret on a durable host (stores.log): another instance must verify the media URLs this one signs. Set the same secret on every instance.', + ) + } const openFor = async (principal: Principal, threadId: string) => { if (!(await canAccess(principal, threadId))) return null return host.open(harness, { threadId, principal }) } - // ponytail: without `mediaSecret`, a random key per handler, so signed URLs - // stop working after a restart. Set `mediaSecret` to keep them. + // ponytail: without `mediaSecret` (only allowed without a log store), a + // random key per handler, so signed URLs stop working after a restart. const secret = options.mediaSecret === undefined ? crypto.getRandomValues(new Uint8Array(32)) @@ -757,10 +767,8 @@ export interface HarnessSocketOptions { socket: WebSocketLike /** The principal your upgrade handler authorized. */ principal: Principal - canAccess?: ( - principal: Principal, - threadId: string, - ) => boolean | Promise + /** May this principal use this thread? Pass `() => true` to allow all. */ + canAccess: CanAccess } /** @@ -768,8 +776,7 @@ export interface HarnessSocketOptions { * `harness.subscribe`. Authorize the upgrade request before you call this. */ export function handleHarnessSocket(options: HarnessSocketOptions): void { - const { host, harness, socket, principal } = options - const canAccess = options.canAccess ?? (() => true) + const { host, harness, socket, principal, canAccess } = options const reader = new AbortController() // Set by the first subscribe, before its awaits, so a second subscribe is // refused and an input sent right after the subscribe waits for the open. diff --git a/packages/ai-harness/src/index.ts b/packages/ai-harness/src/index.ts index 87ec4c6dd0..ddba7ff167 100644 --- a/packages/ai-harness/src/index.ts +++ b/packages/ai-harness/src/index.ts @@ -179,6 +179,7 @@ export type { ControlFrame, HostFrame } from './protocol' export { createHarnessHandler, handleHarnessSocket } from './http' export type { Authorize, + CanAccess, HarnessHandlerOptions, HarnessSocketOptions, } from './http' diff --git a/packages/ai-harness/tests/auth-wait.test.ts b/packages/ai-harness/tests/auth-wait.test.ts index 836df9584e..5750f650e0 100644 --- a/packages/ai-harness/tests/auth-wait.test.ts +++ b/packages/ai-harness/tests/auth-wait.test.ts @@ -385,6 +385,7 @@ describe('sign-ins in a shared thread', () => { const handler = createHarnessHandler({ host, harness, + canAccess: () => true, authorize: (request) => ({ id: request.headers.get('x-user') ?? '' }), }) const control = async (user: string, input: object) => @@ -459,7 +460,9 @@ describe('sign-ins in a shared thread', () => { const handler = createHarnessHandler({ host, harness, + canAccess: () => true, authorize: () => bob, + mediaSecret: 'test-secret', }) const session = await host.open(harness, { threadId: 't', diff --git a/packages/ai-harness/tests/client.test.ts b/packages/ai-harness/tests/client.test.ts index c53af3d337..919102f763 100644 --- a/packages/ai-harness/tests/client.test.ts +++ b/packages/ai-harness/tests/client.test.ts @@ -26,6 +26,7 @@ describe('createHarnessClient', () => { const handler = createHarnessHandler({ host, harness: studio, + canAccess: () => true, authorize: (request) => request.headers.get('authorization') === 'Bearer t' ? { id: 'u' } @@ -71,6 +72,7 @@ describe('createHarnessClient', () => { const handler = createHarnessHandler({ host, harness: studio, + canAccess: () => true, authorize: () => ({ id: 'u' }), }) const client = createHarnessClient({ @@ -102,6 +104,7 @@ describe('createHarnessClient', () => { const handler = createHarnessHandler({ host, harness: studio, + canAccess: () => true, authorize: () => ({ id: 'u' }), }) const client = createHarnessClient({ @@ -133,6 +136,7 @@ describe('createHarnessClient', () => { const handler = createHarnessHandler({ host, harness: studio, + canAccess: () => true, authorize: () => ({ id: 'u' }), }) const client = createHarnessClient({ @@ -164,6 +168,7 @@ describe('createHarnessClient', () => { const handler = createHarnessHandler({ host, harness: studio, + canAccess: () => true, authorize: () => null, }) const client = createHarnessClient({ @@ -201,6 +206,7 @@ describe('client.sendToAgent', () => { const handler = createHarnessHandler({ host, harness: studio, + canAccess: () => true, authorize: () => ({ id: 'u' }), }) const client = createHarnessClient({ diff --git a/packages/ai-harness/tests/http-media.test.ts b/packages/ai-harness/tests/http-media.test.ts index 9d4ea42057..ac280484d8 100644 --- a/packages/ai-harness/tests/http-media.test.ts +++ b/packages/ai-harness/tests/http-media.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, it, vi } from 'vitest' -import { memoryPersistence } from '@tanstack/ai-persistence' +import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' import { createHarnessHandler, createHarnessHost, @@ -127,6 +127,29 @@ afterEach(() => { }) describe('media upload and signed URLs', () => { + it('needs mediaSecret on a durable host, where another instance verifies the URL', () => { + const { adapter } = mockAdapter(() => text('ok')) + const harness = defineHarness({ name: 'test/http-media-durable', adapter }) + const host = createHarnessHost({ + persistence: { + stores: { + log: memoryLogStore(), + runs: memoryPersistence().stores.runs, + }, + }, + }) + const options = { + host, + harness, + authorize: allowUser, + canAccess: () => true, + } + expect(() => createHarnessHandler(options)).toThrow(/mediaSecret/) + expect(() => + createHarnessHandler({ ...options, mediaSecret: 'shared' }), + ).not.toThrow() + }) + it('stores an upload, signs its URL for 1 hour, and serves the signed URL without auth', async () => { vi.useFakeTimers({ toFake: ['Date'], now: NOW }) const { host, upload, signedPath, fetchSigned } = setup() diff --git a/packages/ai-harness/tests/oauth-edges.test.ts b/packages/ai-harness/tests/oauth-edges.test.ts index b23689268e..f8aba7fecf 100644 --- a/packages/ai-harness/tests/oauth-edges.test.ts +++ b/packages/ai-harness/tests/oauth-edges.test.ts @@ -208,6 +208,7 @@ describe('oauthConnector edges', () => { return oauthConnector({ id: 'svc', label: 'Service', + login: 'loopback', oauth, fetch: fetchFn, tools: (token) => [ diff --git a/packages/ai-harness/tests/protocol.test.ts b/packages/ai-harness/tests/protocol.test.ts index b2a6823c6e..7532aa3e8c 100644 --- a/packages/ai-harness/tests/protocol.test.ts +++ b/packages/ai-harness/tests/protocol.test.ts @@ -416,7 +416,13 @@ describe('WebSocket', () => { it('subscribes, applies inputs with receipts, and streams events', async () => { const { host, harness } = setup() const { socket, sent, receive } = fakeSocket() - handleHarnessSocket({ host, harness, socket, principal: { id: 'user-1' } }) + handleHarnessSocket({ + host, + harness, + socket, + principal: { id: 'user-1' }, + canAccess: () => true, + }) receive({ type: 'harness.input', diff --git a/packages/ai-harness/tests/run-route.test.ts b/packages/ai-harness/tests/run-route.test.ts index 7c88b38fc6..a930c69d28 100644 --- a/packages/ai-harness/tests/run-route.test.ts +++ b/packages/ai-harness/tests/run-route.test.ts @@ -37,7 +37,7 @@ const deploy = toolDefinition({ function setup( harness: AnyHarness, - canAccess?: () => boolean, + canAccess: () => boolean = () => true, persistence = memoryPersistence(), ) { const host = createHarnessHost({ persistence }) @@ -45,7 +45,7 @@ function setup( host, harness, authorize: () => ({ id: 'user-1' }), - ...(canAccess ? { canAccess } : {}), + canAccess, }) const call = (path: string, body?: object) => handler( @@ -235,6 +235,7 @@ describe('POST run', () => { const handler = createHarnessHandler({ host, harness: defineHarness({ name: 'test/run-route', adapter }), + canAccess: () => true, authorize: (request) => ({ id: request.headers.get('x-user') ?? '' }), }) const post = (user: string) => diff --git a/packages/ai-harness/tests/transport-edges.test.ts b/packages/ai-harness/tests/transport-edges.test.ts index 1ba4b5a983..48ae11d058 100644 --- a/packages/ai-harness/tests/transport-edges.test.ts +++ b/packages/ai-harness/tests/transport-edges.test.ts @@ -279,6 +279,7 @@ describe('WebSocket edges', () => { harness, socket: first.socket, principal: { id: 'user-1' }, + canAccess: () => true, }) first.receive({ type: 'harness.subscribe', threadId: 'user-1-ws2' }) await vi.waitFor(() => @@ -324,6 +325,7 @@ describe('WebSocket edges', () => { harness, socket: second.socket, principal: { id: 'user-1' }, + canAccess: () => true, }) second.receive({ type: 'harness.subscribe', @@ -346,7 +348,13 @@ describe('WebSocket edges', () => { it('takes an input sent right after the subscribe, before the hello', async () => { const { host, harness } = setup([() => text('hello')]) const { socket, sent, receive } = fakeSocket() - handleHarnessSocket({ host, harness, socket, principal: { id: 'user-1' } }) + handleHarnessSocket({ + host, + harness, + socket, + principal: { id: 'user-1' }, + canAccess: () => true, + }) receive({ type: 'harness.subscribe', threadId: 'user-1-ws3' }) receive({ type: 'harness.input', @@ -366,7 +374,13 @@ describe('WebSocket edges', () => { it('stops reading when the socket cannot send', async () => { const { host, harness } = setup() const { socket, receive } = fakeSocket({ failSend: true }) - handleHarnessSocket({ host, harness, socket, principal: { id: 'user-1' } }) + handleHarnessSocket({ + host, + harness, + socket, + principal: { id: 'user-1' }, + canAccess: () => true, + }) receive({ type: 'harness.subscribe', threadId: 'user-1-dead' }) // Nothing to assert on the socket: the send failure must not throw. await new Promise((resolve) => setTimeout(resolve, 20)) diff --git a/packages/ai-harness/tests/types.test-d.ts b/packages/ai-harness/tests/types.test-d.ts index 14b9f17727..b0bcd6596c 100644 --- a/packages/ai-harness/tests/types.test-d.ts +++ b/packages/ai-harness/tests/types.test-d.ts @@ -1,7 +1,13 @@ import { expectTypeOf, it } from 'vitest' import { z } from 'zod' import { defineAgent } from '@tanstack/ai' -import { defineHarness, definePlugin } from '../src' +import { + createHarnessHandler, + createHarnessHost, + defineHarness, + definePlugin, + oauthConnector, +} from '../src' import { mockAdapter } from './helpers' import type { AnyTextAdapter, KeyedAdapter } from '@tanstack/ai' import type { SessionIndexEntry } from '@tanstack/ai-persistence' @@ -158,3 +164,19 @@ it('refuses a plugin whose metadata the harness does not have', () => { plugins: () => [other], }) }) + +it('needs an explicit canAccess and an explicit login mode', () => { + const host = createHarnessHost() + // @ts-expect-error `canAccess` is required: allow-all must be explicit. + createHarnessHandler({ host, harness, authorize: () => ({ id: 'u' }) }) + createHarnessHandler({ + host, + harness, + authorize: () => ({ id: 'u' }), + canAccess: () => true, + }) + const oauth = { authorizationUrl: 'a', tokenUrl: 't', clientId: 'c' } + // @ts-expect-error `login` is required: no implicit loopback listener. + oauthConnector({ id: 'svc', label: 'Service', oauth }) + oauthConnector({ id: 'svc', label: 'Service', oauth, login: 'device' }) +}) diff --git a/packages/ai-harness/tests/view-client-tools.test.ts b/packages/ai-harness/tests/view-client-tools.test.ts index 87763be74e..9522b91806 100644 --- a/packages/ai-harness/tests/view-client-tools.test.ts +++ b/packages/ai-harness/tests/view-client-tools.test.ts @@ -49,6 +49,7 @@ describe('session view client tools', () => { const handler = createHarnessHandler({ host, harness, + canAccess: () => true, authorize: () => ({ id: 'u' }), }) const controls: Array = [] diff --git a/packages/ai-harness/tests/view.test.ts b/packages/ai-harness/tests/view.test.ts index 88b88c4423..0fe18b7fcb 100644 --- a/packages/ai-harness/tests/view.test.ts +++ b/packages/ai-harness/tests/view.test.ts @@ -309,6 +309,7 @@ describe('createSessionView', () => { const handler = createHarnessHandler({ host, harness, + canAccess: () => true, authorize: () => ({ id: 'u' }), }) const client = createHarnessClient({ @@ -345,6 +346,7 @@ describe('createSessionView', () => { const handler = createHarnessHandler({ host, harness, + canAccess: () => true, authorize: () => ({ id: 'u' }), }) const view = createSessionView( diff --git a/testing/e2e/src/routes/api.harness-protocol.$.ts b/testing/e2e/src/routes/api.harness-protocol.$.ts index 2ada40511f..7f979ab5fe 100644 --- a/testing/e2e/src/routes/api.harness-protocol.$.ts +++ b/testing/e2e/src/routes/api.harness-protocol.$.ts @@ -284,6 +284,8 @@ function handlerFor(request: Request) { host: hostOf, harness, authorize, + // Test only: every principal may use every thread. + canAccess: () => true, // Each request makes a new handler, so a fixed secret keeps a signed URL // working on the next request. mediaSecret: 'e2e-media-secret', From 0fd47ab6a21fa050c5ea4ac6132201dbb9639464 Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:02:00 -0700 Subject: [PATCH 2/3] fix(ai-harness): join and resume a run on any instance of a durable host On serverless hosting, several instances serve one URL with no sticky routing, and a connection can be cut at any time. - `POST run` streamed with no `id:` lines, so a cut stream could not resume. On a durable host each event now carries its log cursor as `id` (the same cursors as `GET events`). The ai-client SSE adapters already re-send the request with `Last-Event-ID`: the run id is the input id, so the retry finds the same turn (live, or replayed from the log on another instance) and streams after that cursor, without running it again. The `resume` branch of `POST run` does the same. - `GET run?runId=` only found turns in this process. When the turn is not local, a durable host now looks up its thread in `stores.runs` (the run id is the operation id, and every chat turn has a run record), opens the thread, and streams that operation's `harness.event` records from the shared log, from the start or `Last-Event-ID`, live until `harness.operation.finished`. Agent runs and child runs stay 404. - `GET run?threadId=` also reports a turn another instance runs as `activeRun` (from `runs.findActiveRun`), so a reloaded ChatClient on another instance knows to join it. Without a log store nothing changes: no ids on `POST run`, since a process-local cursor means nothing on another instance. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-join-any-instance.md | 9 + docs/harness/connect.md | 5 +- packages/ai-harness/src/http.ts | 156 +++++++++++----- .../ai-harness/tests/run-route-hosts.test.ts | 175 ++++++++++++++++++ 4 files changed, 293 insertions(+), 52 deletions(-) create mode 100644 .changeset/harness-join-any-instance.md create mode 100644 packages/ai-harness/tests/run-route-hosts.test.ts diff --git a/.changeset/harness-join-any-instance.md b/.changeset/harness-join-any-instance.md new file mode 100644 index 0000000000..d7089c0ae3 --- /dev/null +++ b/.changeset/harness-join-any-instance.md @@ -0,0 +1,9 @@ +--- +'@tanstack/ai-harness': patch +--- + +On a durable host (`stores.log` and `stores.runs`), any server instance can serve a run. + +- `POST run` sends the log cursor of each event as its SSE `id`. A client that is cut off sends the same request with `Last-Event-ID`, and any instance continues the stream from there. The turn does not run again. +- `GET run?runId=` finds a turn that another instance runs. It streams the turn from the shared log, from its start or from `Last-Event-ID`, until the turn ends. +- `GET run?threadId=` shows a turn that another instance runs as `activeRun`, so a reloaded `useChat` joins it. diff --git a/docs/harness/connect.md b/docs/harness/connect.md index 3c9803a2b8..3510c8ac3e 100644 --- a/docs/harness/connect.md +++ b/docs/harness/connect.md @@ -275,7 +275,7 @@ The approval reaches the harness, and the tool runs on the server. What else `PO - The `forwardedProps` of the request are the input's [context](./inputs#send-context-with-a-message). - With `persistence: true`, `useChat` loads the thread from `GET run?threadId=`. - If a turn still runs after a reload, `useChat` joins it with `GET run?runId=`. The turn shows from its first event, and its answer continues to stream. -- With more than one server, send `GET run?runId=` to the server that runs the turn. Only that server has the live events of the turn. +- On a durable host (`stores.log` and `stores.runs`), each event of `POST run` has its log cursor as its id. If the connection drops, `useChat` sends the request again with `Last-Event-ID`. Any server continues the stream from there, and the turn does not run again. To keep the thread after a reload, set `persistence: true` in `useChat`: @@ -294,7 +294,8 @@ export function ReloadableChat() { The join has these rules: - `canAccess` decides who can join, the same as for every other path. -- The turn must run on the host that answers the request. On another host, the join gets `404`. +- On a durable host, any server can answer the join. It reads the turn from the shared log. +- Without a log store, the turn must run on the host that answers the request. On another host, the join gets `404`. - A turn that ended gives the events that the session still keeps, and then the stream ends. ## Use a WebSocket diff --git a/packages/ai-harness/src/http.ts b/packages/ai-harness/src/http.ts index 9fbc87c2fd..9f71baf959 100644 --- a/packages/ai-harness/src/http.ts +++ b/packages/ai-harness/src/http.ts @@ -6,7 +6,6 @@ import { modelMessagesToUIMessages, readInterruptBinding, resolveResumeRunId, - toServerSentEventsResponse, toServerSentEventsStream, } from '@tanstack/ai' import { parseRangeHeader, resolveBlobRange } from '@tanstack/ai-persistence' @@ -25,7 +24,7 @@ import { isRecord } from './utils' import { SessionMetadataError } from './session-entry' import type { ModelMessage, - StreamChunk, + RunStore, UIMessage, WebSocketLike, } from '@tanstack/ai' @@ -34,7 +33,7 @@ import type { AnyHarness } from './define' import type { HarnessHost, HostEvent } from './host' import type { HostFrame } from './protocol' import type { HarnessSession } from './session/index' -import type { Principal, UserInput } from './types' +import type { Principal, SessionEvent, UserInput } from './types' /** * Decide who sends a request. Return `null` to refuse it with 401. Every @@ -127,6 +126,34 @@ function sseResponse( return new Response(body, { headers: SSE_HEADERS }) } +/** + * Session events as an AG-UI SSE response. With `withIds`, each event has its + * cursor as `id`, so a client that loses the connection resumes with + * `Last-Event-ID`. When the client leaves, the read stops. + */ +function cursorResponse( + read: (signal: AbortSignal) => AsyncIterable, + withIds: boolean, +) { + // Not `request.signal`: some servers abort it once the body is read. + const reader = new AbortController() + const cursors: Array = [] + async function* chunks() { + for await (const entry of read(reader.signal)) { + cursors.push(entry.cursor) + yield entry.event + } + } + return new Response( + toServerSentEventsStream( + chunks(), + reader, + withIds ? (_chunk, index) => cursors.at(index) : undefined, + ), + { headers: SSE_HEADERS }, + ) +} + /** A handler error as JSON: a `MediaError` keeps its status, the rest is 400. */ const failed = (error: unknown) => error instanceof MediaError @@ -260,7 +287,8 @@ async function mediaResponse( * - `GET .../capabilities`: the AG-UI capabilities document. * - `POST .../run`: standard AG-UI. One request is one prompt (or one * `resume`), streamed as SSE. The turn runs as the request's `runId`, and a - * retry with the same `runId` runs once. The prompt keeps every content + * retry with the same `runId` runs once. On a durable host, each event id + * is its log cursor, and a retry with `Last-Event-ID` streams from there. The prompt keeps every content * part of the last user message, and its `forwardedProps` are the input's * `context`. * - `GET .../run?threadId=`: the transcript, the running turn, and the @@ -268,8 +296,9 @@ async function mediaResponse( * `persistence` reads. * - `GET .../run?runId=` (or `X-Run-Id`): the chat turn with that run id, as * SSE from its start, so a reloaded `ChatClient` joins it. Each event id is - * a cursor, so `Last-Event-ID` resumes it. 404 for a turn this host does - * not run. + * a cursor, so `Last-Event-ID` resumes it. On a durable host (`stores.log` + * and `stores.runs`), any instance finds the turn in the shared log. + * Without a log, 404 for a turn this host does not run. * - `GET .../events?threadId=&from=`: the session stream as SSE. Each event id * is a cursor, so `Last-Event-ID` resumes it. * - `POST .../control`: `{ threadId, input }`. Returns the receipt. @@ -316,7 +345,8 @@ export function createHarnessHandler( options: HarnessHandlerOptions, ): (request: Request) => Promise { const { host, harness, authorize, canAccess } = options - if (options.mediaSecret === undefined && durableStoresOf(host)) { + const durable = durableStoresOf(host) + if (options.mediaSecret === undefined && durable) { throw new Error( 'createHarnessHandler needs mediaSecret on a durable host (stores.log): another instance must verify the media URLs this one signs. Set the same secret on every instance.', ) @@ -382,34 +412,49 @@ export function createHarnessHandler( principal: Principal, runId: string, ) => { - const session = await sessionOfTurn(host, harness, runId) - const turn = session?.operation(runId) - if (!session || !turn) return json({ error: 'unknown run' }, 404) - if (!(await canAccess(principal, session.threadId))) { - return json({ error: 'forbidden' }, 403) - } // `offset=-1` is the start. A reconnect sends the last cursor it saw. const offset = new URL(request.url).searchParams.get('offset') const resumeFrom = request.headers.get('Last-Event-ID') ?? (offset === '-1' ? null : offset) - const startedCursor = session - .snapshot() - .activeOperations.find((item) => item.id === runId)?.startedCursor - const from = resumeFrom ?? startedCursor ?? '0' - const reader = new AbortController() - const events = turn.events({ from, signal: reader.signal }) - const cursors: Array = [] - async function* chunks() { - for await (const entry of events) { - cursors.push(entry.cursor) - yield entry.event + const session = await sessionOfTurn(host, harness, runId) + const turn = session?.operation(runId) + if (session && turn) { + if (!(await canAccess(principal, session.threadId))) { + return json({ error: 'forbidden' }, 403) } + const startedCursor = session + .snapshot() + .activeOperations.find((item) => item.id === runId)?.startedCursor + const from = resumeFrom ?? startedCursor ?? '0' + return cursorResponse((signal) => turn.events({ from, signal }), true) + } + // Another instance runs the turn. A durable host finds its thread in the + // run store, and its events in the shared log. Not a child run or an + // agent run: only a chat turn of a thread. + const record = await durable?.runs?.get(runId) + const isChatTurn = + record != null && + record.kind !== 'agent' && + record.parentRunId === undefined + if (!record || !isChatTurn) return json({ error: 'unknown run' }, 404) + const { threadId } = record + if (!(await canAccess(principal, threadId))) { + return json({ error: 'forbidden' }, 403) + } + const entry = await host.sessions.get(threadId) + if (entry?.harness !== undefined && entry.harness !== harness.name) { + return json({ error: 'unknown run' }, 404) } - return new Response( - toServerSentEventsStream(chunks(), reader, (_chunk, index) => - cursors.at(index), - ), - { headers: SSE_HEADERS }, + const shared = await host.open(harness, { threadId, principal }) + // ponytail: from the start of the log without a cursor. Store the + // started position on the run record if long threads make this slow. + return cursorResponse( + (signal) => + followOperation( + shared.events({ from: resumeFrom ?? '0', signal }), + runId, + ), + true, ) } @@ -439,20 +484,23 @@ export function createHarnessHandler( // its events and interrupts. The run id is the input id too, so a // retried request runs once. It runs as this request's principal. const ids = { inputId: params.runId, runId: params.runId, principal } + const lastEventId = request.headers.get('Last-Event-ID') if (params.resume && params.resume.length > 0) { const receipt = await session.resolve(params.resume, ids) if (receipt.status === 'rejected' || !receipt.operationId) { return json({ error: receipt.reason ?? 'rejected' }, 409) } - // Not `request.signal`: some servers abort it once the body is - // read. The response aborts this controller when the client leaves. - const reader = new AbortController() - const events = session.events({ signal: reader.signal }) - return toServerSentEventsResponse( - followOperation(events, receipt.operationId), - { - abortController: reader, - }, + const { operationId } = receipt + return cursorResponse( + (signal) => + followOperation( + session.events({ + ...(durable && lastEventId ? { from: lastEventId } : {}), + signal, + }), + operationId, + ), + durable !== undefined, ) } if (message === undefined) @@ -469,12 +517,14 @@ export function createHarnessHandler( if (receipt.status === 'rejected') { return json({ error: receipt.reason ?? 'rejected' }, 409) } - const reader = new AbortController() - return toServerSentEventsResponse( - operation.stream({ signal: reader.signal }), - { - abortController: reader, - }, + // A durable host sends each log cursor as the event id. A client cut + // off sends the same request with `Last-Event-ID`, maybe to another + // instance: the run id finds the same turn, and it streams from there. + // Without a log a cursor means nothing on another instance, so no id. + const from = (durable && lastEventId) || '0' + return cursorResponse( + (signal) => operation.events({ from, signal }), + durable !== undefined, ) } @@ -486,7 +536,7 @@ export function createHarnessHandler( if (!threadId) return json({ error: 'threadId is required' }, 400) const session = await openFor(principal, threadId) if (!session) return json({ error: 'forbidden' }, 403) - return json(await hydration(session)) + return json(await hydration(session, durable?.runs)) } if (request.method === 'GET' && route === 'events') { @@ -724,9 +774,10 @@ export function createHarnessHandler( * What a `ChatClient` with `persistence` reads when it loads a thread (a * `ChatHydrationResult`): the transcript, the running chat turn (the client * joins it with `GET run?runId=`), and the interrupts that the last turn - * waits for. + * waits for. On a durable host, `runs` finds a turn that another instance + * runs. */ -async function hydration(session: HarnessSession) { +async function hydration(session: HarnessSession, runs?: RunStore) { const { pendingInterrupts, activeOperations } = session.snapshot() const [first] = pendingInterrupts const runId = first @@ -737,21 +788,26 @@ async function hydration(session: HarnessSession) { item.kind === 'chat' && session.operation(item.id)?.status() === 'running', ) + const remote = running + ? undefined + : await runs?.findActiveRun(session.threadId) + const runningId = + running?.id ?? (remote?.kind !== 'agent' ? remote?.runId : undefined) return { messages: modelMessagesToUIMessages(await session.transcript()), - activeRun: running ? { runId: running.id } : null, + activeRun: runningId ? { runId: runningId } : null, interrupts: runId ? { runId, pending: pendingInterrupts } : null, } } /** Session events of one operation, until it finishes. */ async function* followOperation( - events: AsyncIterable<{ operationId: string; event: StreamChunk }>, + events: AsyncIterable, operationId: string, ) { for await (const entry of events) { if (entry.operationId !== operationId) continue - yield entry.event + yield entry if ( entry.event.type === EventType.CUSTOM && entry.event.name === 'harness.operation.finished' diff --git a/packages/ai-harness/tests/run-route-hosts.test.ts b/packages/ai-harness/tests/run-route-hosts.test.ts new file mode 100644 index 0000000000..8bf86d4b59 --- /dev/null +++ b/packages/ai-harness/tests/run-route-hosts.test.ts @@ -0,0 +1,175 @@ +import { describe, expect, it } from 'vitest' +import { memoryLogStore, memoryPersistence } from '@tanstack/ai-persistence' +import { createHarnessHandler, createHarnessHost, defineHarness } from '../src' +import { gate, mockAdapter, text } from './helpers' +import type { Reply } from './helpers' + +/** + * Two server instances over the same stores, with no sticky routing: a run + * that starts on host A must be joinable and resumable from host B. + */ +function twoHosts() { + const { runs, metadata } = memoryPersistence().stores + const persistence = { stores: { log: memoryLogStore(), runs, metadata } } + const release = gate() + // RUN_STARTED, then the rest after `release`. By RUN_STARTED, host A holds + // the run lease, so host B only reads. + const reply: Reply = () => + (async function* () { + const [started, ...rest] = text('Hello from A.') + if (started) yield started + await release.opened + yield* rest + })() + const { adapter, calls } = mockAdapter([reply]) + const harness = defineHarness({ name: 'test/run-hosts', adapter }) + const handlerOf = () => { + const host = createHarnessHost({ persistence, coalesceMs: 0 }) + const handler = createHarnessHandler({ + host, + harness, + authorize: (request) => ({ + id: request.headers.get('x-user') ?? 'user-1', + }), + canAccess: (principal, threadId) => threadId.startsWith(principal.id), + mediaSecret: 'shared', + }) + const call = (path: string, init: RequestInit = {}) => + handler(new Request(`http://h.test/api/${path}`, init)) + return { host, call } + } + const a = handlerOf() + const b = handlerOf() + const post = (on: typeof a, headers: Record = {}) => + on.call('run', { + method: 'POST', + headers: { 'content-type': 'application/json', ...headers }, + body: JSON.stringify({ + threadId: 'user-1-t', + runId: 'run-a', + messages: [{ id: 'm-1', role: 'user', content: 'Hi.' }], + tools: [], + context: [], + }), + }) + const close = () => Promise.all([a.host.close(), b.host.close()]) + return { a, b, post, release, calls, close } +} + +/** SSE events with their `id:` lines. */ +const withIds = (body: string) => + body + .split('\n\n') + .filter((block) => block.includes('data: ')) + .map((block) => ({ + id: /^id: (.*)$/m.exec(block)?.[1], + chunk: JSON.parse(/^data: (.*)$/m.exec(block)?.[1] ?? 'null') as { + type: string + name?: string + delta?: string + }, + })) + +/** Read `response` until RUN_STARTED, then cancel it: a cut connection. */ +async function readThenCut(response: Response) { + const reader = response.body!.pipeThrough(new TextDecoderStream()).getReader() + let body = '' + while (!body.includes('RUN_STARTED') || !body.endsWith('\n\n')) { + const { value, done } = await reader.read() + if (done) break + body += value + } + await reader.cancel() + return withIds(body) +} + +describe('a run on a durable host, across instances', () => { + it('sends the log cursor as the id of each POST run event', async () => { + const { a, post, release, close } = twoHosts() + release.open() + const events = withIds(await (await post(a)).text()) + expect(events.length).toBeGreaterThan(3) + expect(events.every((event) => /^\d+$/.test(event.id ?? ''))).toBe(true) + await close() + }) + + it('joins a run that host A runs from host B, live, from the start', async () => { + const { a, b, post, release, close } = twoHosts() + const started = await readThenCut(await post(a)) + expect(started.at(-1)?.chunk.type).toBe('RUN_STARTED') + + const joined = await b.call('run?runId=run-a&offset=-1') + expect(joined.status).toBe(200) + release.open() + const events = withIds(await joined.text()) + // The same events, from the start of the turn, as host A sent. + expect(events.slice(0, started.length)).toEqual(started) + expect(events.some((event) => event.chunk.delta === 'Hello from A.')).toBe( + true, + ) + expect(events.at(-1)?.chunk.name).toBe('harness.operation.finished') + expect(events.every((event) => event.id !== undefined)).toBe(true) + await close() + }) + + it('resumes a cut join on host B after Last-Event-ID, with no repeated event', async () => { + const { a, b, post, release, close } = twoHosts() + release.open() + const all = withIds(await (await post(a)).text()) + const cut = all[1]?.id + if (cut === undefined) throw new Error('Too few events.') + + const resumed = await b.call('run?runId=run-a', { + headers: { 'Last-Event-ID': cut }, + }) + const got = withIds(await resumed.text()) + // Every event after the cut, once, to the end of the turn. + expect(got.slice(0, all.length - 2)).toEqual(all.slice(2)) + expect(got.at(-1)?.chunk.name).toBe('harness.operation.finished') + await close() + }) + + it('resumes a cut POST run on host B: the retry streams from Last-Event-ID and runs nothing again', async () => { + const { a, b, post, release, calls, close } = twoHosts() + const first = await readThenCut(await post(a)) + const cut = first.at(-1)?.id + if (cut === undefined) throw new Error('No cursor on the first event.') + + // The client re-sends the same POST with the last cursor it saw. + const retried = post(b, { 'Last-Event-ID': cut }) + release.open() + const rest = withIds(await (await retried).text()) + expect(rest[0]?.chunk.type).not.toBe('RUN_STARTED') + expect(rest.some((event) => event.chunk.delta === 'Hello from A.')).toBe( + true, + ) + expect(rest.at(-1)?.chunk.name).toBe('harness.operation.finished') + expect(rest.every((event) => Number(event.id) > Number(cut))).toBe(true) + expect(calls).toHaveLength(1) + await close() + }) + + it('shows the run as activeRun on host B, so a reloaded client joins it', async () => { + const { a, b, post, release, close } = twoHosts() + const posted = await post(a) + await readThenCut(posted) + const hydrated = await (await b.call('run?threadId=user-1-t')).json() + expect(hydrated.activeRun).toEqual({ runId: 'run-a' }) + release.open() + await (await b.call('run?runId=run-a&offset=-1')).text() + await close() + }) + + it('refuses a join on host B of a thread the principal may not use, and 404s an unknown run', async () => { + const { a, b, post, release, close } = twoHosts() + await readThenCut(await post(a)) + const refused = await b.call('run?runId=run-a', { + headers: { 'x-user': 'user-2' }, + }) + expect(refused.status).toBe(403) + expect((await b.call('run?runId=nope&offset=-1')).status).toBe(404) + release.open() + await (await b.call('run?runId=run-a&offset=-1')).text() + await close() + }) +}) From 0d59133e6d09a3fc43e66dd00b49ef1bdd57254c Mon Sep 17 00:00:00 2001 From: Jack Herrington Date: Wed, 14 Oct 2026 10:02:35 -0700 Subject: [PATCH 3/3] docs(ai-harness): mark the session view experimental, and say what a cursor is `createSessionView`, `SessionViewState`, and `SessionViewEvents` become public API on publish. Tag them `@experimental` so their shape can still change in a minor release. The HTTP transport page now says that event cursors are opaque strings: log positions on a durable host, process-local counters without a log store. So a deployment with more than one instance needs `stores.log`. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/harness-view-experimental.md | 5 +++++ docs/harness/connect.md | 9 +++++++++ packages/ai-harness/src/view/index.ts | 8 +++++++- packages/ai-harness/src/view/types.ts | 6 +++++- 4 files changed, 26 insertions(+), 2 deletions(-) create mode 100644 .changeset/harness-view-experimental.md diff --git a/.changeset/harness-view-experimental.md b/.changeset/harness-view-experimental.md new file mode 100644 index 0000000000..cdb14f7fa5 --- /dev/null +++ b/.changeset/harness-view-experimental.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-harness': patch +--- + +Mark `createSessionView`, `SessionViewState`, and `SessionViewEvents` as `@experimental`. Document that event cursors are opaque strings: log positions on a durable host, and process-local counters without a log store. diff --git a/docs/harness/connect.md b/docs/harness/connect.md index 3510c8ac3e..dfd4fe46ba 100644 --- a/docs/harness/connect.md +++ b/docs/harness/connect.md @@ -55,6 +55,15 @@ The handler answers these paths under your route: - `GET snapshot?threadId=`: the status, running operations, and waiting approvals. - `GET agent-transcript?threadId=&subagentRunId=`: the saved messages of one child agent of the thread. See [Show what a child wrote after a reload](./subagents#show-what-a-child-wrote-after-a-reload). +## Cursors + +Each event id of `GET events` and `GET run`, and the `cursor` of each `harness.event` frame, is an opaque string. Send it back as `Last-Event-ID` or `from`, as it is. Do not parse it. + +- On a durable host (`stores.log`), a cursor is a position in the log of the thread. It works on every instance and after a restart. +- Without a log store, a cursor is a counter in the memory of one process. Another instance or a restart does not know it. + +To run more than one instance behind one URL, use `stores.log`. + ## Choose what clients can change A client can send a prompt and answer questions. Other inputs can change how the session works, so a client can use only what the harness exposes. Nothing is exposed by default. diff --git a/packages/ai-harness/src/view/index.ts b/packages/ai-harness/src/view/index.ts index 81a6f737b4..91e9c03417 100644 --- a/packages/ai-harness/src/view/index.ts +++ b/packages/ai-harness/src/view/index.ts @@ -177,7 +177,11 @@ export interface SessionViewSource { type MediaBody = ArrayBuffer | Uint8Array type MediaInfo = { name: string; mimeType: string } -/** What `view.on(name, handler)` sends. */ +/** + * What `view.on(name, handler)` sends. + * + * @experimental The event names and values can change in a minor release. + */ export interface SessionViewEvents { approval: Approval clientTool: ClientToolCall @@ -341,6 +345,8 @@ function coalesce(task: () => Promise) { * view.on('approval', (approval) => approval.approve()) * await view.send('fix the failing test') * ``` + * + * @experimental The view API and its state can change in a minor release. */ export function createSessionView(source: SessionViewSource) { const internal = createStore(emptyState()) diff --git a/packages/ai-harness/src/view/types.ts b/packages/ai-harness/src/view/types.ts index 950ffeaa09..a3deda362a 100644 --- a/packages/ai-harness/src/view/types.ts +++ b/packages/ai-harness/src/view/types.ts @@ -180,7 +180,11 @@ export interface ViewAgent { cancel: () => Promise } -/** Everything a UI can show about a session. */ +/** + * Everything a UI can show about a session. + * + * @experimental The fields can change in a minor release. + */ export interface SessionViewState { threadId: string status: 'idle' | 'running' | 'requires_action'