From 2496cc14ed31cc6918dbd8f54ae8ec1fc7a17da7 Mon Sep 17 00:00:00 2001 From: Ashish Kumar Singh Date: Wed, 19 Aug 2026 14:20:43 -0500 Subject: [PATCH 1/7] Add an agent session driver and Gadget durability tests AgentSession drives one agent session over the same Cap'n Web API the browser uses: a fresh account and workspace, one chat across turns, complete paginated history, and an optional source snapshot. Two methods support tests that need a known implementation rather than whatever an agent produced. seedGadget() writes hand-authored source into the workspace. restartGadgets() restarts every Gadget server by applying an empty code update, which is what the platform does on every code change. gadget-durability.test.ts uses both to pin platform behaviour with no model involved. Storage survives a restart and memory does not, outstanding stubs become invalid, and the data holds across five restarts and across one that interrupts a write. It also shows that a check-then-write implementation oversells under concurrent calls. startHarness() gains enableGadgetExecution, which keeps the Worker Loader so Gadget code can run. It defaults to false, so the existing suites are unchanged. --- packages/integration-tests/README.md | 44 +- .../__tests__/agent-session.test.ts | 143 ++++++ .../__tests__/gadget-durability.test.ts | 143 ++++++ .../fixtures/seeded-gadgets.ts | 92 ++++ packages/integration-tests/package.json | 2 + .../src/agent-session-internals.ts | 169 +++++++ .../integration-tests/src/agent-session.ts | 449 ++++++++++++++++++ packages/integration-tests/src/harness.ts | 12 +- packages/integration-tests/src/rpc-client.ts | 3 + pnpm-lock.yaml | 3 + 10 files changed, 1055 insertions(+), 5 deletions(-) create mode 100644 packages/integration-tests/__tests__/agent-session.test.ts create mode 100644 packages/integration-tests/__tests__/gadget-durability.test.ts create mode 100644 packages/integration-tests/fixtures/seeded-gadgets.ts create mode 100644 packages/integration-tests/src/agent-session-internals.ts create mode 100644 packages/integration-tests/src/agent-session.ts diff --git a/packages/integration-tests/README.md b/packages/integration-tests/README.md index ffd86dbab..f6f8f5f1c 100644 --- a/packages/integration-tests/README.md +++ b/packages/integration-tests/README.md @@ -7,11 +7,19 @@ toolkit those tests are built from. Part of `pnpm test`, so CI runs it like any pnpm --filter @gadgets/integration-tests test:run ``` +`workshop-backend/__integration__` also covers more than one module. It runs in-process under +`@cloudflare/vitest-pool-workers`, so it reaches Durable Object internals through `cloudflare:test`. +This package runs out-of-process and reaches only the public Cap'n Web API. + ## The toolkit -Three source-only modules, consumed both by the tests here and by per-vendor suites in repos that +Four source-only entry points, consumed both by the tests here and by per-vendor suites in repos that vendor this one as a submodule: +- **`src/agent-session.ts`** — `AgentSession`, a production-RPC driver for live agent evaluations. + It creates a fresh account and workspace, keeps one chat across turns, waits through callback-driven + agent restarts, reads complete paginated history, discovers workpieces, and optionally accepts and + snapshots source. It owns transport lifecycle only; evaluation semantics stay in the consuming suite. - **`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 @@ -32,6 +40,40 @@ vendor this one as a submodule: - **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. +## Live agent evaluations + +Boot the harness with `enableGadgetExecution: true`; the default is false so existing suites do not +need a Worker Loader. Configure the Workshop through `patchWorkshop` with the real deployment model +credentials, then create the driver from the harness URL: + +```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: "anthropic,openai,google,cloudflare", + }; + }, +}); + +using session = await AgentSession.create(harness.url, { modelId: "claude-sonnet-5" }); +const first = await session.run("Build a small status page."); +const accepted = await session.run("Add an incident timeline.", { acceptChanges: true }); +``` + +`run()` never accepts changes by default. When acceptance is requested it includes the current live +draft, then returns a Yjs V2 source snapshot keyed by each workpiece's `filesRoot`. Verifiers can use +`getGadget()` or typed `connectToGadget()`, selecting either the accepted or current chat branch. + +The unit suite uses no fake model protocol. A credentialed live evaluation is the end-to-end proof +for model execution and must be invoked explicitly by its consuming repository; it is not registered +as a silently skipped test here. + ## The fixture gatekeeper `fixtures/gatekeeper-test/` is a real Worker speaking the real gatekeeper protocol, whose verification 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..16415094f --- /dev/null +++ b/packages/integration-tests/__tests__/agent-session.test.ts @@ -0,0 +1,143 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { Doc, Text } from "yjs"; +import type { + AiChatAuthorInfo, AiChatHistoryPage, AiChatMessage, AiChatMetadata, WorkpieceSummary, +} from "@gadgets/workshop-shared/api"; +import { + finalAssistantText, +} from "../src/agent-session.js"; +import { + AgentTurnCompletion, buildSourceSnapshot, 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("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"); +}); + +it("maps Yjs V2 source by each workpiece filesRoot", () => { + const doc = new Doc(); + const legacy = doc.getMap(""); + const modern = doc.getMap("12"); + legacy.set("client.js", new Text("legacy client")); + modern.set("server.js", new Text("modern server")); + const workpieces: WorkpieceSummary[] = [ + { id: 1, type: "gadget", title: "Legacy", filesRoot: "" }, + { id: 12, type: "gadget", title: "Modern", filesRoot: "12" }, + { id: 13, type: "gadget", title: "Empty" }, + ]; + + const snapshot = buildSourceSnapshot(doc, 9, workpieces); + expect(snapshot.version).toBe(9); + expect(snapshot.workpieces.get("")?.summary.id).toBe(1); + expect(snapshot.workpieces.get("")?.files.get("client.js")).toBe("legacy client"); + expect(snapshot.workpieces.get("12")?.files.get("server.js")).toBe("modern server"); + expect(snapshot.workpieces.size).toBe(2); +}); diff --git a/packages/integration-tests/__tests__/gadget-durability.test.ts b/packages/integration-tests/__tests__/gadget-durability.test.ts new file mode 100644 index 000000000..b7c73be11 --- /dev/null +++ b/packages/integration-tests/__tests__/gadget-durability.test.ts @@ -0,0 +1,143 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { AgentSession } from "../src/agent-session.js"; +import { startHarness, type Harness, type WorkerConfig } from "../src/harness.js"; +import { DURABLE_NOTES_SERVER, OVERSELLING_DESK_SERVER } from "../fixtures/seeded-gadgets.js"; + +// Scope: a Gadget facet restarting, and whether an application's state survives it. A whole Durable +// Object resetting is a separate concern, covered in `workshop-backend/__integration__`, which runs +// in-process and can abort one object through `cloudflare:test`. + +// Seeded gadgets never call a model, but AgentSession.create() selects one from listModels(), which +// is empty unless the Workshop believes a gateway is configured. These values are never dialled. +function offlineModelConfig(config: WorkerConfig): void { + config.vars = { + ...config.vars, + CF_AI_GATEWAY: "unused", + CF_AI_GATEWAY_ACCOUNT_ID: "unused", + CF_AI_GATEWAY_API_TOKEN: "unused", + CF_AI_GATEWAY_PROVIDERS: "cloudflare", + }; +} + +interface NotesApi { + put(input: { id: string; body: string }): Promise<{ ok: boolean }>; + all(): Promise<{ id: string; body: string }[]>; + liveness(): Promise<{ instance: string; callsSinceStart: number }>; +} + +interface DeskApi { + defineSlot(input: { slotId: string; capacity: number }): Promise<{ ok: boolean }>; + book(input: { bookingId: string; slotId: string }): + Promise<{ ok: true } | { ok: false; error: string }>; + bookedCount(input: { slotId: string }): Promise; +} + +let harness: Harness; + +beforeAll(async () => { + harness = await startHarness({ + gatekeepers: [], + enableGadgetExecution: true, + patchWorkshop: offlineModelConfig, + }); +}, 120_000); + +afterAll(async () => { + await harness?.server.close(); +}); + +describe("Gadget durability across abrupt server restarts", () => { + it("restarts the server for real, keeping storage and dropping memory", async () => { + using session = await AgentSession.create(harness.url, { usernamePrefix: "durable" }); + const id = await session.seedGadget({ + title: "Notes", bindingName: "NOTES", files: { "server.js": DURABLE_NOTES_SERVER }, + }); + + const before = await session.connectToGadget(id, "accepted"); + await before.put({ id: "a", body: "first" }); + const livenessBefore = await before.liveness(); + expect(livenessBefore.callsSinceStart).toBeGreaterThan(0); + + await session.restartGadgets(); + + // The abort invalidates outstanding stubs, which is what proves it actually happened. Without + // this the rest of the suite could pass against a restart that never occurred. + await expect(before.all()).rejects.toThrow(); + before[Symbol.dispose](); + + using after = await session.connectToGadget(id, "accepted"); + // Read the counter before anything else, since every other method bumps it. + const livenessAfter = await after.liveness(); + expect(livenessAfter.instance).not.toBe(livenessBefore.instance); + expect(livenessAfter.callsSinceStart).toBe(0); + expect(await after.all()).toEqual([{ id: "a", body: "first" }]); + }, 120_000); + + it("keeps every row across repeated restarts", async () => { + using session = await AgentSession.create(harness.url, { usernamePrefix: "durable" }); + const id = await session.seedGadget({ + title: "Notes", bindingName: "NOTES", files: { "server.js": DURABLE_NOTES_SERVER }, + }); + + const instances: string[] = []; + for (let round = 0; round < 5; round++) { + using api = await session.connectToGadget(id, "accepted"); + await api.put({ id: `note-${round}`, body: `body-${round}` }); + expect(await api.all()).toHaveLength(round + 1); + instances.push((await api.liveness()).instance); + await session.restartGadgets(); + } + + using api = await session.connectToGadget(id, "accepted"); + expect(await api.all()).toEqual( + Array.from({ length: 5 }, (_unused, round) => ({ id: `note-${round}`, body: `body-${round}` }))); + // Every round really did run in a fresh instance. + expect(new Set(instances).size).toBe(5); + }, 180_000); + + it("leaves storage consistent when a restart lands on an unfinished write", async () => { + using session = await AgentSession.create(harness.url, { usernamePrefix: "durable" }); + const id = await session.seedGadget({ + title: "Notes", bindingName: "NOTES", files: { "server.js": DURABLE_NOTES_SERVER }, + }); + + const writer = await session.connectToGadget(id, "accepted"); + await writer.put({ id: "settled", body: "already committed" }); + // Dispatched but deliberately not awaited, so the restart races it. + const inFlight = writer.put({ id: "racing", body: "may or may not land" }).catch(() => undefined); + await session.restartGadgets(); + await inFlight; + writer[Symbol.dispose](); + + using api = await session.connectToGadget(id, "accepted"); + const rows = await api.all(); + // Either outcome is correct for the racing write; what must hold is that the committed row is + // intact, the table is readable, and no half-written row appeared. + expect(rows).toContainEqual({ id: "settled", body: "already committed" }); + expect(rows.length).toBeGreaterThanOrEqual(1); + expect(rows.length).toBeLessThanOrEqual(2); + for (const row of rows) expect(typeof row.body).toBe("string"); + }, 120_000); +}); + +describe("Concurrent RPC interleaving", () => { + // Pins the platform behaviour that any "never oversells" assertion depends on: if a naive + // implementation could not oversell here, such an assertion would hold for the wrong reason. + it("lets a check-then-write implementation oversell", async () => { + using session = await AgentSession.create(harness.url, { usernamePrefix: "oversell" }); + const id = await session.seedGadget({ + title: "Broken Desk", bindingName: "DESK", files: { "server.js": OVERSELLING_DESK_SERVER }, + }); + + using api = await session.connectToGadget(id, "accepted"); + const capacity = 3; + await api.defineSlot({ slotId: "race", capacity }); + + const results = await Promise.all(Array.from({ length: 10 }, (_unused, index) => + api.book({ bookingId: `race-${index}`, slotId: "race" }))); + const accepted = results.filter(result => result.ok).length; + + expect(accepted).toBeGreaterThan(capacity); + expect(await api.bookedCount({ slotId: "race" })).toBeGreaterThan(capacity); + }, 120_000); +}); diff --git a/packages/integration-tests/fixtures/seeded-gadgets.ts b/packages/integration-tests/fixtures/seeded-gadgets.ts new file mode 100644 index 000000000..e108100ab --- /dev/null +++ b/packages/integration-tests/fixtures/seeded-gadgets.ts @@ -0,0 +1,92 @@ +/** + * Hand-written Gadget sources for tests that need a known implementation rather than whatever an + * agent produced. Kept as strings because that is exactly what `AgentSession.seedGadget()` writes + * into the workspace's Yjs document. + */ + +/** + * Stores notes in SQL and deliberately also keeps state in memory, so a test can tell a restart + * apart from a no-op: `rows` must survive one and `instance`/`callsSinceStart` must not. + */ +export const DURABLE_NOTES_SERVER = ` +import { DurableObject } from "cloudflare:workers"; + +export class Gadget extends DurableObject { + constructor(ctx, env) { + super(ctx, env); + this.ctx = ctx; + // Memory, which a restart must discard. + this.instance = crypto.randomUUID(); + this.callsSinceStart = 0; + this.ctx.storage.sql.exec( + "CREATE TABLE IF NOT EXISTS notes (id TEXT PRIMARY KEY, body TEXT NOT NULL)"); + } + + async put(input) { + this.callsSinceStart++; + this.ctx.storage.sql.exec( + "INSERT INTO notes (id, body) VALUES (?, ?) " + + "ON CONFLICT(id) DO UPDATE SET body = excluded.body", + input.id, input.body); + return { ok: true }; + } + + async all() { + this.callsSinceStart++; + return [...this.ctx.storage.sql.exec("SELECT id, body FROM notes ORDER BY id")] + .map(row => ({ id: row.id, body: row.body })); + } + + async liveness() { + return { instance: this.instance, callsSinceStart: this.callsSinceStart }; + } +} +`; + +/** + * A booking gadget written the obvious wrong way: it reads the count, checks capacity, then awaits + * before inserting. Exists to prove that concurrent RPC calls really do interleave here, which is + * the premise the eval suite's overselling check depends on. + */ +export const OVERSELLING_DESK_SERVER = ` +import { DurableObject } from "cloudflare:workers"; + +export class Gadget extends DurableObject { + constructor(ctx, env) { + super(ctx, env); + this.ctx = ctx; + this.ctx.storage.sql.exec( + "CREATE TABLE IF NOT EXISTS slots (slotId TEXT PRIMARY KEY, capacity INTEGER NOT NULL)"); + this.ctx.storage.sql.exec( + "CREATE TABLE IF NOT EXISTS bookings (bookingId TEXT PRIMARY KEY, slotId TEXT NOT NULL)"); + } + + async defineSlot(input) { + this.ctx.storage.sql.exec( + "INSERT OR REPLACE INTO slots (slotId, capacity) VALUES (?, ?)", input.slotId, input.capacity); + return { ok: true }; + } + + async book(input) { + const slot = [...this.ctx.storage.sql.exec( + "SELECT capacity FROM slots WHERE slotId = ?", input.slotId)].at(0); + if (slot === undefined) return { ok: false, error: "UNKNOWN_SLOT" }; + const booked = [...this.ctx.storage.sql.exec( + "SELECT COUNT(*) AS n FROM bookings WHERE slotId = ?", input.slotId)].at(0).n; + if (booked >= slot.capacity) return { ok: false, error: "SLOT_FULL" }; + + // The bug under test: yielding between the check and the write lets every other queued call + // pass the same capacity check before any of them has inserted. + await scheduler.wait(1); + + this.ctx.storage.sql.exec( + "INSERT INTO bookings (bookingId, slotId) VALUES (?, ?)", input.bookingId, input.slotId); + return { ok: true }; + } + + async bookedCount(input) { + return [...this.ctx.storage.sql.exec( + "SELECT COUNT(*) AS n FROM bookings WHERE slotId = ?", input.slotId)].at(0).n; + } +} +`; diff --git a/packages/integration-tests/package.json b/packages/integration-tests/package.json index 6c4ef5ec0..21613fcdd 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,6 +18,7 @@ "@gadgets/workshop-shared": "workspace:*", "capnweb": "catalog:", "jsonc-parser": "^3.3.1", + "yjs": "^13.6.31", "zod": "^4.4.3" }, "devDependencies": { 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..2246ad88e --- /dev/null +++ b/packages/integration-tests/src/agent-session-internals.ts @@ -0,0 +1,169 @@ +import type { + AiChatHistoryPage, AiChatMessage, AiChatMetadata, WorkpieceSummary, +} from "@gadgets/workshop-shared/api"; +import * as Y from "yjs"; + +export type SourceWorkpiece = { + /** Workpiece whose `filesRoot` keys this entry. */ + summary: WorkpieceSummary; + /** UTF-8 source text keyed by file name. */ + files: ReadonlyMap; +}; + +export type SourceSnapshot = { + /** Workshop code version included in the snapshot. */ + version: number; + /** Source-bearing workpieces keyed by `WorkpieceSummary.filesRoot`. */ + workpieces: ReadonlyMap; +}; + +export class AgentTurnCompletion { + readonly promise: Promise; + #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; + 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; +} + +export function buildSourceSnapshot( + doc: Y.Doc, version: number, summaries: readonly WorkpieceSummary[]): SourceSnapshot { + const workpieces = new Map(); + for (const summary of summaries) { + if (summary.filesRoot === undefined) continue; + const files = new Map(); + doc.getMap(summary.filesRoot).forEach((text, name) => { + files.set(name, text.toString()); + }); + workpieces.set(summary.filesRoot, { summary, files }); + } + return { version, workpieces }; +} diff --git a/packages/integration-tests/src/agent-session.ts b/packages/integration-tests/src/agent-session.ts new file mode 100644 index 000000000..cdd4d7856 --- /dev/null +++ b/packages/integration-tests/src/agent-session.ts @@ -0,0 +1,449 @@ +import type { RpcCompatible, RpcStub } from "capnweb"; +import type { + AiChatMessage, AiChatMetadata, AiChatStreamEvent, AiChatSubscriber, AiChatAuthorInfo, + AuthenticatedApi, CodeSubscriber, CodeUpdate, GadgetClient, OutputFormatOffer, Overseer, PublicApi, + WorkpieceId, WorkpieceSummary, WorkpiecesSubscriber, +} from "@gadgets/workshop-shared/api"; +import * as Y from "yjs"; +import { + AgentTurnCompletion, buildSourceSnapshot, loadAllChatHistory, +} from "./agent-session-internals.js"; +import type { SourceSnapshot } 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; + /** 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"; + +/** Sources accepted from a turn, keyed by each `WorkpieceSummary.filesRoot`. */ +export type AgentSourceSnapshot = SourceSnapshot; + +/** 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[]; + /** Present only when `acceptChanges` was requested. */ + source?: AgentSourceSnapshot; +}; + +/** 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 {} + draftUpdate( + _chatId: number, _timestamp: Date, _author: AiChatAuthorInfo, _update: Uint8Array): void {} + draftCleared(_chatId: 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(); } +} + +class SourceSubscriber extends RpcTarget implements CodeSubscriber { + readonly readyPromise: Promise; + version = 0; + #doc: Y.Doc; + #resolveReady: () => void = () => {}; + + constructor(doc: Y.Doc) { + super(); + this.#doc = doc; + this.readyPromise = new Promise(resolve => { this.#resolveReady = resolve; }); + } + + update(update: CodeUpdate): void { + Y.applyUpdateV2(this.#doc, update.update); + this.version = update.version; + } + + 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); + let authenticatedApi: RpcStub | undefined; + let overseer: RpcStub | undefined; + let session: AgentSession | undefined; + try { + 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); + 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); + let source: AgentSourceSnapshot | undefined; + if (options.acceptChanges) { + source = await this.#acceptChanges(this.#chatId, history); + history = await this.#loadHistory(this.#chatId); + } + return { + chatId: this.#chatId, + history, + workpieces: this.workpieces(), + agentErrors: history.flatMap(entry => entry.type === "error" ? [entry.message] : []), + ...(source === undefined ? {} : { source }), + }; + } 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. + * + * Installation is fire-and-forget on the first `/api` request — the same request that opens this + * session — so a turn started immediately can race it and get a system prompt with no formats + * section, silently changing what the agent is told it can instantiate. + */ + 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; + }); + } + + /** + * Create a gadget with hand-written source, bypassing the agent entirely. + * + * For tests that need a known implementation — a deliberately broken one to prove an assertion + * can fail, or a deliberately correct one to measure the platform rather than the model. The + * gadget is permanent rather than provisional to a chat, so connect to it on the accepted branch. + * + * `files` must contain `server.js` exporting a Durable Object class named `Gadget`; only `.js` + * entries become worker modules. + */ + async seedGadget(spec: { + title: string; + bindingName: string; + files: Record; + }): Promise { + this.#assertUsable(); + using gadget = await this.#overseer.createGadget(spec.title, undefined, spec.bindingName); + const id = await gadget.getId(); + // The server owns the files root, so read it back rather than re-deriving the naming rule. + const filesRoot = await waitFor(`workpiece ${id} to publish its files root`, async () => + this.#workpieceSubscriber.entries.get(id)?.filesRoot ?? null); + + const doc = new Y.Doc(); + const subscriber = new SourceSubscriber(doc); + using subscriberStub = stubFor(subscriber); + let subscription: RpcStub<{}> | undefined; + const updates: Uint8Array[] = []; + const collect = (update: Uint8Array) => { updates.push(update); }; + try { + // Sync the doc before writing. A write from an unsynced doc is a concurrent Y.Map set that + // resolves by client ID, i.e. a coin flip; a write from a synced one carries a causal delete + // of the existing entry and deterministically wins. + subscription = await this.#overseer.subscribeToCode(subscriberStub); + await subscriber.readyPromise; + doc.on("updateV2", collect); + doc.transact(() => { + const root = doc.getMap(filesRoot); + for (const [name, content] of Object.entries(spec.files)) { + const text = new Y.Text(); + text.insert(0, content); + root.set(name, text); + } + }); + doc.off("updateV2", collect); + if (updates.length === 0) throw new Error("Seeding a gadget produced no code update"); + // updateCode bumps the workspace code version, which both aborts the facet and changes the + // Worker Loader cache key, so the next connect loads exactly what was just written. + await this.#overseer.updateCode(Y.mergeUpdatesV2(updates)); + return id; + } finally { + doc.off("updateV2", collect); + subscription?.[Symbol.dispose](); + doc.destroy(); + } + } + + /** + * Abruptly restart every Gadget server in this workspace, as the platform itself does whenever + * code changes. Storage is preserved; in-memory state is discarded. + * + * Existing gadget stubs are invalidated and throw on their next call, so callers must reconnect — + * which is also how a caller proves the restart really happened rather than silently no-opping. + * + * Implemented as an empty update to the mainline code, which advances the workspace's code + * version. That is what forces the restart, and it is also why this must not be called *between* + * turns of a multi-turn conversation: the agent stamps the code version it observed onto its + * history and rejects an inconsistent one on replay. Call it within the final turn, or in a + * single-turn task. + */ + async restartGadgets(): Promise { + this.#assertUsable(); + if (this.#turn !== undefined) { + throw new Error("Cannot restart gadgets while an agent turn is running"); + } + await this.#overseer.updateCode(Y.encodeStateAsUpdateV2(new Y.Doc())); + } + + /** Merge the current provisional chat branch and return its accepted source snapshot. */ + 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 history = await this.#loadHistory(this.#chatId); + return this.#acceptChanges(this.#chatId, history); + } + + async #acceptChanges(chatId: number, history: readonly AiChatMessage[]): Promise { + const mergeThrough = history.at(-1)?.sequence ?? null; + await this.#overseer.mergeChanges(chatId, mergeThrough, { includeDraft: true }); + return this.#readAcceptedSource(); + } + + /** 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)); + } + + async #readAcceptedSource(): Promise { + const doc = new Y.Doc(); + const subscriber = new SourceSubscriber(doc); + using subscriberStub = stubFor(subscriber); + let subscription: RpcStub<{}> | undefined; + try { + subscription = await this.#overseer.subscribeToCode(subscriberStub); + await subscriber.readyPromise; + return buildSourceSnapshot(doc, subscriber.version, this.workpieces()); + } finally { + subscription?.[Symbol.dispose](); + doc.destroy(); + } + } + + #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..baa194960 100644 --- a/packages/integration-tests/src/rpc-client.ts +++ b/packages/integration-tests/src/rpc-client.ts @@ -10,6 +10,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. * diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 325914f63..83e7c484f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -696,6 +696,9 @@ importers: jsonc-parser: specifier: ^3.3.1 version: 3.3.1 + yjs: + specifier: ^13.6.31 + version: 13.6.31 zod: specifier: ^4.4.3 version: 4.4.3 From 809eb2a73271520ca20356f02df0190fa630f173 Mon Sep 17 00:00:00 2001 From: Ashish Kumar Singh Date: Wed, 19 Aug 2026 15:55:52 -0500 Subject: [PATCH 2/7] Let the interceptor pass a named host through A handler receives the URL, the method, and the headers, but never the body, so it cannot stand in for a host a suite has to reach with a real POST. passThroughHosts exempts such a host before the request is taken apart. Every other host still throws. --- packages/integration-tests/README.md | 3 +- .../__tests__/network-interceptor.test.ts | 29 +++++++++++++++++++ .../src/network-interceptor.ts | 24 +++++++++++++-- 3 files changed, 52 insertions(+), 4 deletions(-) diff --git a/packages/integration-tests/README.md b/packages/integration-tests/README.md index f6f8f5f1c..0b2c5e9c9 100644 --- a/packages/integration-tests/README.md +++ b/packages/integration-tests/README.md @@ -28,7 +28,8 @@ vendor this one as a submodule: `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. + in, which is what makes it reusable across gatekeepers. `passThroughHosts` exempts a host a suite + genuinely has to reach; a handler cannot, because it never receives the request body. - **`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. diff --git a/packages/integration-tests/__tests__/network-interceptor.test.ts b/packages/integration-tests/__tests__/network-interceptor.test.ts index c37ee431e..cf400cb42 100644 --- a/packages/integration-tests/__tests__/network-interceptor.test.ts +++ b/packages/integration-tests/__tests__/network-interceptor.test.ts @@ -121,3 +121,32 @@ it("uninstall restores the fetch that install captured", async () => { await expect(fetch("https://later.test/")).rejects.toThrow(/Unmocked outbound request/); interceptor.uninstall(); }); + +it("passes an allowed host through to the real fetch, body intact", async () => { + const seen: { url: string; method: string; body: string }[] = []; + const captured = globalThis.fetch; + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const request = new Request(input, init); + seen.push({ url: request.url, method: request.method, body: await request.text() }); + return Response.json({ ok: true }); + }) as typeof globalThis.fetch; + + const interceptor = new NetworkInterceptor([], { passThroughHosts: ["allowed.test"] }); + interceptor.install(); + try { + const response = await fetch("https://allowed.test/v1/chat", { + method: "POST", + body: JSON.stringify({ prompt: "hello" }), + }); + expect(await response.json()).toEqual({ ok: true }); + // A handler never receives a body, so passing one through has to bypass handlers entirely. + expect(seen).toEqual([ + { url: "https://allowed.test/v1/chat", method: "POST", body: '{"prompt":"hello"}' }, + ]); + + await expect(fetch("https://denied.test/probe")).rejects.toThrow(/Unmocked outbound request/); + } finally { + interceptor.uninstall(); + globalThis.fetch = captured; + } +}); diff --git a/packages/integration-tests/src/network-interceptor.ts b/packages/integration-tests/src/network-interceptor.ts index 506408229..18066a18a 100644 --- a/packages/integration-tests/src/network-interceptor.ts +++ b/packages/integration-tests/src/network-interceptor.ts @@ -18,13 +18,29 @@ export type Handler = (url: URL, method: string, headers: Headers) => Response | null | Promise; +/** Optional settings for {@link NetworkInterceptor}. */ +export type InterceptorOptions = { + /** + * Hostnames whose requests reach the real network untouched. + * + * A handler cannot stand in for a host that a suite genuinely has to reach, because a handler + * receives the URL, the method, and the headers, but never the body. Passing a real POST through + * therefore has to happen here, before the request is taken apart. + * + * Keep the list to hosts the suite cannot do without. Every other host still throws. + */ + passThroughHosts?: readonly string[]; +}; + export class NetworkInterceptor { readonly #handlers: readonly Handler[]; + readonly #passThroughHosts: ReadonlySet; #realFetch: typeof globalThis.fetch | null = null; #unmockedCalls: string[] = []; - constructor(handlers: Handler[] = []) { + constructor(handlers: Handler[] = [], options: InterceptorOptions = {}) { this.#handlers = [...handlers]; + this.#passThroughHosts = new Set(options.passThroughHosts ?? []); } install(): void { @@ -40,8 +56,10 @@ export class NetworkInterceptor { : input.url; const url = new URL(raw); - // The harness dispatches its own traffic over loopback; let that through untouched. - if (url.hostname === "localhost" || url.hostname === "127.0.0.1" || url.hostname === "[::1]") { + // The harness dispatches its own traffic over loopback; let that through untouched. An + // explicitly allowed host goes the same way, before the body can be disturbed below. + if (url.hostname === "localhost" || url.hostname === "127.0.0.1" || url.hostname === "[::1]" + || this.#passThroughHosts.has(url.hostname)) { return realFetch(input, init); } From dce021ac3d3001dbb0420d4ea8b764297e54f4f6 Mon Sep 17 00:00:00 2001 From: Ashish Kumar Singh Date: Wed, 19 Aug 2026 16:03:30 -0500 Subject: [PATCH 3/7] Pin that a Gadget's client.js must parse for its server to start Every .js file in a Gadget becomes a module in its Worker, so workerd parses client.js at load even though the server never imports it. A test that checks a Gadget through its RPC therefore already covers the syntax of both files, and a separate parse step would add nothing. --- .../__tests__/gadget-durability.test.ts | 38 ++++++++++++++++++- .../fixtures/seeded-gadgets.ts | 9 +++++ 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/packages/integration-tests/__tests__/gadget-durability.test.ts b/packages/integration-tests/__tests__/gadget-durability.test.ts index b7c73be11..6729d4612 100644 --- a/packages/integration-tests/__tests__/gadget-durability.test.ts +++ b/packages/integration-tests/__tests__/gadget-durability.test.ts @@ -1,7 +1,9 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { AgentSession } from "../src/agent-session.js"; import { startHarness, type Harness, type WorkerConfig } from "../src/harness.js"; -import { DURABLE_NOTES_SERVER, OVERSELLING_DESK_SERVER } from "../fixtures/seeded-gadgets.js"; +import { + DURABLE_NOTES_SERVER, OVERSELLING_DESK_SERVER, VALID_SERVER_ONLY, +} from "../fixtures/seeded-gadgets.js"; // Scope: a Gadget facet restarting, and whether an application's state survives it. A whole Durable // Object resetting is a separate concern, covered in `workshop-backend/__integration__`, which runs @@ -25,6 +27,10 @@ interface NotesApi { liveness(): Promise<{ instance: string; callsSinceStart: number }>; } +interface PingApi { + ping(): Promise; +} + interface DeskApi { defineSlot(input: { slotId: string; capacity: number }): Promise<{ ok: boolean }>; book(input: { bookingId: string; slotId: string }): @@ -120,6 +126,36 @@ describe("Gadget durability across abrupt server restarts", () => { }, 120_000); }); +describe("Gadget module loading", () => { + // Every `.js` file in a Gadget becomes a module in its Worker, so workerd parses client.js at load + // even though the server never imports it. A test that verifies a Gadget through its RPC therefore + // already covers the syntax of both files, and needs no separate parse step. + it("refuses to start the server when client.js cannot parse", async () => { + using session = await AgentSession.create(harness.url, { usernamePrefix: "syntax" }); + const id = await session.seedGadget({ + title: "Unparseable", + bindingName: "UNPARSEABLE", + files: { "server.js": VALID_SERVER_ONLY, "client.js": "this is ( not valid javascript ===" }, + }); + + // Cap'n Web pipelines, so connecting resolves and the load failure surfaces on the first call. + using api = await session.connectToGadget(id, "accepted"); + await expect(api.ping()).rejects.toThrow(/SyntaxError/); + }, 120_000); + + it("starts the server when both files parse", async () => { + using session = await AgentSession.create(harness.url, { usernamePrefix: "syntax" }); + const id = await session.seedGadget({ + title: "Parseable", + bindingName: "PARSEABLE", + files: { "server.js": VALID_SERVER_ONLY, "client.js": "const unused = 1;" }, + }); + + using api = await session.connectToGadget(id, "accepted"); + expect(await api.ping()).toBe("ok"); + }, 120_000); +}); + describe("Concurrent RPC interleaving", () => { // Pins the platform behaviour that any "never oversells" assertion depends on: if a naive // implementation could not oversell here, such an assertion would hold for the wrong reason. diff --git a/packages/integration-tests/fixtures/seeded-gadgets.ts b/packages/integration-tests/fixtures/seeded-gadgets.ts index e108100ab..c06013d2e 100644 --- a/packages/integration-tests/fixtures/seeded-gadgets.ts +++ b/packages/integration-tests/fixtures/seeded-gadgets.ts @@ -90,3 +90,12 @@ export class Gadget extends DurableObject { } } `; + +/** A server that loads and answers, paired with a `client.js` that cannot parse. */ +export const VALID_SERVER_ONLY = ` +import { DurableObject } from "cloudflare:workers"; + +export class Gadget extends DurableObject { + async ping() { return "ok"; } +} +`; From ef19c88541857f6a608fabf86fd3842709ec4066 Mon Sep 17 00:00:00 2001 From: Ashish Kumar Singh Date: Mon, 24 Aug 2026 11:56:48 -0500 Subject: [PATCH 4/7] Connect agent sessions to Access previews --- .../__tests__/rpc-client.test.ts | 42 +++++++++++++++++++ packages/integration-tests/package.json | 2 + .../integration-tests/src/agent-session.ts | 21 +++++++--- packages/integration-tests/src/rpc-client.ts | 21 +++++++++- pnpm-lock.yaml | 13 ++++++ 5 files changed, 92 insertions(+), 7 deletions(-) create mode 100644 packages/integration-tests/__tests__/rpc-client.test.ts 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 21613fcdd..49a8417d0 100644 --- a/packages/integration-tests/package.json +++ b/packages/integration-tests/package.json @@ -19,11 +19,13 @@ "capnweb": "catalog:", "jsonc-parser": "^3.3.1", "yjs": "^13.6.31", + "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.ts b/packages/integration-tests/src/agent-session.ts index cdd4d7856..2c6859ba4 100644 --- a/packages/integration-tests/src/agent-session.ts +++ b/packages/integration-tests/src/agent-session.ts @@ -1,6 +1,6 @@ import type { RpcCompatible, RpcStub } from "capnweb"; import type { - AiChatMessage, AiChatMetadata, AiChatStreamEvent, AiChatSubscriber, AiChatAuthorInfo, + AiChatMessage, AiChatMetadata, AiChatStreamEvent, AiChatSubscriber, AiChatAuthorInfo, AiModelConfig, AuthenticatedApi, CodeSubscriber, CodeUpdate, GadgetClient, OutputFormatOffer, Overseer, PublicApi, WorkpieceId, WorkpieceSummary, WorkpiecesSubscriber, } from "@gadgets/workshop-shared/api"; @@ -24,6 +24,10 @@ function connectTyped(gadget: RpcStub, chatId?: number) { 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. */ @@ -162,14 +166,21 @@ export class AgentSession implements Disposable { /** 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); + const publicApi = connect(baseUrl, { accessToken: options.accessToken }); let authenticatedApi: RpcStub | undefined; let overseer: RpcStub | undefined; let session: AgentSession | undefined; try { - 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); + 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(), diff --git a/packages/integration-tests/src/rpc-client.ts b/packages/integration-tests/src/rpc-client.ts index baa194960..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, @@ -46,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 83e7c484f..0d22696ba 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 yjs: specifier: ^13.6.31 version: 13.6.31 @@ -709,6 +712,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 @@ -2653,6 +2659,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==} @@ -6721,6 +6730,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 From e969de600df1566c553b8499900acc11c48b0739 Mon Sep 17 00:00:00 2001 From: Ashish Kumar Singh Date: Mon, 24 Aug 2026 12:15:43 -0500 Subject: [PATCH 5/7] Expose agent session usage to eval harnesses --- packages/integration-tests/README.md | 14 ++++++++++++-- .../__tests__/agent-session.test.ts | 11 +++++++++++ .../src/agent-session-internals.ts | 2 ++ packages/integration-tests/src/agent-session.ts | 8 ++++++++ 4 files changed, 33 insertions(+), 2 deletions(-) diff --git a/packages/integration-tests/README.md b/packages/integration-tests/README.md index 0b2c5e9c9..a331f3b8f 100644 --- a/packages/integration-tests/README.md +++ b/packages/integration-tests/README.md @@ -31,8 +31,8 @@ vendor this one as a submodule: in, which is what makes it reusable across gatekeepers. `passThroughHosts` exempts a host a suite genuinely has to reach; a handler cannot, because it never receives the request body. - **`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. + browser uses. Local sessions use in-band password authentication. Preview sessions can attach a + Cloudflare Access application JWT to the WebSocket handshake. ## Writing a test here - **No test may assume a clean slate.** Everything in a file shares one harness, `it.concurrent` runs @@ -67,6 +67,16 @@ const first = await session.run("Build a small status page."); const accepted = await session.run("Add an incident timeline.", { acceptChanges: true }); ``` +The same driver can use a deployed Access preview. The supplied JWT authenticates the WebSocket and +the session calls `authenticateFromCfAccess()` instead of creating a password account: + +```ts +using session = await AgentSession.create(new URL(previewUrl), { + accessToken, + modelId: "@cf/zai-org/glm-5.2", +}); +``` + `run()` never accepts changes by default. When acceptance is requested it includes the current live draft, then returns a Yjs V2 source snapshot keyed by each workpiece's `filesRoot`. Verifiers can use `getGadget()` or typed `connectToGadget()`, selecting either the accepted or current chat branch. diff --git a/packages/integration-tests/__tests__/agent-session.test.ts b/packages/integration-tests/__tests__/agent-session.test.ts index 16415094f..50641cefa 100644 --- a/packages/integration-tests/__tests__/agent-session.test.ts +++ b/packages/integration-tests/__tests__/agent-session.test.ts @@ -48,6 +48,17 @@ describe("AgentTurnCompletion", () => { 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); diff --git a/packages/integration-tests/src/agent-session-internals.ts b/packages/integration-tests/src/agent-session-internals.ts index 2246ad88e..f9102b656 100644 --- a/packages/integration-tests/src/agent-session-internals.ts +++ b/packages/integration-tests/src/agent-session-internals.ts @@ -19,6 +19,7 @@ export type SourceSnapshot = { export class AgentTurnCompletion { readonly promise: Promise; + lastMetadata: AiChatMetadata | undefined; #chatId: number | null; #resolve: () => void = () => {}; #reject: (error: Error) => void = () => {}; @@ -78,6 +79,7 @@ export class AgentTurnCompletion { return; } if (chat.id !== this.#chatId) return; + this.lastMetadata = chat; if (this.#isSettled) { if (chat.activeAgent && this.#stopRequested) this.#requestStop(); return; diff --git a/packages/integration-tests/src/agent-session.ts b/packages/integration-tests/src/agent-session.ts index 2c6859ba4..8c5fdfd5f 100644 --- a/packages/integration-tests/src/agent-session.ts +++ b/packages/integration-tests/src/agent-session.ts @@ -58,6 +58,8 @@ export type AgentTurnResult = { 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 }; /** Present only when `acceptChanges` was requested. */ source?: AgentSourceSnapshot; }; @@ -241,6 +243,12 @@ export class AgentSession implements Disposable { 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 }), + }, ...(source === undefined ? {} : { source }), }; } catch (error) { From 035b2167e24d2a08cd13d7d8edf52bc54270b19b Mon Sep 17 00:00:00 2001 From: Ashish Kumar Singh Date: Mon, 24 Aug 2026 12:41:40 -0500 Subject: [PATCH 6/7] Adapt agent sessions to git-backed code --- packages/integration-tests/README.md | 126 ++++-------- .../__tests__/agent-session.test.ts | 29 +-- .../__tests__/gadget-durability.test.ts | 179 ------------------ .../__tests__/network-interceptor.test.ts | 29 --- .../fixtures/seeded-gadgets.ts | 101 ---------- packages/integration-tests/package.json | 1 - .../src/agent-session-internals.ts | 33 +--- .../integration-tests/src/agent-session.ts | 151 ++------------- .../src/network-interceptor.ts | 24 +-- pnpm-lock.yaml | 3 - 10 files changed, 63 insertions(+), 613 deletions(-) delete mode 100644 packages/integration-tests/__tests__/gadget-durability.test.ts delete mode 100644 packages/integration-tests/fixtures/seeded-gadgets.ts diff --git a/packages/integration-tests/README.md b/packages/integration-tests/README.md index a331f3b8f..3d6cd04a3 100644 --- a/packages/integration-tests/README.md +++ b/packages/integration-tests/README.md @@ -1,51 +1,31 @@ # 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 ``` -`workshop-backend/__integration__` also covers more than one module. It runs in-process under -`@cloudflare/vitest-pool-workers`, so it reaches Durable Object internals through `cloudflare:test`. -This package runs out-of-process and reaches only the public Cap'n Web API. - -## The toolkit - -Four source-only entry points, consumed both by the tests here and by per-vendor suites in repos that -vendor this one as a submodule: - -- **`src/agent-session.ts`** — `AgentSession`, a production-RPC driver for live agent evaluations. - It creates a fresh account and workspace, keeps one chat across turns, waits through callback-driven - agent restarts, reads complete paginated history, discovers workpieces, and optionally accepts and - snapshots source. It owns transport lifecycle only; evaluation semantics stay in the consuming suite. -- **`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. `passThroughHosts` exempts a host a suite - genuinely has to reach; a handler cannot, because it never receives the request body. -- **`src/rpc-client.ts`** — speaks Cap'n Web over a WebSocket to `/api`, the same transport the - browser uses. Local sessions use in-band password authentication. Preview sessions can attach a - Cloudflare Access application JWT to the WebSocket handshake. - -## 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. - -## Live agent evaluations - -Boot the harness with `enableGadgetExecution: true`; the default is false so existing suites do not -need a Worker Loader. Configure the Workshop through `patchWorkshop` with the real deployment model -credentials, then create the driver from the harness URL: +`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({ @@ -57,18 +37,24 @@ const harness = await startHarness({ 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: "anthropic,openai,google,cloudflare", + CF_AI_GATEWAY_PROVIDERS: "cloudflare", }; }, }); -using session = await AgentSession.create(harness.url, { modelId: "claude-sonnet-5" }); -const first = await session.run("Build a small status page."); -const accepted = await session.run("Add an incident timeline.", { acceptChanges: true }); +using session = await AgentSession.create(harness.url, { + modelId: "@cf/zai-org/glm-5.2", +}); +const result = await session.run("Build a small status page."); ``` -The same driver can use a deployed Access preview. The supplied JWT authenticates the WebSocket and -the session calls `authenticateFromCfAccess()` instead of creating a password account: +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), { @@ -77,43 +63,13 @@ using session = await AgentSession.create(new URL(previewUrl), { }); ``` -`run()` never accepts changes by default. When acceptance is requested it includes the current live -draft, then returns a Yjs V2 source snapshot keyed by each workpiece's `filesRoot`. Verifiers can use -`getGadget()` or typed `connectToGadget()`, selecting either the accepted or current chat branch. - -The unit suite uses no fake model protocol. A credentialed live evaluation is the end-to-end proof -for model execution and must be invoked explicitly by its consuming repository; it is not registered -as a silently skipped test here. - -## 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: +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. -- 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. +## Test isolation -## Further reading +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) 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. +[`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 index 50641cefa..5985dd7ec 100644 --- a/packages/integration-tests/__tests__/agent-session.test.ts +++ b/packages/integration-tests/__tests__/agent-session.test.ts @@ -1,14 +1,10 @@ import { afterEach, describe, expect, it, vi } from "vitest"; -import { Doc, Text } from "yjs"; -import type { - AiChatAuthorInfo, AiChatHistoryPage, AiChatMessage, AiChatMetadata, WorkpieceSummary, -} from "@gadgets/workshop-shared/api"; +import type { AiChatAuthorInfo, AiChatHistoryPage, AiChatMessage, AiChatMetadata } + from "@gadgets/workshop-shared/api"; import { finalAssistantText, } from "../src/agent-session.js"; -import { - AgentTurnCompletion, buildSourceSnapshot, loadAllChatHistory, -} from "../src/agent-session-internals.js"; +import { AgentTurnCompletion, loadAllChatHistory } from "../src/agent-session-internals.js"; function metadata(active: boolean): AiChatMetadata { return { @@ -133,22 +129,3 @@ it("returns the final non-empty assistant message from canonical history", () => expect(finalAssistantText(history)).toBe("first"); }); -it("maps Yjs V2 source by each workpiece filesRoot", () => { - const doc = new Doc(); - const legacy = doc.getMap(""); - const modern = doc.getMap("12"); - legacy.set("client.js", new Text("legacy client")); - modern.set("server.js", new Text("modern server")); - const workpieces: WorkpieceSummary[] = [ - { id: 1, type: "gadget", title: "Legacy", filesRoot: "" }, - { id: 12, type: "gadget", title: "Modern", filesRoot: "12" }, - { id: 13, type: "gadget", title: "Empty" }, - ]; - - const snapshot = buildSourceSnapshot(doc, 9, workpieces); - expect(snapshot.version).toBe(9); - expect(snapshot.workpieces.get("")?.summary.id).toBe(1); - expect(snapshot.workpieces.get("")?.files.get("client.js")).toBe("legacy client"); - expect(snapshot.workpieces.get("12")?.files.get("server.js")).toBe("modern server"); - expect(snapshot.workpieces.size).toBe(2); -}); diff --git a/packages/integration-tests/__tests__/gadget-durability.test.ts b/packages/integration-tests/__tests__/gadget-durability.test.ts deleted file mode 100644 index 6729d4612..000000000 --- a/packages/integration-tests/__tests__/gadget-durability.test.ts +++ /dev/null @@ -1,179 +0,0 @@ -import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { AgentSession } from "../src/agent-session.js"; -import { startHarness, type Harness, type WorkerConfig } from "../src/harness.js"; -import { - DURABLE_NOTES_SERVER, OVERSELLING_DESK_SERVER, VALID_SERVER_ONLY, -} from "../fixtures/seeded-gadgets.js"; - -// Scope: a Gadget facet restarting, and whether an application's state survives it. A whole Durable -// Object resetting is a separate concern, covered in `workshop-backend/__integration__`, which runs -// in-process and can abort one object through `cloudflare:test`. - -// Seeded gadgets never call a model, but AgentSession.create() selects one from listModels(), which -// is empty unless the Workshop believes a gateway is configured. These values are never dialled. -function offlineModelConfig(config: WorkerConfig): void { - config.vars = { - ...config.vars, - CF_AI_GATEWAY: "unused", - CF_AI_GATEWAY_ACCOUNT_ID: "unused", - CF_AI_GATEWAY_API_TOKEN: "unused", - CF_AI_GATEWAY_PROVIDERS: "cloudflare", - }; -} - -interface NotesApi { - put(input: { id: string; body: string }): Promise<{ ok: boolean }>; - all(): Promise<{ id: string; body: string }[]>; - liveness(): Promise<{ instance: string; callsSinceStart: number }>; -} - -interface PingApi { - ping(): Promise; -} - -interface DeskApi { - defineSlot(input: { slotId: string; capacity: number }): Promise<{ ok: boolean }>; - book(input: { bookingId: string; slotId: string }): - Promise<{ ok: true } | { ok: false; error: string }>; - bookedCount(input: { slotId: string }): Promise; -} - -let harness: Harness; - -beforeAll(async () => { - harness = await startHarness({ - gatekeepers: [], - enableGadgetExecution: true, - patchWorkshop: offlineModelConfig, - }); -}, 120_000); - -afterAll(async () => { - await harness?.server.close(); -}); - -describe("Gadget durability across abrupt server restarts", () => { - it("restarts the server for real, keeping storage and dropping memory", async () => { - using session = await AgentSession.create(harness.url, { usernamePrefix: "durable" }); - const id = await session.seedGadget({ - title: "Notes", bindingName: "NOTES", files: { "server.js": DURABLE_NOTES_SERVER }, - }); - - const before = await session.connectToGadget(id, "accepted"); - await before.put({ id: "a", body: "first" }); - const livenessBefore = await before.liveness(); - expect(livenessBefore.callsSinceStart).toBeGreaterThan(0); - - await session.restartGadgets(); - - // The abort invalidates outstanding stubs, which is what proves it actually happened. Without - // this the rest of the suite could pass against a restart that never occurred. - await expect(before.all()).rejects.toThrow(); - before[Symbol.dispose](); - - using after = await session.connectToGadget(id, "accepted"); - // Read the counter before anything else, since every other method bumps it. - const livenessAfter = await after.liveness(); - expect(livenessAfter.instance).not.toBe(livenessBefore.instance); - expect(livenessAfter.callsSinceStart).toBe(0); - expect(await after.all()).toEqual([{ id: "a", body: "first" }]); - }, 120_000); - - it("keeps every row across repeated restarts", async () => { - using session = await AgentSession.create(harness.url, { usernamePrefix: "durable" }); - const id = await session.seedGadget({ - title: "Notes", bindingName: "NOTES", files: { "server.js": DURABLE_NOTES_SERVER }, - }); - - const instances: string[] = []; - for (let round = 0; round < 5; round++) { - using api = await session.connectToGadget(id, "accepted"); - await api.put({ id: `note-${round}`, body: `body-${round}` }); - expect(await api.all()).toHaveLength(round + 1); - instances.push((await api.liveness()).instance); - await session.restartGadgets(); - } - - using api = await session.connectToGadget(id, "accepted"); - expect(await api.all()).toEqual( - Array.from({ length: 5 }, (_unused, round) => ({ id: `note-${round}`, body: `body-${round}` }))); - // Every round really did run in a fresh instance. - expect(new Set(instances).size).toBe(5); - }, 180_000); - - it("leaves storage consistent when a restart lands on an unfinished write", async () => { - using session = await AgentSession.create(harness.url, { usernamePrefix: "durable" }); - const id = await session.seedGadget({ - title: "Notes", bindingName: "NOTES", files: { "server.js": DURABLE_NOTES_SERVER }, - }); - - const writer = await session.connectToGadget(id, "accepted"); - await writer.put({ id: "settled", body: "already committed" }); - // Dispatched but deliberately not awaited, so the restart races it. - const inFlight = writer.put({ id: "racing", body: "may or may not land" }).catch(() => undefined); - await session.restartGadgets(); - await inFlight; - writer[Symbol.dispose](); - - using api = await session.connectToGadget(id, "accepted"); - const rows = await api.all(); - // Either outcome is correct for the racing write; what must hold is that the committed row is - // intact, the table is readable, and no half-written row appeared. - expect(rows).toContainEqual({ id: "settled", body: "already committed" }); - expect(rows.length).toBeGreaterThanOrEqual(1); - expect(rows.length).toBeLessThanOrEqual(2); - for (const row of rows) expect(typeof row.body).toBe("string"); - }, 120_000); -}); - -describe("Gadget module loading", () => { - // Every `.js` file in a Gadget becomes a module in its Worker, so workerd parses client.js at load - // even though the server never imports it. A test that verifies a Gadget through its RPC therefore - // already covers the syntax of both files, and needs no separate parse step. - it("refuses to start the server when client.js cannot parse", async () => { - using session = await AgentSession.create(harness.url, { usernamePrefix: "syntax" }); - const id = await session.seedGadget({ - title: "Unparseable", - bindingName: "UNPARSEABLE", - files: { "server.js": VALID_SERVER_ONLY, "client.js": "this is ( not valid javascript ===" }, - }); - - // Cap'n Web pipelines, so connecting resolves and the load failure surfaces on the first call. - using api = await session.connectToGadget(id, "accepted"); - await expect(api.ping()).rejects.toThrow(/SyntaxError/); - }, 120_000); - - it("starts the server when both files parse", async () => { - using session = await AgentSession.create(harness.url, { usernamePrefix: "syntax" }); - const id = await session.seedGadget({ - title: "Parseable", - bindingName: "PARSEABLE", - files: { "server.js": VALID_SERVER_ONLY, "client.js": "const unused = 1;" }, - }); - - using api = await session.connectToGadget(id, "accepted"); - expect(await api.ping()).toBe("ok"); - }, 120_000); -}); - -describe("Concurrent RPC interleaving", () => { - // Pins the platform behaviour that any "never oversells" assertion depends on: if a naive - // implementation could not oversell here, such an assertion would hold for the wrong reason. - it("lets a check-then-write implementation oversell", async () => { - using session = await AgentSession.create(harness.url, { usernamePrefix: "oversell" }); - const id = await session.seedGadget({ - title: "Broken Desk", bindingName: "DESK", files: { "server.js": OVERSELLING_DESK_SERVER }, - }); - - using api = await session.connectToGadget(id, "accepted"); - const capacity = 3; - await api.defineSlot({ slotId: "race", capacity }); - - const results = await Promise.all(Array.from({ length: 10 }, (_unused, index) => - api.book({ bookingId: `race-${index}`, slotId: "race" }))); - const accepted = results.filter(result => result.ok).length; - - expect(accepted).toBeGreaterThan(capacity); - expect(await api.bookedCount({ slotId: "race" })).toBeGreaterThan(capacity); - }, 120_000); -}); diff --git a/packages/integration-tests/__tests__/network-interceptor.test.ts b/packages/integration-tests/__tests__/network-interceptor.test.ts index cf400cb42..c37ee431e 100644 --- a/packages/integration-tests/__tests__/network-interceptor.test.ts +++ b/packages/integration-tests/__tests__/network-interceptor.test.ts @@ -121,32 +121,3 @@ it("uninstall restores the fetch that install captured", async () => { await expect(fetch("https://later.test/")).rejects.toThrow(/Unmocked outbound request/); interceptor.uninstall(); }); - -it("passes an allowed host through to the real fetch, body intact", async () => { - const seen: { url: string; method: string; body: string }[] = []; - const captured = globalThis.fetch; - globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { - const request = new Request(input, init); - seen.push({ url: request.url, method: request.method, body: await request.text() }); - return Response.json({ ok: true }); - }) as typeof globalThis.fetch; - - const interceptor = new NetworkInterceptor([], { passThroughHosts: ["allowed.test"] }); - interceptor.install(); - try { - const response = await fetch("https://allowed.test/v1/chat", { - method: "POST", - body: JSON.stringify({ prompt: "hello" }), - }); - expect(await response.json()).toEqual({ ok: true }); - // A handler never receives a body, so passing one through has to bypass handlers entirely. - expect(seen).toEqual([ - { url: "https://allowed.test/v1/chat", method: "POST", body: '{"prompt":"hello"}' }, - ]); - - await expect(fetch("https://denied.test/probe")).rejects.toThrow(/Unmocked outbound request/); - } finally { - interceptor.uninstall(); - globalThis.fetch = captured; - } -}); diff --git a/packages/integration-tests/fixtures/seeded-gadgets.ts b/packages/integration-tests/fixtures/seeded-gadgets.ts deleted file mode 100644 index c06013d2e..000000000 --- a/packages/integration-tests/fixtures/seeded-gadgets.ts +++ /dev/null @@ -1,101 +0,0 @@ -/** - * Hand-written Gadget sources for tests that need a known implementation rather than whatever an - * agent produced. Kept as strings because that is exactly what `AgentSession.seedGadget()` writes - * into the workspace's Yjs document. - */ - -/** - * Stores notes in SQL and deliberately also keeps state in memory, so a test can tell a restart - * apart from a no-op: `rows` must survive one and `instance`/`callsSinceStart` must not. - */ -export const DURABLE_NOTES_SERVER = ` -import { DurableObject } from "cloudflare:workers"; - -export class Gadget extends DurableObject { - constructor(ctx, env) { - super(ctx, env); - this.ctx = ctx; - // Memory, which a restart must discard. - this.instance = crypto.randomUUID(); - this.callsSinceStart = 0; - this.ctx.storage.sql.exec( - "CREATE TABLE IF NOT EXISTS notes (id TEXT PRIMARY KEY, body TEXT NOT NULL)"); - } - - async put(input) { - this.callsSinceStart++; - this.ctx.storage.sql.exec( - "INSERT INTO notes (id, body) VALUES (?, ?) " + - "ON CONFLICT(id) DO UPDATE SET body = excluded.body", - input.id, input.body); - return { ok: true }; - } - - async all() { - this.callsSinceStart++; - return [...this.ctx.storage.sql.exec("SELECT id, body FROM notes ORDER BY id")] - .map(row => ({ id: row.id, body: row.body })); - } - - async liveness() { - return { instance: this.instance, callsSinceStart: this.callsSinceStart }; - } -} -`; - -/** - * A booking gadget written the obvious wrong way: it reads the count, checks capacity, then awaits - * before inserting. Exists to prove that concurrent RPC calls really do interleave here, which is - * the premise the eval suite's overselling check depends on. - */ -export const OVERSELLING_DESK_SERVER = ` -import { DurableObject } from "cloudflare:workers"; - -export class Gadget extends DurableObject { - constructor(ctx, env) { - super(ctx, env); - this.ctx = ctx; - this.ctx.storage.sql.exec( - "CREATE TABLE IF NOT EXISTS slots (slotId TEXT PRIMARY KEY, capacity INTEGER NOT NULL)"); - this.ctx.storage.sql.exec( - "CREATE TABLE IF NOT EXISTS bookings (bookingId TEXT PRIMARY KEY, slotId TEXT NOT NULL)"); - } - - async defineSlot(input) { - this.ctx.storage.sql.exec( - "INSERT OR REPLACE INTO slots (slotId, capacity) VALUES (?, ?)", input.slotId, input.capacity); - return { ok: true }; - } - - async book(input) { - const slot = [...this.ctx.storage.sql.exec( - "SELECT capacity FROM slots WHERE slotId = ?", input.slotId)].at(0); - if (slot === undefined) return { ok: false, error: "UNKNOWN_SLOT" }; - const booked = [...this.ctx.storage.sql.exec( - "SELECT COUNT(*) AS n FROM bookings WHERE slotId = ?", input.slotId)].at(0).n; - if (booked >= slot.capacity) return { ok: false, error: "SLOT_FULL" }; - - // The bug under test: yielding between the check and the write lets every other queued call - // pass the same capacity check before any of them has inserted. - await scheduler.wait(1); - - this.ctx.storage.sql.exec( - "INSERT INTO bookings (bookingId, slotId) VALUES (?, ?)", input.bookingId, input.slotId); - return { ok: true }; - } - - async bookedCount(input) { - return [...this.ctx.storage.sql.exec( - "SELECT COUNT(*) AS n FROM bookings WHERE slotId = ?", input.slotId)].at(0).n; - } -} -`; - -/** A server that loads and answers, paired with a `client.js` that cannot parse. */ -export const VALID_SERVER_ONLY = ` -import { DurableObject } from "cloudflare:workers"; - -export class Gadget extends DurableObject { - async ping() { return "ok"; } -} -`; diff --git a/packages/integration-tests/package.json b/packages/integration-tests/package.json index 49a8417d0..9cd1a0fba 100644 --- a/packages/integration-tests/package.json +++ b/packages/integration-tests/package.json @@ -18,7 +18,6 @@ "@gadgets/workshop-shared": "workspace:*", "capnweb": "catalog:", "jsonc-parser": "^3.3.1", - "yjs": "^13.6.31", "ws": "^8.21.0", "zod": "^4.4.3" }, diff --git a/packages/integration-tests/src/agent-session-internals.ts b/packages/integration-tests/src/agent-session-internals.ts index f9102b656..8291bb3be 100644 --- a/packages/integration-tests/src/agent-session-internals.ts +++ b/packages/integration-tests/src/agent-session-internals.ts @@ -1,21 +1,5 @@ -import type { - AiChatHistoryPage, AiChatMessage, AiChatMetadata, WorkpieceSummary, -} from "@gadgets/workshop-shared/api"; -import * as Y from "yjs"; - -export type SourceWorkpiece = { - /** Workpiece whose `filesRoot` keys this entry. */ - summary: WorkpieceSummary; - /** UTF-8 source text keyed by file name. */ - files: ReadonlyMap; -}; - -export type SourceSnapshot = { - /** Workshop code version included in the snapshot. */ - version: number; - /** Source-bearing workpieces keyed by `WorkpieceSummary.filesRoot`. */ - workpieces: ReadonlyMap; -}; +import type { AiChatHistoryPage, AiChatMessage, AiChatMetadata } from "@gadgets/workshop-shared/api"; + export class AgentTurnCompletion { readonly promise: Promise; @@ -156,16 +140,3 @@ export async function loadAllChatHistory( return messages; } -export function buildSourceSnapshot( - doc: Y.Doc, version: number, summaries: readonly WorkpieceSummary[]): SourceSnapshot { - const workpieces = new Map(); - for (const summary of summaries) { - if (summary.filesRoot === undefined) continue; - const files = new Map(); - doc.getMap(summary.filesRoot).forEach((text, name) => { - files.set(name, text.toString()); - }); - workpieces.set(summary.filesRoot, { summary, files }); - } - return { version, workpieces }; -} diff --git a/packages/integration-tests/src/agent-session.ts b/packages/integration-tests/src/agent-session.ts index 8c5fdfd5f..ae98c8727 100644 --- a/packages/integration-tests/src/agent-session.ts +++ b/packages/integration-tests/src/agent-session.ts @@ -1,14 +1,11 @@ import type { RpcCompatible, RpcStub } from "capnweb"; import type { AiChatMessage, AiChatMetadata, AiChatStreamEvent, AiChatSubscriber, AiChatAuthorInfo, AiModelConfig, - AuthenticatedApi, CodeSubscriber, CodeUpdate, GadgetClient, OutputFormatOffer, Overseer, PublicApi, - WorkpieceId, WorkpieceSummary, WorkpiecesSubscriber, + AuthenticatedApi, GadgetClient, OutputFormatOffer, Overseer, PublicApi, WorkpieceId, + WorkpieceSummary, WorkpiecesSubscriber, } from "@gadgets/workshop-shared/api"; -import * as Y from "yjs"; -import { - AgentTurnCompletion, buildSourceSnapshot, loadAllChatHistory, -} from "./agent-session-internals.js"; -import type { SourceSnapshot } from "./agent-session-internals.js"; +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; @@ -45,8 +42,6 @@ export type AgentTurnOptions = { /** The branch a verifier should connect to. */ export type AgentGadgetBranch = "accepted" | "chat"; -/** Sources accepted from a turn, keyed by each `WorkpieceSummary.filesRoot`. */ -export type AgentSourceSnapshot = SourceSnapshot; /** Authoritative state returned after one agent turn settles. */ export type AgentTurnResult = { @@ -60,8 +55,6 @@ export type AgentTurnResult = { agentErrors: string[]; /** Provider usage available from chat metadata after this turn. */ usage: { totalTokens?: number; costUsd?: number }; - /** Present only when `acceptChanges` was requested. */ - source?: AgentSourceSnapshot; }; /** Return the final user-visible assistant text from canonical chat history. */ @@ -82,9 +75,9 @@ class ChatSubscriber extends RpcTarget implements AiChatSubscriber { metadata(chat: AiChatMetadata): void { this.completion?.metadata(chat); } deleted(_chatId: number): void {} message(_entry: AiChatMessage): void {} - draftUpdate( - _chatId: number, _timestamp: Date, _author: AiChatAuthorInfo, _update: Uint8Array): void {} - draftCleared(_chatId: number): void {} + changeApplied( + _chatId: number, _generation: number, _revision: number, _author: AiChatAuthorInfo, + _change: CodeChange, _submission?: {clientId: string; seq: number}): void {} stream(_chatId: number, _event: AiChatStreamEvent): void {} } @@ -103,25 +96,6 @@ class WorkpieceSubscriber extends RpcTarget implements WorkpiecesSubscriber { ready(): void { this.#resolveReady(); } } -class SourceSubscriber extends RpcTarget implements CodeSubscriber { - readonly readyPromise: Promise; - version = 0; - #doc: Y.Doc; - #resolveReady: () => void = () => {}; - - constructor(doc: Y.Doc) { - super(); - this.#doc = doc; - this.readyPromise = new Promise(resolve => { this.#resolveReady = resolve; }); - } - - update(update: CodeUpdate): void { - Y.applyUpdateV2(this.#doc, update.update); - this.version = update.version; - } - - ready(): void { this.#resolveReady(); } -} /** * Drives the production Workshop RPC lifecycle for one fresh user and workspace. @@ -233,9 +207,8 @@ export class AgentSession implements Disposable { await completion.promise; let history = await this.#loadHistory(this.#chatId); - let source: AgentSourceSnapshot | undefined; if (options.acceptChanges) { - source = await this.#acceptChanges(this.#chatId, history); + await this.acceptChanges(); history = await this.#loadHistory(this.#chatId); } return { @@ -249,7 +222,6 @@ export class AgentSession implements Disposable { ...(completion.lastMetadata?.totalCost === undefined ? {} : { costUsd: completion.lastMetadata.totalCost }), }, - ...(source === undefined ? {} : { source }), }; } catch (error) { this.#failed = true; @@ -308,96 +280,15 @@ export class AgentSession implements Disposable { }); } - /** - * Create a gadget with hand-written source, bypassing the agent entirely. - * - * For tests that need a known implementation — a deliberately broken one to prove an assertion - * can fail, or a deliberately correct one to measure the platform rather than the model. The - * gadget is permanent rather than provisional to a chat, so connect to it on the accepted branch. - * - * `files` must contain `server.js` exporting a Durable Object class named `Gadget`; only `.js` - * entries become worker modules. - */ - async seedGadget(spec: { - title: string; - bindingName: string; - files: Record; - }): Promise { - this.#assertUsable(); - using gadget = await this.#overseer.createGadget(spec.title, undefined, spec.bindingName); - const id = await gadget.getId(); - // The server owns the files root, so read it back rather than re-deriving the naming rule. - const filesRoot = await waitFor(`workpiece ${id} to publish its files root`, async () => - this.#workpieceSubscriber.entries.get(id)?.filesRoot ?? null); - - const doc = new Y.Doc(); - const subscriber = new SourceSubscriber(doc); - using subscriberStub = stubFor(subscriber); - let subscription: RpcStub<{}> | undefined; - const updates: Uint8Array[] = []; - const collect = (update: Uint8Array) => { updates.push(update); }; - try { - // Sync the doc before writing. A write from an unsynced doc is a concurrent Y.Map set that - // resolves by client ID, i.e. a coin flip; a write from a synced one carries a causal delete - // of the existing entry and deterministically wins. - subscription = await this.#overseer.subscribeToCode(subscriberStub); - await subscriber.readyPromise; - doc.on("updateV2", collect); - doc.transact(() => { - const root = doc.getMap(filesRoot); - for (const [name, content] of Object.entries(spec.files)) { - const text = new Y.Text(); - text.insert(0, content); - root.set(name, text); - } - }); - doc.off("updateV2", collect); - if (updates.length === 0) throw new Error("Seeding a gadget produced no code update"); - // updateCode bumps the workspace code version, which both aborts the facet and changes the - // Worker Loader cache key, so the next connect loads exactly what was just written. - await this.#overseer.updateCode(Y.mergeUpdatesV2(updates)); - return id; - } finally { - doc.off("updateV2", collect); - subscription?.[Symbol.dispose](); - doc.destroy(); - } - } - - /** - * Abruptly restart every Gadget server in this workspace, as the platform itself does whenever - * code changes. Storage is preserved; in-memory state is discarded. - * - * Existing gadget stubs are invalidated and throw on their next call, so callers must reconnect — - * which is also how a caller proves the restart really happened rather than silently no-opping. - * - * Implemented as an empty update to the mainline code, which advances the workspace's code - * version. That is what forces the restart, and it is also why this must not be called *between* - * turns of a multi-turn conversation: the agent stamps the code version it observed onto its - * history and rejects an inconsistent one on replay. Call it within the final turn, or in a - * single-turn task. - */ - async restartGadgets(): Promise { - this.#assertUsable(); - if (this.#turn !== undefined) { - throw new Error("Cannot restart gadgets while an agent turn is running"); - } - await this.#overseer.updateCode(Y.encodeStateAsUpdateV2(new Y.Doc())); - } - - /** Merge the current provisional chat branch and return its accepted source snapshot. */ - async acceptChanges(): Promise { + /** 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 history = await this.#loadHistory(this.#chatId); - return this.#acceptChanges(this.#chatId, history); - } - - async #acceptChanges(chatId: number, history: readonly AiChatMessage[]): Promise { - const mergeThrough = history.at(-1)?.sequence ?? null; - await this.#overseer.mergeChanges(chatId, mergeThrough, { includeDraft: true }); - return this.#readAcceptedSource(); + 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"); + } } /** Dispose subscriptions, callback targets, RPC capabilities, and the WebSocket session. */ @@ -428,20 +319,6 @@ export class AgentSession implements Disposable { return loadAllChatHistory(before => this.#overseer.getChatHistory(chatId, before)); } - async #readAcceptedSource(): Promise { - const doc = new Y.Doc(); - const subscriber = new SourceSubscriber(doc); - using subscriberStub = stubFor(subscriber); - let subscription: RpcStub<{}> | undefined; - try { - subscription = await this.#overseer.subscribeToCode(subscriberStub); - await subscriber.readyPromise; - return buildSourceSnapshot(doc, subscriber.version, this.workpieces()); - } finally { - subscription?.[Symbol.dispose](); - doc.destroy(); - } - } #stopCurrentAgent(): Promise { if (this.#chatId === undefined) return Promise.resolve(); diff --git a/packages/integration-tests/src/network-interceptor.ts b/packages/integration-tests/src/network-interceptor.ts index 18066a18a..506408229 100644 --- a/packages/integration-tests/src/network-interceptor.ts +++ b/packages/integration-tests/src/network-interceptor.ts @@ -18,29 +18,13 @@ export type Handler = (url: URL, method: string, headers: Headers) => Response | null | Promise; -/** Optional settings for {@link NetworkInterceptor}. */ -export type InterceptorOptions = { - /** - * Hostnames whose requests reach the real network untouched. - * - * A handler cannot stand in for a host that a suite genuinely has to reach, because a handler - * receives the URL, the method, and the headers, but never the body. Passing a real POST through - * therefore has to happen here, before the request is taken apart. - * - * Keep the list to hosts the suite cannot do without. Every other host still throws. - */ - passThroughHosts?: readonly string[]; -}; - export class NetworkInterceptor { readonly #handlers: readonly Handler[]; - readonly #passThroughHosts: ReadonlySet; #realFetch: typeof globalThis.fetch | null = null; #unmockedCalls: string[] = []; - constructor(handlers: Handler[] = [], options: InterceptorOptions = {}) { + constructor(handlers: Handler[] = []) { this.#handlers = [...handlers]; - this.#passThroughHosts = new Set(options.passThroughHosts ?? []); } install(): void { @@ -56,10 +40,8 @@ export class NetworkInterceptor { : input.url; const url = new URL(raw); - // The harness dispatches its own traffic over loopback; let that through untouched. An - // explicitly allowed host goes the same way, before the body can be disturbed below. - if (url.hostname === "localhost" || url.hostname === "127.0.0.1" || url.hostname === "[::1]" - || this.#passThroughHosts.has(url.hostname)) { + // The harness dispatches its own traffic over loopback; let that through untouched. + if (url.hostname === "localhost" || url.hostname === "127.0.0.1" || url.hostname === "[::1]") { return realFetch(input, init); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0d22696ba..eaef363fb 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -699,9 +699,6 @@ importers: ws: specifier: ^8.21.0 version: 8.21.3 - yjs: - specifier: ^13.6.31 - version: 13.6.31 zod: specifier: ^4.4.3 version: 4.4.3 From bd350cdb786356e7c73604eee157eb5c7add33c3 Mon Sep 17 00:00:00 2001 From: Ashish Kumar Singh Date: Mon, 24 Aug 2026 13:50:42 -0500 Subject: [PATCH 7/7] Delete preview workspaces after evals --- packages/integration-tests/src/agent-session.ts | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/packages/integration-tests/src/agent-session.ts b/packages/integration-tests/src/agent-session.ts index ae98c8727..c86dd9bfa 100644 --- a/packages/integration-tests/src/agent-session.ts +++ b/packages/integration-tests/src/agent-session.ts @@ -266,10 +266,8 @@ export class AgentSession implements Disposable { /** * Wait until the deployment's standard output formats are installed. - * - * Installation is fire-and-forget on the first `/api` request — the same request that opens this - * session — so a turn started immediately can race it and get a system prompt with no formats - * section, silently changing what the agent is told it can instantiate. + * 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(); @@ -291,6 +289,13 @@ export class AgentSession implements Disposable { } } + /** 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;