diff --git a/packages/integration-tests/README.md b/packages/integration-tests/README.md index ffd86dbab..3d6cd04a3 100644 --- a/packages/integration-tests/README.md +++ b/packages/integration-tests/README.md @@ -1,66 +1,75 @@ # integration-tests -End-to-end tests that drive the real Workshop and a real gatekeeper over the actual RPC API, plus the -toolkit those tests are built from. Part of `pnpm test`, so CI runs it like any other package. +End-to-end tests that drive the Workshop and gatekeepers through the public Cap'n Web API. The package +also exports the small runtime bridge used by live agent evals. -```bash +```sh pnpm --filter @gadgets/integration-tests test:run ``` -## The toolkit - -Three source-only modules, consumed both by the tests here and by per-vendor suites in repos that -vendor this one as a submodule: - -- **`src/harness.ts`** — boots `workshop-backend` and any set of gatekeepers as real Workers under - [`wrangler`'s `createTestHarness()`](https://developers.cloudflare.com/changelog/post/2026-07-21-integration-test-harness/), - patching their checked-in `wrangler.jsonc` in memory. Parameterised over gatekeepers on purpose: a - suite for a new gatekeeper should be "point the harness at the package", not a forked copy. -- **`src/network-interceptor.ts`** — `NetworkInterceptor`, mechanism only. It patches - `globalThis.fetch` (the harness routes Worker subrequests back through the Node process, so that is - enough), passes loopback through, and **throws on anything a handler didn't match** — a test cannot - reach the real internet. What a given vendor's endpoints answer lives in a handler module you pass - in, which is what makes it reusable across gatekeepers. -- **`src/rpc-client.ts`** — speaks Cap'n Web over a WebSocket to `/api`, the same transport the - browser uses: sign-up, reading connected accounts, and `ObserverConfigRecorder`, which records the - overseer's `configure()` calls and answers from a scripted queue. - -## Writing a test here -- **No test may assume a clean slate.** Everything in a file shares one harness, `it.concurrent` runs - the cases together, and storage is never reset. Take fresh identities from `nextUsernames()` and use - per-test resource URLs; account labels are allocated for you, so two tests can't pick the same one. -- **The escape assertion lives in `afterAll`, not `afterEach`** — an `afterEach` fires while sibling - tests are still running, so it would inspect and clear state they are still using. - -## The fixture gatekeeper - -`fixtures/gatekeeper-test/` is a real Worker speaking the real gatekeeper protocol, whose verification -outcome the tests set over an HTTP control route. It exists because the overseer cases need a -gatekeeper that will refuse an observer *on command*, and every shipping one can do that only at a -cost that would dominate the test: - -- The OAuth ones need a whole vendor auth surface mocked before an account exists at all. -- The Context Library only refuses after an observation has been *recorded*, which takes a gadget read - session, a slash command, or an AI-chat catalog snapshot — and it is a singleton, so it cannot - produce two simultaneously failing bindings. - -Adding a test hook to those workers was considered and rejected: a "mark observed" hook would stub the -very state the tracker maintains, and an injected dev credential for an OAuth gatekeeper would bypass -exactly the flow that makes a real vendor worth testing. - -Two deliberate departures from a shipping gatekeeper, both to keep the fixture cheap: - -- No `capnweb-validate` build step; `main` points straight at source. `@validateRpc()` would require - the fixture to carry its own `wrangler types` output — half a megabyte of generated `.d.ts` for a - test double. The harness's handling of a generated `main` is covered anyway, by `workshop-backend`. -- One control knob, `allow`. A settled denial and an expired credential reach the overseer identically - — both as a thrown error, which it deliberately cannot tell apart because it treats every failure as - repairable — so the reason string is what carries the difference. Tests cover both narratives by - choosing reason text. - -## Further reading - -[`docs/integration-testing.md`](../../docs/integration-testing.md) covers the reasoning behind the -shape of all this: why fake timers cannot work here, why a fixture gatekeeper rather than a real one, -how storage isolation works, and the capnweb, wrangler, and workerd traps to expect. Read it before -changing the toolkit or starting a suite of your own. +`workshop-backend/__integration__` runs inside workerd and can use `cloudflare:test` internals. This +package runs from Node and reaches only the same public API as a real client. + +## Exported toolkit + +- `harness` boots the backend and selected gatekeepers as real Workers with + `wrangler`'s `createTestHarness()`. +- `rpc-client` opens Cap'n Web sessions, authenticates test users, and supplies observer callbacks. +- `agent-session` drives one production agent chat, waits for it to settle, reads complete history, + discovers workpieces, connects to Gadget RPCs, and exposes provider usage metadata. +- `network-interceptor` supplies explicit HTTP handlers to gatekeeper suites and rejects unexpected + external requests. + +The toolkit owns transport and lifecycle only. A consuming test or eval owns prompts, scores, and +behavioral assertions. + +## Local agent session + +Keep Gadget execution enabled and configure the Workshop with existing model credentials: + +```ts +const harness = await startHarness({ + enableGadgetExecution: true, + gatekeepers: [], + patchWorkshop(config) { + config.vars = { + ...config.vars, + CF_AI_GATEWAY: process.env.CF_AI_GATEWAY, + CF_AI_GATEWAY_ACCOUNT_ID: process.env.CF_AI_GATEWAY_ACCOUNT_ID, + CF_AI_GATEWAY_API_TOKEN: process.env.CF_AI_GATEWAY_API_TOKEN, + CF_AI_GATEWAY_PROVIDERS: "cloudflare", + }; + }, +}); + +using session = await AgentSession.create(harness.url, { + modelId: "@cf/zai-org/glm-5.2", +}); +const result = await session.run("Build a small status page."); +``` + +A caller can also add one directly configured user model through `userModel`. This lets local runs use +existing `CLOUDFLARE_ACCOUNT_ID` and `CLOUDFLARE_API_TOKEN` credentials without changing the backend. + +## Preview agent session + +A deployed preview uses Cloudflare Access instead of password signup. Supply an Access application +JWT. The client sends it on the WebSocket handshake and then calls `authenticateFromCfAccess()`. + +```ts +using session = await AgentSession.create(new URL(previewUrl), { + accessToken, + modelId: "@cf/zai-org/glm-5.2", +}); +``` + +The preview supplies its own model catalog and Workers AI binding. The same prompts and Gadget RPC +verifiers can therefore run against local workerd or a deployed preview. + +## Test isolation + +The local harness keeps storage for its full lifetime. Tests must create fresh identities and unique +resource URLs rather than assume a reset. Dispose every returned RPC stub and close the harness. + +[`docs/integration-testing.md`](../../docs/integration-testing.md) describes the in-process and +out-of-process boundaries in more detail. diff --git a/packages/integration-tests/__tests__/agent-session.test.ts b/packages/integration-tests/__tests__/agent-session.test.ts new file mode 100644 index 000000000..5985dd7ec --- /dev/null +++ b/packages/integration-tests/__tests__/agent-session.test.ts @@ -0,0 +1,131 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { AiChatAuthorInfo, AiChatHistoryPage, AiChatMessage, AiChatMetadata } + from "@gadgets/workshop-shared/api"; +import { + finalAssistantText, +} from "../src/agent-session.js"; +import { AgentTurnCompletion, loadAllChatHistory } from "../src/agent-session-internals.js"; + +function metadata(active: boolean): AiChatMetadata { + return { + id: 7, + title: "test", + started: new Date(0), + lastActive: new Date(0), + ...(active ? { activeAgent: { type: "agent", id: "model", name: "Model" } } : {}), + }; +} + +function message(sequence: number): AiChatMessage { + return { + chatId: 7, + sequence, + timestamp: new Date(sequence), + author: { type: "user", id: "user", name: "User" }, + type: "message", + message: String(sequence), + }; +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("AgentTurnCompletion", () => { + it("settles after active becomes idle for the debounce period", async () => { + vi.useFakeTimers(); + const completion = new AgentTurnCompletion(7, () => Promise.resolve(), 1_000, 25); + completion.metadata(metadata(true)); + completion.metadata(metadata(false)); + + await vi.advanceTimersByTimeAsync(24); + expect(completion.settled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await expect(completion.promise).resolves.toBeUndefined(); + }); + + it("retains the final chat usage metadata", async () => { + vi.useFakeTimers(); + const completion = new AgentTurnCompletion(7, () => Promise.resolve(), 1_000, 25); + completion.metadata(metadata(true)); + completion.metadata({ ...metadata(false), totalTokens: 123, totalCost: 0.45 }); + await vi.advanceTimersByTimeAsync(25); + await completion.promise; + + expect(completion.lastMetadata).toMatchObject({ totalTokens: 123, totalCost: 0.45 }); + }); + + it("resets idle settlement when a callback restarts the agent", async () => { + vi.useFakeTimers(); + const completion = new AgentTurnCompletion(7, () => Promise.resolve(), 1_000, 25); + completion.metadata(metadata(true)); + completion.metadata(metadata(false)); + await vi.advanceTimersByTimeAsync(20); + completion.metadata(metadata(true)); + await vi.advanceTimersByTimeAsync(20); + expect(completion.settled).toBe(false); + + completion.metadata(metadata(false)); + await vi.advanceTimersByTimeAsync(25); + await expect(completion.promise).resolves.toBeUndefined(); + }); + + it("stops the agent on timeout", async () => { + vi.useFakeTimers(); + let stops = 0; + const completion = new AgentTurnCompletion( + 7, () => { stops++; return Promise.resolve(); }, 100, 25); + await vi.advanceTimersByTimeAsync(100); + + await expect(completion.promise).rejects.toThrow("Timed out after 100ms"); + expect(stops).toBe(1); + }); + + it("stops the agent when cancelled", async () => { + let stops = 0; + const controller = new AbortController(); + const completion = new AgentTurnCompletion( + 7, () => { stops++; return Promise.resolve(); }, 1_000, 25, controller.signal); + controller.abort(); + + await expect(completion.promise).rejects.toThrow("Agent turn was cancelled"); + expect(stops).toBe(1); + }); + + it("clears timers and abort listeners when disposed", () => { + vi.useFakeTimers(); + const controller = new AbortController(); + const completion = new AgentTurnCompletion( + 7, () => Promise.resolve(), 1_000, 25, controller.signal); + completion.dispose(); + controller.abort(); + vi.runAllTimers(); + expect(completion.settled).toBe(false); + }); +}); + +it("loads history pages in authoritative ascending order", async () => { + const pages = new Map([ + [undefined, { messages: [message(4), message(5)], compacted: { to: 4, summary: "tail" } }], + [4, { messages: [message(2), message(3)], compacted: { to: 2, summary: "middle" } }], + [2, { messages: [message(0), message(1)] }], + ]); + const history = await loadAllChatHistory(before => { + const page = pages.get(before); + if (!page) throw new Error(`No page for ${before}`); + return Promise.resolve(page); + }); + expect(history.map(entry => entry.sequence)).toEqual([0, 1, 2, 3, 4, 5]); +}); + +it("returns the final non-empty assistant message from canonical history", () => { + const agent: AiChatAuthorInfo = { type: "agent", id: "model", name: "Model" }; + const history = [ + message(0), + { ...message(1), author: agent, message: "first" }, + { ...message(2), author: agent, message: "" }, + ]; + + expect(finalAssistantText(history)).toBe("first"); +}); + diff --git a/packages/integration-tests/__tests__/rpc-client.test.ts b/packages/integration-tests/__tests__/rpc-client.test.ts new file mode 100644 index 000000000..02e27b034 --- /dev/null +++ b/packages/integration-tests/__tests__/rpc-client.test.ts @@ -0,0 +1,42 @@ +import { createServer, type IncomingMessage, type Server } from "node:http"; +import { once } from "node:events"; +import { afterEach, expect, it } from "vitest"; +import { WebSocketServer } from "ws"; +import { connect } from "../src/rpc-client.js"; + +const servers = new Set<{ http: Server; ws: WebSocketServer }>(); + +afterEach(async () => { + await Promise.all([...servers].map(async ({ http, ws }) => { + ws.close(); + http.close(); + await Promise.allSettled([once(ws, "close"), once(http, "close")]); + })); + servers.clear(); +}); + +it("sends the Access application token and same-origin header on preview WebSockets", async () => { + const http = createServer(); + const ws = new WebSocketServer({ server: http }); + servers.add({ http, ws }); + http.listen(0, "127.0.0.1"); + await once(http, "listening"); + + const address = http.address(); + if (address === null || typeof address === "string") throw new Error("Test server has no TCP port"); + const baseUrl = new URL(`http://127.0.0.1:${address.port}/preview`); + const requestPromise = new Promise(resolve => { + ws.once("connection", (socket, request) => { + resolve(request); + socket.close(); + }); + }); + + { + using _api = connect(baseUrl, { accessToken: "access.jwt.value" }); + const request = await requestPromise; + expect(request.url).toBe("/api"); + expect(request.headers.origin).toBe(baseUrl.origin); + expect(request.headers.cookie).toBe("CF_Authorization=access.jwt.value"); + } +}); diff --git a/packages/integration-tests/package.json b/packages/integration-tests/package.json index 6c4ef5ec0..9cd1a0fba 100644 --- a/packages/integration-tests/package.json +++ b/packages/integration-tests/package.json @@ -4,6 +4,7 @@ "private": true, "type": "module", "exports": { + "./agent-session": "./src/agent-session.ts", "./harness": "./src/harness.ts", "./network-interceptor": "./src/network-interceptor.ts", "./rpc-client": "./src/rpc-client.ts" @@ -17,11 +18,13 @@ "@gadgets/workshop-shared": "workspace:*", "capnweb": "catalog:", "jsonc-parser": "^3.3.1", + "ws": "^8.21.0", "zod": "^4.4.3" }, "devDependencies": { "@cloudflare/workers-types": "^5.20260808.1", "@types/node": "^26.1.0", + "@types/ws": "^8.18.1", "typescript": "catalog:", "vitest": "catalog:", "wrangler": "catalog:" diff --git a/packages/integration-tests/src/agent-session-internals.ts b/packages/integration-tests/src/agent-session-internals.ts new file mode 100644 index 000000000..8291bb3be --- /dev/null +++ b/packages/integration-tests/src/agent-session-internals.ts @@ -0,0 +1,142 @@ +import type { AiChatHistoryPage, AiChatMessage, AiChatMetadata } from "@gadgets/workshop-shared/api"; + + +export class AgentTurnCompletion { + readonly promise: Promise; + lastMetadata: AiChatMetadata | undefined; + #chatId: number | null; + #resolve: () => void = () => {}; + #reject: (error: Error) => void = () => {}; + #stopAgent: () => Promise; + #settleDebounceMs: number; + #hardTimeout: ReturnType | undefined; + #idleTimer: ReturnType | undefined; + #signal: AbortSignal | undefined; + #sawActive = false; + #isSettled = false; + #stopRequested = false; + #pendingMetadata: AiChatMetadata[] = []; + + constructor( + chatId: number | null, + stopAgent: () => Promise, + timeoutMs: number, + settleDebounceMs: number, + signal?: AbortSignal) { + this.#chatId = chatId; + this.#stopAgent = stopAgent; + this.#settleDebounceMs = settleDebounceMs; + this.#signal = signal; + this.promise = new Promise((resolve, reject) => { + this.#resolve = resolve; + this.#reject = error => reject(error); + }); + // Mark the promise handled while the RPC that reveals a new chat ID is still in flight. + this.promise.catch(() => {}); + this.#hardTimeout = setTimeout(() => { + this.#stopRequested = true; + this.#requestStop(); + this.#fail(new Error(`Timed out after ${timeoutMs}ms waiting for the agent turn`)); + }, timeoutMs); + signal?.addEventListener("abort", this.#onAbort, { once: true }); + if (signal?.aborted) this.#onAbort(); + } + + get settled(): boolean { + return this.#isSettled; + } + + attach(chatId: number): void { + if (this.#chatId !== null && this.#chatId !== chatId) { + throw new Error(`Agent turn is already attached to chat ${this.#chatId}`); + } + this.#chatId = chatId; + const metadata = this.#pendingMetadata; + this.#pendingMetadata = []; + for (const entry of metadata) this.metadata(entry); + if (this.#stopRequested) this.#requestStop(); + } + + metadata(chat: AiChatMetadata): void { + if (this.#chatId === null) { + this.#pendingMetadata.push(chat); + return; + } + if (chat.id !== this.#chatId) return; + this.lastMetadata = chat; + if (this.#isSettled) { + if (chat.activeAgent && this.#stopRequested) this.#requestStop(); + return; + } + if (chat.activeAgent) { + this.#sawActive = true; + this.#clearIdleTimer(); + } else if (this.#sawActive) { + this.#clearIdleTimer(); + this.#idleTimer = setTimeout(() => this.#succeed(), this.#settleDebounceMs); + } + } + + cancel(): void { + if (this.#isSettled) return; + this.#stopRequested = true; + this.#requestStop(); + this.#fail(new Error("Agent turn was cancelled")); + } + + dispose(): void { + this.#clearTimers(); + this.#signal?.removeEventListener("abort", this.#onAbort); + } + + #onAbort = (): void => { + this.cancel(); + }; + + #requestStop(): void { + if (this.#chatId !== null) this.#stopAgent().catch(() => {}); + } + + #succeed(): void { + if (this.#isSettled) return; + this.#isSettled = true; + this.dispose(); + this.#resolve(); + } + + #fail(error: Error): void { + if (this.#isSettled) return; + this.#isSettled = true; + this.dispose(); + this.#reject(error); + } + + #clearIdleTimer(): void { + if (this.#idleTimer !== undefined) clearTimeout(this.#idleTimer); + this.#idleTimer = undefined; + } + + #clearTimers(): void { + this.#clearIdleTimer(); + if (this.#hardTimeout !== undefined) clearTimeout(this.#hardTimeout); + this.#hardTimeout = undefined; + } +} + +export async function loadAllChatHistory( + loadPage: (beforeSequence?: number) => Promise): Promise { + let page = await loadPage(); + let messages = page.messages; + const boundaries = new Set(); + while (page.compacted) { + const boundary = page.compacted.to; + if (boundaries.has(boundary)) { + throw new Error(`Chat history repeated compaction boundary ${boundary}`); + } + boundaries.add(boundary); + page = await loadPage(boundary); + messages = [...page.messages, ...messages]; + } + return messages; +} + diff --git a/packages/integration-tests/src/agent-session.ts b/packages/integration-tests/src/agent-session.ts new file mode 100644 index 000000000..c86dd9bfa --- /dev/null +++ b/packages/integration-tests/src/agent-session.ts @@ -0,0 +1,350 @@ +import type { RpcCompatible, RpcStub } from "capnweb"; +import type { + AiChatMessage, AiChatMetadata, AiChatStreamEvent, AiChatSubscriber, AiChatAuthorInfo, AiModelConfig, + AuthenticatedApi, GadgetClient, OutputFormatOffer, Overseer, PublicApi, WorkpieceId, + WorkpieceSummary, WorkpiecesSubscriber, +} from "@gadgets/workshop-shared/api"; +import type { CodeChange } from "@gadgets/workshop-shared/code-change"; +import { AgentTurnCompletion, loadAllChatHistory } from "./agent-session-internals.js"; +import { RpcTarget, connect, nextUsernames, signUp, stubFor, waitFor } from "./rpc-client.js"; + +const DEFAULT_TIMEOUT_MS = 120_000; +const DEFAULT_SETTLE_DEBOUNCE_MS = 500; + +function connectTyped>( + gadget: RpcStub, chatId?: number): Promise>; +function connectTyped(gadget: RpcStub, chatId?: number) { + return gadget.connectToGadget(chatId); +} + +/** Options for creating an isolated production Workshop agent session. */ +export type AgentSessionOptions = { + /** Model to use. It must appear in the new workspace's `listModels()` result. Defaults to the first. */ + modelId?: string; + /** Access application JWT. When present, authenticate as its Access identity instead of signing up. */ + accessToken?: string; + /** Optional model to add to the fresh local account before its workspace opens. */ + userModel?: { profile: AiChatAuthorInfo; config: AiModelConfig }; + /** Alphanumeric prefix for the fresh account name. */ + usernamePrefix?: string; + /** Hard limit for each agent turn. Defaults to two minutes. */ + timeoutMs?: number; +}; + +/** Options for one prompt in an agent session. */ +export type AgentTurnOptions = { + /** Merge all proposed changes, including the current live draft, after the agent settles. */ + acceptChanges?: boolean; + /** Cancels the turn by calling `stopAgent()` and rejecting the run. */ + signal?: AbortSignal; +}; + +/** The branch a verifier should connect to. */ +export type AgentGadgetBranch = "accepted" | "chat"; + + +/** Authoritative state returned after one agent turn settles. */ +export type AgentTurnResult = { + /** Chat created by the first turn and reused by later turns. */ + chatId: number; + /** Complete canonical chat history in ascending sequence order. */ + history: AiChatMessage[]; + /** Workpieces known when the turn finished. */ + workpieces: WorkpieceSummary[]; + /** Agent error messages posted during the turn. Empty when the turn completed normally. */ + agentErrors: string[]; + /** Provider usage available from chat metadata after this turn. */ + usage: { totalTokens?: number; costUsd?: number }; +}; + +/** Return the final user-visible assistant text from canonical chat history. */ +export function finalAssistantText(history: readonly AiChatMessage[]): string { + for (let index = history.length - 1; index >= 0; index--) { + const entry = history[index]; + if (entry?.type === "message" && entry.author.type === "agent" && entry.message !== "") { + return entry.message; + } + } + return ""; +} + +class ChatSubscriber extends RpcTarget implements AiChatSubscriber { + completion: AgentTurnCompletion | undefined; + + streamGeneration(_generation: number): void {} + metadata(chat: AiChatMetadata): void { this.completion?.metadata(chat); } + deleted(_chatId: number): void {} + message(_entry: AiChatMessage): void {} + changeApplied( + _chatId: number, _generation: number, _revision: number, _author: AiChatAuthorInfo, + _change: CodeChange, _submission?: {clientId: string; seq: number}): void {} + stream(_chatId: number, _event: AiChatStreamEvent): void {} +} + +class WorkpieceSubscriber extends RpcTarget implements WorkpiecesSubscriber { + readonly entries = new Map(); + readonly readyPromise: Promise; + #resolveReady: () => void = () => {}; + + constructor() { + super(); + this.readyPromise = new Promise(resolve => { this.#resolveReady = resolve; }); + } + + entry(summary: WorkpieceSummary): void { this.entries.set(summary.id, summary); } + removed(id: WorkpieceId): void { this.entries.delete(id); } + ready(): void { this.#resolveReady(); } +} + + +/** + * Drives the production Workshop RPC lifecycle for one fresh user and workspace. + * + * The class deliberately does not interpret agent output. Callers own verification and evaluation. + * Dispose the session when finished; verifier stubs returned by this class remain caller-owned. + */ +export class AgentSession implements Disposable { + /** Model selected from the workspace's `listModels()` result. */ + readonly modelId: string; + /** ID of the fresh workspace owned by this session's fresh user. */ + readonly workspaceId: string; + #publicApi: RpcStub; + #authenticatedApi: RpcStub; + #overseer: RpcStub; + #chatSubscriber = new ChatSubscriber(); + #chatSubscriberStub: RpcStub | undefined; + #chatSubscription: RpcStub<{}> | undefined; + #workpieceSubscriber = new WorkpieceSubscriber(); + #workpieceSubscriberStub: RpcStub | undefined; + #workpieceSubscription: RpcStub<{}> | undefined; + #chatId: number | undefined; + #turn: AgentTurnCompletion | undefined; + #timeoutMs: number; + #settleDebounceMs: number; + #disposed = false; + #failed = false; + + private constructor( + publicApi: RpcStub, + authenticatedApi: RpcStub, + overseer: RpcStub, + workspaceId: string, + modelId: string, + timeoutMs: number) { + this.#publicApi = publicApi; + this.#authenticatedApi = authenticatedApi; + this.#overseer = overseer; + this.workspaceId = workspaceId; + this.modelId = modelId; + this.#timeoutMs = timeoutMs; + this.#settleDebounceMs = DEFAULT_SETTLE_DEBOUNCE_MS; + } + + /** Create a fresh account and workspace, then establish subscriptions before any chat starts. */ + static async create(baseUrl: URL, options: AgentSessionOptions = {}): Promise { + const publicApi = connect(baseUrl, { accessToken: options.accessToken }); + let authenticatedApi: RpcStub | undefined; + let overseer: RpcStub | undefined; + let session: AgentSession | undefined; + try { + if (options.accessToken === undefined) { + const username = nextUsernames(options.usernamePrefix ?? "agent").at(0); + if (username === undefined) throw new Error("Failed to allocate an integration-test username"); + authenticatedApi = await signUp(publicApi, username); + } else { + authenticatedApi = await publicApi.authenticateFromCfAccess(); + } + if (options.userModel !== undefined) { + await authenticatedApi.addModel(options.userModel.profile, options.userModel.config); + } + overseer = await authenticatedApi.newGadget(); + const [metadata, models] = await Promise.all([ + overseer.getMetadata(), + overseer.listModels(), + ]); + const modelId = AgentSession.#selectModel(models, options.modelId); + session = new AgentSession( + publicApi, authenticatedApi, overseer, metadata.id, modelId, + options.timeoutMs ?? DEFAULT_TIMEOUT_MS); + await session.#initializeSubscriptions(); + return session; + } catch (error) { + if (session === undefined) { + overseer?.[Symbol.dispose](); + authenticatedApi?.[Symbol.dispose](); + publicApi[Symbol.dispose](); + } else { + session[Symbol.dispose](); + } + throw error; + } + } + + /** + * Send a prompt. The first call creates a chat; later calls continue that same chat. + * History is fetched through every compaction page after the turn settles. + */ + async run(prompt: string, options: AgentTurnOptions = {}): Promise { + this.#assertUsable(); + if (this.#turn !== undefined) throw new Error("An agent turn is already running"); + + const existingChatId = this.#chatId ?? null; + const completion = new AgentTurnCompletion( + existingChatId, + () => this.#stopCurrentAgent(), + this.#timeoutMs, + this.#settleDebounceMs, + options.signal); + this.#turn = completion; + this.#chatSubscriber.completion = completion; + try { + if (this.#chatId === undefined) { + this.#chatId = await this.#overseer.newChat(prompt, this.modelId); + completion.attach(this.#chatId); + } else { + await this.#overseer.sendChatMessage(this.#chatId, prompt, this.modelId); + } + await completion.promise; + + let history = await this.#loadHistory(this.#chatId); + if (options.acceptChanges) { + await this.acceptChanges(); + history = await this.#loadHistory(this.#chatId); + } + return { + chatId: this.#chatId, + history, + workpieces: this.workpieces(), + agentErrors: history.flatMap(entry => entry.type === "error" ? [entry.message] : []), + usage: { + ...(completion.lastMetadata?.totalTokens === undefined + ? {} : { totalTokens: completion.lastMetadata.totalTokens }), + ...(completion.lastMetadata?.totalCost === undefined + ? {} : { costUsd: completion.lastMetadata.totalCost }), + }, + }; + } catch (error) { + this.#failed = true; + throw error; + } finally { + completion.dispose(); + if (this.#chatSubscriber.completion === completion) { + this.#chatSubscriber.completion = undefined; + } + if (this.#turn === completion) this.#turn = undefined; + } + } + + /** Stop and reject the active turn. Does nothing while idle. */ + cancel(): void { + this.#turn?.cancel(); + } + + /** Current workpieces discovered through `subscribeToWorkpieces()`. */ + workpieces(): WorkpieceSummary[] { + return [...this.#workpieceSubscriber.entries.values()]; + } + + /** Obtain a caller-owned verifier capability for one gadget workpiece. */ + getGadget(id: WorkpieceId): Promise> { + this.#assertUsable(); + return this.#overseer.getGadget(id); + } + + /** + * Connect a caller-owned, typed verifier stub to accepted code or this session's chat branch. + */ + async connectToGadget>( + id: WorkpieceId, branch: AgentGadgetBranch = "chat"): Promise> { + this.#assertUsable(); + using gadget = await this.#overseer.getGadget(id); + if (branch === "chat" && this.#chatId === undefined) { + throw new Error("The session has no chat branch yet"); + } + return connectTyped(gadget, branch === "chat" ? this.#chatId : undefined); + } + + /** + * Wait until the deployment's standard output formats are installed. + * The first API request starts format installation without waiting for it. Wait here so the first + * agent prompt always sees the installed formats. + */ + async waitForOutputFormats(): Promise { + this.#assertUsable(); + return waitFor( + "the output formats to install (is workshop-backend built? see its README)", async () => { + const offers = await this.#authenticatedApi.listOutputFormats(); + return offers.length > 0 ? offers : null; + }); + } + + /** Accept every change proposed by the current chat. */ + async acceptChanges(): Promise { + this.#assertUsable(); + if (this.#turn !== undefined) throw new Error("Cannot accept changes while an agent turn is running"); + if (this.#chatId === undefined) throw new Error("The session has no chat branch to accept"); + const result = await this.#overseer.mergeChanges(this.#chatId); + if (result.outcome !== "merged") { + throw new Error("The chat became stale before its changes could be accepted"); + } + } + + /** Delete the isolated workspace and all data created by the session. */ + async deleteWorkspace(): Promise { + this.#assertUsable(); + if (this.#turn !== undefined) throw new Error("Cannot delete the workspace during an agent turn"); + await this.#overseer.deleteSelf(); + } + + /** Dispose subscriptions, callback targets, RPC capabilities, and the WebSocket session. */ + [Symbol.dispose](): void { + if (this.#disposed) return; + this.#disposed = true; + this.#turn?.cancel(); + this.#turn?.dispose(); + this.#chatSubscription?.[Symbol.dispose](); + this.#workpieceSubscription?.[Symbol.dispose](); + this.#chatSubscriberStub?.[Symbol.dispose](); + this.#workpieceSubscriberStub?.[Symbol.dispose](); + this.#overseer[Symbol.dispose](); + this.#authenticatedApi[Symbol.dispose](); + this.#publicApi[Symbol.dispose](); + } + + async #initializeSubscriptions(): Promise { + this.#chatSubscriberStub = stubFor(this.#chatSubscriber); + this.#chatSubscription = await this.#overseer.subscribeToChat(this.#chatSubscriberStub); + this.#workpieceSubscriberStub = stubFor(this.#workpieceSubscriber); + this.#workpieceSubscription = await this.#overseer.subscribeToWorkpieces( + this.#workpieceSubscriberStub); + await this.#workpieceSubscriber.readyPromise; + } + + async #loadHistory(chatId: number): Promise { + return loadAllChatHistory(before => this.#overseer.getChatHistory(chatId, before)); + } + + + #stopCurrentAgent(): Promise { + if (this.#chatId === undefined) return Promise.resolve(); + return this.#overseer.stopAgent(this.#chatId); + } + + #assertUsable(): void { + if (this.#disposed) throw new Error("AgentSession is disposed"); + if (this.#failed) throw new Error("AgentSession cannot be reused after a failed or cancelled turn"); + } + + static #selectModel(models: AiChatAuthorInfo[], requested: string | undefined): string { + if (models.length === 0) throw new Error("The Workshop exposes no configured agent models"); + if (requested === undefined) { + const first = models.at(0); + if (first === undefined) throw new Error("The Workshop exposes no configured agent models"); + return first.id; + } + if (!models.some(model => model.id === requested)) { + throw new Error(`Model "${requested}" is not exposed by this workspace`); + } + return requested; + } +} diff --git a/packages/integration-tests/src/harness.ts b/packages/integration-tests/src/harness.ts index 199facdeb..9cc0654ad 100644 --- a/packages/integration-tests/src/harness.ts +++ b/packages/integration-tests/src/harness.ts @@ -81,6 +81,7 @@ function readWorkerConfig(dir: string): WorkerConfig { function workshopConfig( gatekeepers: { binding: string; name: string }[], + enableGadgetExecution: boolean, patch?: (config: WorkerConfig) => void): WorkerConfig { const config = readWorkerConfig(WORKSHOP_DIR); @@ -96,9 +97,9 @@ function workshopConfig( // No CF_ACCESS_AUD, so /api takes the unauthenticated path and password signup is available. config.vars = { ...config.vars, ADMINS: [ADMIN_USERNAME] }; - // Gadget code is never executed here (a gatekeeper is in observer scope purely by having a - // vendorId), so drop the Worker Loader rather than requiring it to start. - delete config.worker_loaders; + // Most integration tests do not execute Gadget code, so avoid requiring the Worker Loader unless + // the caller is driving a production agent session that can use executeCode. + if (!enableGadgetExecution) delete config.worker_loaders; patch?.(config); return config; @@ -122,6 +123,8 @@ export type Harness = { }; export async function startHarness(opts: { + /** Retain the Workshop's checked-in Worker Loader so agents and Gadgets can execute code. */ + enableGadgetExecution?: boolean; gatekeepers: GatekeeperSpec[]; patchWorkshop?: (config: WorkerConfig) => void; /** Defaults to this repo's root. Override when a gatekeeper lives outside it. */ @@ -139,7 +142,8 @@ export async function startHarness(opts: { root: opts.root ?? REPO_ROOT, // workshop-backend is primary, so unrouted requests (e.g. /api) go to it. workers: [ - { config: workshopConfig(gatekeepers, opts.patchWorkshop) }, + { config: workshopConfig( + gatekeepers, opts.enableGadgetExecution ?? false, opts.patchWorkshop) }, ...gatekeepers.map(({ config }) => ({ config })), ], }); diff --git a/packages/integration-tests/src/rpc-client.ts b/packages/integration-tests/src/rpc-client.ts index 83ec8da96..13fd45a1b 100644 --- a/packages/integration-tests/src/rpc-client.ts +++ b/packages/integration-tests/src/rpc-client.ts @@ -2,6 +2,7 @@ import { createHash } from "node:crypto"; import { RpcStub, RpcTarget, newWebSocketRpcSession } from "capnweb"; +import NodeWebSocket from "ws"; import type { AuthenticatedApi, ConnectedAccountsSubscriber, ObserverAccountChoice, ObserverBindingNeed, ObserverConfigCallback, PublicApi, @@ -10,6 +11,9 @@ import type { AccountDescription, SupportedResource, VendorDescription, } from "@gadgets/workshop-shared/gatekeeper"; +/** Canonical callback-target base paired with this package's Cap'n Web instance. */ +export { RpcTarget }; + /** * Poll `attempt` until it returns non-null. * @@ -43,11 +47,27 @@ export function nextUsernames(...prefixes: string[]): string[] { return prefixes.map(prefix => `${prefix}${n}`); } +/** Options for opening the Workshop's Cap'n Web session. */ +export type WorkshopConnectionOptions = { + /** Access application JWT, sent as the preview's authorization cookie. */ + accessToken?: string; +}; + /** Open an RPC session against the Workshop's /api endpoint. */ -export function connect(baseUrl: URL): RpcStub { +export function connect( + baseUrl: URL, options: WorkshopConnectionOptions = {}): RpcStub { const wsUrl = new URL("/api", baseUrl); wsUrl.protocol = wsUrl.protocol === "https:" ? "wss:" : "ws:"; - return newWebSocketRpcSession(wsUrl.toString()); + if (options.accessToken === undefined) { + return newWebSocketRpcSession(wsUrl.toString()); + } + const nodeSocket = new NodeWebSocket(wsUrl.toString(), { + origin: baseUrl.origin, + headers: { Cookie: `CF_Authorization=${options.accessToken}` }, + }); + // `ws` implements the standard client protocol Cap'n Web consumes, but its Node declarations add + // binary modes and omit Workers-only server methods, so the otherwise-compatible types diverge. + return newWebSocketRpcSession(nodeSocket as unknown as WebSocket); } /** diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 325914f63..eaef363fb 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -696,6 +696,9 @@ importers: jsonc-parser: specifier: ^3.3.1 version: 3.3.1 + ws: + specifier: ^8.21.0 + version: 8.21.3 zod: specifier: ^4.4.3 version: 4.4.3 @@ -706,6 +709,9 @@ importers: '@types/node': specifier: 26.1.0 version: 26.1.0 + '@types/ws': + specifier: ^8.18.1 + version: 8.18.1 typescript: specifier: 'catalog:' version: 7.0.2 @@ -2650,6 +2656,9 @@ packages: '@types/unist@3.0.3': resolution: {integrity: sha512-ko/gIFJRv177XgZsZcBwnqJN5x/Gien8qNOn0D5bQU/zAzVf9Zt3BlcUiLqhV9y4ARk0GbT3tnUiPNgnTXzc/Q==} + '@types/ws@8.18.1': + resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==} + '@types/yauzl@2.10.3': resolution: {integrity: sha512-oJoftv0LSuaDZE3Le4DbKX+KS9G36NzOeSap90UIK0yMA/NhKJhqlSGtNDORNRaIbQfzjXDrQa0ytJ6mNRGz/Q==} @@ -6718,6 +6727,10 @@ snapshots: '@types/unist@3.0.3': {} + '@types/ws@8.18.1': + dependencies: + '@types/node': 26.1.0 + '@types/yauzl@2.10.3': dependencies: '@types/node': 26.1.0