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/.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/.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/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..dfd4fe46ba 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`. @@ -51,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. @@ -271,7 +284,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`: @@ -290,7 +303,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 @@ -302,7 +316,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..9f71baf959 100644 --- a/packages/ai-harness/src/http.ts +++ b/packages/ai-harness/src/http.ts @@ -6,11 +6,10 @@ import { modelMessagesToUIMessages, readInterruptBinding, resolveResumeRunId, - toServerSentEventsResponse, 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 { @@ -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 @@ -45,22 +44,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 } @@ -121,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 @@ -254,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 @@ -262,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. @@ -309,16 +344,21 @@ 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 + 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.', + ) + } 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)) @@ -372,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, ) } @@ -429,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) @@ -459,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, ) } @@ -476,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') { @@ -714,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 @@ -727,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' @@ -757,10 +823,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 +832,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/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' 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-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() + }) +}) 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',