Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
JavaScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158import { requestedWebReader, WEB_INSTRUCTIONS } from "../shared/web.ts";import { SHELL_INSTRUCTIONS } from "../shared/shell.ts";import assert from "node:assert/strict";import { createHmac } from "node:crypto";import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";import { createServer } from "node:net";import { tmpdir } from "node:os";import { join, resolve } from "node:path";import { test } from "node:test";import { AgentClient } from "agents/client";import { WebSocketChatTransport } from "agents/chat/transport";import { MessageType } from "agents/chat";import WebSocket from "ws";import { unstable_dev } from "wrangler";import { DEFAULT_INSTRUCTIONS, MAX_INSTRUCTIONS_LENGTH,} from "../shared/instructions.ts";import { Secret } from "../configuration/secrets.ts";import { createOwnerSession } from "../worker/session.ts";import { customerBindings, installation } from "./fixtures/config.mjs";
const root = "/agents/personal-agent/personal";const pathFor = (id) => `${root}/sub/conversation/${id}`;const textOf = (history) => history .flatMap((m) => m.parts.filter((p) => p.type === "text").map((p) => p.text)) .join("\n");
test("explicit URL reads and pseudo-call retries require a real reader tool", () => { const message = (role, text) => ({ role, content: [{ type: "text", text }], }); assert.equal( requestedWebReader([ message( "user", "Visit https://news.ycombinator.com/item?id=49580164 and return the response in markdown.", ), ]), "read_url", ); assert.equal( requestedWebReader([ message( "user", "https://developers.cloudflare.com/fundamentals/oauth/create-an-oauth-client/ what's this telling me?", ), ]), "read_url", ); assert.equal( requestedWebReader([ message("user", "Render https://example.com in a browser."), ]), "browser_read", ); assert.equal( requestedWebReader([ message("user", "Visit https://example.com"), message( "assistant", 'I will try that. [read_url(url="https://example.com")]', ), message("user", "Doesn't seem to have worked"), ]), "read_url", ); assert.equal( requestedWebReader([message("user", "Visit https://example.com")], true), undefined, "tool-result continuations must be allowed to answer", ); assert.equal( requestedWebReader([ message("user", "Format https://example.com as a Markdown link"), ]), undefined, );});async function waitFor(predicate) { const deadline = Date.now() + 20_000; while (!(await predicate())) { if (Date.now() > deadline) throw new Error("Timed out waiting for Think"); await new Promise((resolve) => setTimeout(resolve, 30)); }}async function freePort() { const server = createServer(); await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); const { port } = server.address(); await new Promise((resolve) => server.close(resolve)); return port;}
function deniedSocket(url, headers, status) { return new Promise((resolve, reject) => { const socket = new WebSocket(url, { headers, handshakeTimeout: 5_000 }); socket.on("open", () => { socket.close(); reject(new Error("Unauthorized socket accepted")); }); socket.on("error", () => {}); socket.on("unexpected-response", (_request, response) => { response.resume(); socket.terminate(); try { assert.equal(response.statusCode, status); resolve(); } catch (error) { reject(error); } }); });}
test( "persistent conversations support CRUD, isolated native streaming, deletion, reconnect and recovery", { timeout: 180_000 }, async (t) => { const logs = []; for (const method of ["log", "info", "warn", "error"]) { const original = console[method]; t.mock.method(console, method, (...args) => { logs.push(args.map(String).join(" ")); original(...args); }); } const port = await freePort(); const origin = `http://127.0.0.1:${port}`; const secret = new Secret(customerBindings.FLAREBOT_SESSION_SECRET); const cookie = ( await createOwnerSession(secret, { ...installation, runtimeOrigin: origin, }) ).split(";")[0]; const headers = { Cookie: cookie, Origin: origin }; const persistence = await mkdtemp(join(tmpdir(), "flarebot-think-")); const config = JSON.parse( await readFile("dist/release/deployment.json", "utf8"), ); const configPath = join(persistence, "wrangler.json"); await writeFile( configPath, JSON.stringify({ ...config, name: "flarebot-think-test", no_bundle: false, keep_names: true, main: resolve("tests/fixtures/think-worker.ts"), assets: { ...config.assets, directory: resolve("dist/release/assets") }, }), ); const start = () => unstable_dev("tests/fixtures/think-worker.ts", { config: configPath, vars: { ...customerBindings, FLAREBOT_ENV: "development", FLAREBOT_DEV_OVERRIDES: JSON.stringify({ runtimeOrigin: origin }), }, local: true, ip: "127.0.0.1", port, inspectorPort: 0, persist: true, persistTo: persistence, logLevel: "error", experimental: { disableExperimentalWarning: true, watch: false }, }); const clients = []; const ownerFrames = []; let worker; try { worker = await start(); class AuthenticatedSocket extends WebSocket { constructor(url, protocols) { super(url, protocols, { headers, closeTimeout: 100 }); } } async function connectOwner() { const client = new AgentClient({ host: `127.0.0.1:${port}`, protocol: "ws", agent: "PersonalAgent", name: "personal", WebSocket: AuthenticatedSocket, }); clients.push(client); client.addEventListener("message", (event) => ownerFrames.push(event.data), ); await client.ready; return client; } let owner = await connectOwner(); const inspect = async () => (await fetch(`${origin}/__fixture/inspect`)).json(); const create = async (name) => owner.call("createConversation", name === undefined ? [] : [name]); assert.deepEqual(await owner.call("listConversations"), []); for (const name of [ null, 42, {}, "", " ", "a".repeat(121), "line\nbreak", "bad\u0000name", ]) { await assert.rejects(create(name), /Conversation name/); } assert.deepEqual(await owner.call("listConversations"), []); const history = async (id) => { const response = await fetch(origin + pathFor(id) + "/get-messages", { headers, }); assert.equal(response.status, 200, await response.clone().text()); assert.equal(response.headers.get("cache-control"), "no-store"); return response.json(); }; const firstMetadata = await create(" Research "); const firstId = firstMetadata.id; const secondMetadata = await create(); const secondId = secondMetadata.id; assert.equal(firstMetadata.name, "Research"); assert.equal(secondMetadata.name, "New conversation"); assert.deepEqual(Object.keys(firstMetadata).sort(), [ "createdAt", "id", "name", "updatedAt", ]); assert.ok(Number.isFinite(Date.parse(firstMetadata.createdAt))); for (const id of [ null, 1, {}, "", "../../personal", firstId.toUpperCase(), `${firstId}/get-messages`, ]) { await assert.rejects( owner.call("renameConversation", [id, "Renamed"]), /Invalid conversation ID/, ); await assert.rejects( owner.call("deleteConversation", [id]), /Invalid conversation ID/, ); } await assert.rejects( owner.call("renameConversation", [crypto.randomUUID(), "Renamed"]), /Conversation not found/, ); await assert.rejects( owner.call("renameConversation", [firstId, ""]), /Conversation name/, ); const renamed = await owner.call("renameConversation", [ firstId, "Renamed research", ]); assert.equal(renamed.name, "Renamed research"); assert.equal(renamed.createdAt, firstMetadata.createdAt); assert.ok(renamed.updatedAt >= firstMetadata.updatedAt); await assert.rejects( owner.call("subAgent", ["Conversation", crypto.randomUUID()]), /not callable/, ); await assert.rejects( owner.call("prepareConversation", [firstId]), /not callable/, ); for (const id of [firstId, secondId]) { assert.deepEqual(await history(id), []); for (const suffix of ["", "/get-messages"]) { assert.equal( (await fetch(origin + pathFor(id) + suffix)).status, 401, ); assert.equal( ( await fetch(origin + pathFor(id) + suffix, { headers: { ...headers, Origin: "https://attacker.example.com" }, }) ).status, 403, ); } await deniedSocket( origin.replace("http", "ws") + pathFor(id), { Origin: origin }, 401, ); } for (const path of [ pathFor(crypto.randomUUID()), `${pathFor(firstId)}/sub/conversation/${secondId}`, `/agents/conversation/${firstId}`, `${pathFor(firstId)}/arbitrary`, ]) { assert.equal((await fetch(origin + path, { headers })).status, 404); await deniedSocket(origin.replace("http", "ws") + path, headers, 404); } async function connect(id) { const frames = []; const client = new AgentClient({ host: `127.0.0.1:${port}`, protocol: "ws", agent: "PersonalAgent", basePath: pathFor(id).slice(1), WebSocket: AuthenticatedSocket, minReconnectionDelay: 50, maxReconnectionDelay: 250, }); const transport = new WebSocketChatTransport({ agent: client }); client.addEventListener("message", (event) => { const frame = JSON.parse(event.data); frames.push(frame); if (frame.type === MessageType.CF_AGENT_STREAM_RESUMING) transport.handleStreamResuming(frame); if (frame.type === MessageType.CF_AGENT_STREAM_RESUME_NONE) transport.handleStreamResumeNone(frame); if (frame.type === MessageType.CF_AGENT_STREAM_PENDING) transport.handleStreamPending(); }); clients.push(client); await Promise.race([ client.ready, new Promise((_, reject) => { const timer = setTimeout( () => reject( new Error( `Facet readiness timed out: ${JSON.stringify(frames)}`, ), ), 10_000, ); timer.unref(); }), ]); return { client, transport, frames }; } const first = await connect(firstId); const observer = await connect(firstId); const second = await connect(secondId); async function send(connection, id, text, body) { const messages = [ ...(await history(id)), { id: crypto.randomUUID(), role: "user", parts: [{ type: "text", text }], }, ]; const chunks = []; const stream = await connection.transport.sendMessages({ chatId: id, messages, body, trigger: "submit-message", abortSignal: new AbortController().signal, }); const done = (async () => { for await (const chunk of stream) chunks.push(chunk); })(); return { chunks, done }; } const nameOf = async (id) => (await owner.call("listConversations")).find((item) => item.id === id) ?.name; const titleCalls = async () => (await inspect()).titleCalls; await assert.rejects( owner.call("nameConversationFromFirstMessage", [secondId, "Injected"]), /not callable/, ); for (const [message, expected] of [ ["title:Configure Flarebot locally", "Configure Flarebot locally"], ["title:failure", "New conversation"], ["title:persistence-failure", "New conversation"], ["title:long-output", "New conversation"], ["title:unsafe-output", "New conversation"], ["title:bad-output", "New conversation"], ["title:opaque-output", "New conversation"], ["title:prefixed-output", "New conversation"], ["title:", "New conversation"], ["title:" + "a".repeat(121), "New conversation"], ["title:line\nbreak", "New conversation"], ['title:read_url(url="private")', "New conversation"], [ "title:Visit https://user:password@example.com/?token=private", "New conversation", ], ["title:API key private-credential", "New conversation"], ]) { const id = (await create()).id; assert.equal(await nameOf(id), "New conversation"); const connection = await connect(id); const before = (await titleCalls()).length; await ( await send(connection, id, message) ).done; await waitFor(async () => (await titleCalls()).length === before + 1); if (expected !== "New conversation") await waitFor(async () => (await nameOf(id)) === expected); const originalMessages = await history(id); const duplicate = await connection.transport.sendMessages({ chatId: id, messages: originalMessages.filter((item) => item.role === "user"), trigger: "submit-message", abortSignal: new AbortController().signal, }); for await (const _chunk of duplicate) { /* drain native duplicate replay */ } await ( await send(connection, id, "second") ).done; assert.equal(await nameOf(id), expected); assert.equal((await titleCalls()).length, before + 1); connection.client.close(); await owner.call("deleteConversation", [id]); } assert.ok( ownerFrames.some( (frame) => JSON.parse(frame).type === "conversations-changed", ), ); for (const customName of ["Custom name", "New conversation"]) { const id = (await create(customName)).id; const connection = await connect(id); const before = (await titleCalls()).length; await ( await send(connection, id, "title:Configure Flarebot locally") ).done; assert.equal(await nameOf(id), customName); assert.equal((await titleCalls()).length, before); connection.client.close(); await owner.call("deleteConversation", [id]); } const raceId = (await create()).id; const race = await connect(raceId); const beforeRace = (await titleCalls()).length; const raceTurn = await send(race, raceId, "title:slow"); await raceTurn.done; await waitFor(async () => (await titleCalls()).length === beforeRace + 1); assert.equal( await nameOf(raceId), "New conversation", "chat completes before title inference", ); await owner.call("renameConversation", [raceId, "Manual name"]); await owner.call("renameConversation", [raceId, "New conversation"]); await new Promise((resolve) => setTimeout(resolve, 1400)); await ( await send(race, raceId, "second") ).done; assert.equal( await nameOf(raceId), "New conversation", "manual rename back to default still wins", ); assert.equal((await titleCalls()).length, beforeRace + 1); race.client.close(); await owner.call("deleteConversation", [raceId]); await ( await send(second, secondId, "title:Understand Cloudflare OAuth") ).done; await waitFor( async () => (await nameOf(secondId)) === "Understand Cloudflare OAuth", ); const defaults = await owner.call("getInstructionSettings"); assert.deepEqual(defaults, { instructions: DEFAULT_INSTRUCTIONS, defaultInstructions: DEFAULT_INSTRUCTIONS, customized: false, updatedAt: null, }); await assert.rejects(owner.call("readInstructions"), /not callable/); const assertInstructions = async (connection, id, expected) => { const turn = await send(connection, id, "instructions"); await turn.done; const reply = textOf([(await history(id)).at(-1)]); const system = JSON.parse( reply.slice("Reply ".length, -" complete".length), ); assert.equal(system.length, 1); const prefix = expected + WEB_INSTRUCTIONS + SHELL_INSTRUCTIONS; assert.equal( system[0].slice(0, prefix.length), prefix, "model sees current effective instructions and existing tool guidance, no stale frozen prompt", ); assert.match( system[0].slice(prefix.length), /^\n\nScheduling:\nUse createSchedule/, ); assert.match( system[0], /\nCurrent UTC time: \d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{3}Z\n$/, ); }; await assertInstructions(first, firstId, DEFAULT_INSTRUCTIONS); const customInstructions = "Use concise Polish answers.\nKeep my personal preferences private.\tAsk when uncertain."; const customSettings = await owner.call("updateInstructions", [ customInstructions, ]); assert.equal(customSettings.instructions, customInstructions); assert.equal(customSettings.customized, true); assert.ok(Number.isFinite(Date.parse(customSettings.updatedAt))); for (const value of [ null, false, 42, {}, "", " \n\t", "x".repeat(MAX_INSTRUCTIONS_LENGTH + 1), "bad\u0000text", "bad\u007ftext", ]) { await assert.rejects( owner.call("updateInstructions", [value]), /Instructions must/, ); assert.deepEqual( await owner.call("getInstructionSettings"), customSettings, ); } await Promise.all([ assertInstructions(first, firstId, customInstructions), assertInstructions(second, secondId, customInstructions), ]); const instructionId = (await create()).id; const instructionConnection = await connect(instructionId); await assertInstructions( instructionConnection, instructionId, customInstructions, ); instructionConnection.client.close(); await owner.call("deleteConversation", [instructionId]); const resetSettings = await owner.call("resetInstructions"); assert.equal(resetSettings.customized, false); assert.equal(resetSettings.instructions, DEFAULT_INSTRUCTIONS); await assertInstructions(first, firstId, DEFAULT_INSTRUCTIONS); await owner.call("updateInstructions", [customInstructions]); assert.ok( !JSON.stringify(await owner.call("getStatus")).includes( customInstructions, ), ); const defaultSettings = await owner.call("getModelSettings"); assert.deepEqual(defaultSettings, { configuration: { provider: "workers-ai", model: "@cf/meta/llama-3.3-70b-instruct-fp8-fast", }, providers: { "workers-ai": true, anthropic: true, openrouter: false, }, providerSetupUrl: null, credentials: { anthropic: "missing", openrouter: "missing" }, }); const catalog = await owner.call("getModelCatalog"); const external = { provider: "anthropic", model: catalog.anthropic[0] }; const apiKey = "sk-ant-fixture-original-key-never-return"; const replacementKey = "sk-ant-fixture-replacement-key-never-return"; await assert.rejects( owner.call("readModelConfiguration"), /not callable/, ); await owner.call("updateModelSettings", [external]); const missingKeyTurn = await send(first, firstId, "configuration"); await assert.rejects(missingKeyTurn.done, /Add an Anthropic API key/); const configured = await owner.call("setProviderKey", [ "anthropic", apiKey, ]); assert.deepEqual(configured, { ...defaultSettings, configuration: external, credentials: { anthropic: "configured", openrouter: "missing" }, }); for (const invalid of [ null, {}, { ...external, apiKey }, { ...external, model: "arbitrary" }, { provider: "workers-ai", model: external.model }, ]) { await assert.rejects( owner.call("updateModelSettings", [invalid]), /supported provider and model/, ); assert.deepEqual(await owner.call("getModelSettings"), configured); } await assert.rejects( owner.call("setProviderKey", ["anthropic", "bad\nkey"]), /API key/, ); assert.deepEqual(await owner.call("getModelSettings"), configured); const assertConfiguration = async (connection, id, configuration) => { const turn = await send(connection, id, "configuration"); await turn.done; assert.equal( textOf([(await history(id)).at(-1)]), `Reply ${configuration.provider}/${configuration.model} complete`, ); }; // The composer sends configuration in the native request, not a separate // settings mutation. Each step sees that same model and effort. const scopedId = (await create("Scoped model and effort")).id; const scoped = await connect(scopedId); const selected = { ...external, effort: "low" }; assert.equal( (await scoped.client.call("getConversationModelSettings")).override, null, ); await ( await send(scoped, scopedId, "tool", { modelConfiguration: selected }) ).done; const calls = await scoped.client.call("fixtureModelCalls"); assert.ok(calls.length >= 2, "tool follow-up must invoke the model"); for (const call of calls) { assert.deepEqual(call.configuration, selected); assert.deepEqual(call.options.anthropic, { effort: "low", thinking: { type: "adaptive" }, }); } assert.deepEqual( (await scoped.client.call("getConversationModelSettings")).override, selected, ); assert.equal( (await second.client.call("getConversationModelSettings")).override, null, ); assert.deepEqual( (await owner.call("getModelSettings")).configuration, external, ); const invalidEffort = await send(scoped, scopedId, "configuration", { modelConfiguration: { ...selected, effort: "off" }, }); await assert.rejects(invalidEffort.done, /supported effort/); assert.deepEqual( (await scoped.client.call("getConversationModelSettings")).override, selected, ); scoped.client.close(); const reopened = await connect(scopedId); assert.deepEqual( (await reopened.client.call("getConversationModelSettings")).override, selected, ); await assertConfiguration(reopened, scopedId, selected); // Reset applies an explicit snapshot to this turn while clearing the // conversation override for future default-following turns. await ( await send(reopened, scopedId, "configuration", { modelConfiguration: defaultSettings.configuration, useDefaultModel: true, }) ).done; assert.equal( (await reopened.client.call("getConversationModelSettings")).override, null, ); reopened.client.close(); await owner.call("deleteConversation", [scopedId]); const openRouterConfiguration = { provider: "openrouter", model: catalog["openrouter"][0], }; const openRouterKey = "openrouter-fixture-private-key-never-return"; await assert.rejects( owner.call("updateModelSettings", [openRouterConfiguration]), /Enable OpenRouter in Settings/, ); const enabled = await fetch(`${origin}/__fixture/enable-openrouter`, { method: "POST", }); assert.equal(enabled.status, 204); await enabled.arrayBuffer(); await owner.call("updateModelSettings", [openRouterConfiguration]); const missingOpenRouterKey = await send(first, firstId, "configuration"); await assert.rejects( missingOpenRouterKey.done, /Add an OpenRouter API key/, ); const openRouterSettings = await owner.call("setProviderKey", [ "openrouter", openRouterKey, ]); assert.deepEqual(openRouterSettings.credentials, { anthropic: "configured", openrouter: "configured", }); await assertConfiguration(first, firstId, openRouterConfiguration); assert.ok( !JSON.stringify((await inspect()).credentialsStorage).includes( openRouterKey, ), ); await owner.call("setProviderKey", ["openrouter", null]); assert.deepEqual((await owner.call("getModelSettings")).credentials, { anthropic: "configured", openrouter: "missing", }); // A removal only affects its own provider. Keep OpenRouter configured to verify // independent ciphertext survives the later legacy Anthropic migration. await owner.call("setProviderKey", ["openrouter", openRouterKey]); await owner.call("updateModelSettings", [external]); await Promise.all([ assertConfiguration(first, firstId, external), assertConfiguration(second, secondId, external), ]); const newId = (await create("New settings conversation")).id; const newlyCreated = await connect(newId); await assertConfiguration(newlyCreated, newId, external); newlyCreated.client.close(); await owner.call("deleteConversation", [newId]); const providerFailure = await send(first, firstId, "credential-error"); await assert.rejects( providerFailure.done, /Anthropic rejected authentication/, ); await owner.call("setProviderKey", ["anthropic", replacementKey]); const secondWorkersModel = { provider: "workers-ai", model: catalog["workers-ai"][1], }; await owner.call("updateModelSettings", [secondWorkersModel]); await assertConfiguration(first, firstId, secondWorkersModel); await owner.call("updateModelSettings", [defaultSettings.configuration]); const firstTurn = await send(first, firstId, "slow-first"); await waitFor(() => firstTurn.chunks.some((c) => c.type === "text-delta"), ); assert.ok( !firstTurn.chunks.some((c) => c.type === "finish"), "stream yields before turn finishes", ); const secondTurn = await send(second, secondId, "second"); await secondTurn.done; await firstTurn.done; await waitFor(async () => /Reply slow-first complete/.test(textOf(await history(firstId))), ); await waitFor(async () => /Reply second complete/.test(textOf(await history(secondId))), ); assert.ok(!textOf(await history(secondId)).includes("slow-first")); await waitFor(() => JSON.stringify(observer.frames).includes("slow-first"), ); const toolTurn = await send(first, firstId, "tool"); await toolTurn.done; const toolHistory = await history(firstId); assert.ok( toolHistory .flatMap((m) => m.parts) .some( (p) => p.type === "tool-fixtureEcho" && p.state === "output-available" && p.output.echoed === "native tool output", ), ); assert.match(textOf(toolHistory), /Reply tool complete/);
const cancelTurn = await send(first, firstId, "slow-cancel"); await waitFor(() => cancelTurn.chunks.some((c) => c.type === "text-delta"), ); assert.equal(first.transport.cancelActiveServerTurn(), true); await assert.rejects(cancelTurn.done, { name: "AbortError" }); await waitFor( async () => (await history(firstId)).at(-1)?.role === "assistant", ); const cancelled = (await history(firstId)).at(-1); assert.equal(textOf([cancelled]), "Reply "); const errorTurn = await send(first, firstId, "error"); await assert.rejects(errorTurn.done, /Workers AI request failed/); const afterError = await send(first, firstId, "after-error"); await afterError.done; assert.match( textOf(await history(firstId)), /Reply after-error complete/, );
// Detaching a tab preserves the durable server turn; another native transport // reattaches to the buffered stream without submitting the user message twice. const detachedTurn = await send(first, firstId, "slow-detach"); detachedTurn.done.catch(() => {}); await waitFor(() => detachedTurn.chunks.some((c) => c.type === "text-delta"), ); first.client.close(); const reconnect = await connect(firstId); const resumed = await reconnect.transport.reconnectToStream({ chatId: firstId, }); assert.ok(resumed); const resumedChunks = []; for await (const chunk of resumed) resumedChunks.push(chunk); assert.ok(resumedChunks.some((c) => c.type === "text-delta")); assert.equal( (await history(firstId)).filter( (m) => m.role === "user" && textOf([m]) === "slow-detach", ).length, 1, ); const stableSecondHistory = await history(secondId);
// Delete a live stream while observers, history requests and stale frames // still address it. Native parent bridging must not recreate the facet. const deletedId = (await create("Delete while running")).id; const liveDeleted = await connect(deletedId); const deletedObserver = await connect(deletedId); const deletedTurn = await send(liveDeleted, deletedId, "slow-delete"); deletedTurn.done.catch(() => {}); await waitFor(() => deletedTurn.chunks.some((c) => c.type === "text-delta"), ); const closedCodes = []; for (const connection of [liveDeleted, deletedObserver]) connection.client.addEventListener("close", (event) => closedCodes.push(event.code), ); const staleFrames = setInterval( () => liveDeleted.client.send( JSON.stringify({ type: MessageType.CF_AGENT_STREAM_RESUME }), ), 5, ); try { await Promise.all([ owner.call("deleteConversation", [deletedId]), owner.call("deleteConversation", [deletedId]), ...Array.from({ length: 5 }, async () => { const response = await fetch( origin + pathFor(deletedId) + "/get-messages", { headers }, ); assert.ok([200, 404, 503].includes(response.status)); }), ]); await waitFor(() => closedCodes.length >= 2); } finally { clearInterval(staleFrames); liveDeleted.client.close(); deletedObserver.client.close(); } assert.deepEqual(closedCodes.slice(0, 2), [4004, 4004]); await waitFor(async () => !(await inspect()).facets.includes(deletedId)); await assert.rejects( owner.call("renameConversation", [deletedId, "Resurrect"]), /Conversation not found/, ); for (const id of [deletedId, crypto.randomUUID()]) { await owner.call("deleteConversation", [id]); assert.equal( (await fetch(origin + pathFor(id) + "/get-messages", { headers })) .status, 404, ); await deniedSocket( origin.replace("http", "ws") + pathFor(id), headers, 404, ); } const stableMetadata = await owner.call("listConversations"); assert.deepEqual( stableMetadata.map((item) => item.id).sort(), [firstId, secondId].sort(), ); assert.deepEqual( stableMetadata.find((item) => item.id === firstId), renamed, ); assert.deepEqual((await inspect()).facets, [firstId, secondId].sort());
// Failed teardown stays inaccessible and is retried on the next full wake. const failedDeleteId = (await create("Failed teardown")).id; await fetch(`${origin}/__fixture/fail-delete`, { method: "POST" }); await assert.rejects( owner.call("deleteConversation", [failedDeleteId]), /Fixture deletion failure/, ); const interruptedCreateId = await ( await fetch(`${origin}/__fixture/interrupted-create`, { method: "POST", }) ).json(); for (const id of [failedDeleteId, interruptedCreateId]) { assert.equal( (await fetch(origin + pathFor(id) + "/get-messages", { headers })) .status, 404, ); await deniedSocket( origin.replace("http", "ws") + pathFor(id), headers, 404, ); } assert.deepEqual(await owner.call("listConversations"), stableMetadata); await owner.call("updateModelSettings", [external]); const beforeRestartTitleCalls = await titleCalls(); const beforeRestartSettings = await owner.call("getModelSettings"); const beforeRestartStorage = (await inspect()).settingsStorage; const beforeRestartCredentials = (await inspect()).credentialsStorage; const beforeRestartEnabledProviders = (await inspect()) .enabledProvidersStorage; assert.ok(!JSON.stringify(beforeRestartStorage).includes(replacementKey)); const recoveryTurn = await send(reconnect, firstId, "recover"); recoveryTurn.done.catch(() => {}); await waitFor(() => recoveryTurn.chunks.some((c) => c.type === "text-delta"), ); // Allow Think's native buffered stream checkpoint to reach SQLite. await new Promise((resolve) => setTimeout(resolve, 400)); // Emulate the previous schema at rest after this turn has read its key. // Migration runs on the actual Worker restart below. const retiredResponse = await fetch( `${origin}/__fixture/retired-opencode`, { method: "POST" }, ); assert.equal(retiredResponse.status, 204); await retiredResponse.arrayBuffer(); const legacyResponse = await fetch(`${origin}/__fixture/legacy-key`, { method: "POST", }); assert.equal(legacyResponse.status, 204); await legacyResponse.arrayBuffer(); const stagedRetired = await inspect(); assert.equal( JSON.parse(stagedRetired.settingsStorage[0].configuration).provider, "opencode-go", ); assert.ok( stagedRetired.credentialsStorage.some( (row) => row.provider === "opencode-go", ), ); assert.ok( stagedRetired.enabledProvidersStorage.some( (row) => row.provider === "opencode-go", ), ); for (const client of clients) client.close(); await worker.stop(); worker = undefined; worker = await start(); owner = await connectOwner(); const migratedSettings = await owner.call("getModelSettings"); assert.deepEqual(migratedSettings, { ...beforeRestartSettings, configuration: defaultSettings.configuration, }); assert.equal( (await owner.call("getInstructionSettings")).instructions, customInstructions, ); assert.equal( (await owner.call("getInstructionSettings")).customized, true, ); const migrated = await inspect(); assert.deepEqual(migrated.settingsStorage, [ { ...beforeRestartStorage[0], configuration: JSON.stringify(defaultSettings.configuration), }, ]); assert.deepEqual(migrated.credentialsStorage, beforeRestartCredentials); assert.deepEqual( migrated.enabledProvidersStorage, beforeRestartEnabledProviders, ); assert.ok( !migrated.credentialsStorage.some( (row) => row.provider === "opencode-go", ), ); assert.ok( !migrated.enabledProvidersStorage.some( (row) => row.provider === "opencode-go", ), ); assert.equal(migratedSettings.providers.openrouter, true); assert.equal(migratedSettings.credentials.openrouter, "configured"); assert.equal(migratedSettings.credentials.anthropic, "configured");
// A second full wake must leave the migrated rows unchanged. owner.close(); await worker.stop(); worker = undefined; worker = await start(); owner = await connectOwner(); assert.deepEqual(await owner.call("getModelSettings"), migratedSettings); const migratedAgain = await inspect(); assert.deepEqual(migratedAgain.settingsStorage, migrated.settingsStorage); assert.deepEqual( migratedAgain.credentialsStorage, migrated.credentialsStorage, ); assert.deepEqual( migratedAgain.enabledProvidersStorage, migrated.enabledProvidersStorage, ); await owner.call("updateModelSettings", [external]); assert.deepEqual(await owner.call("listConversations"), stableMetadata); const restarted = await inspect(); assert.deepEqual(restarted.facets, [firstId, secondId].sort()); assert.deepEqual( restarted.metadata, [firstId, secondId].sort().map((id) => ({ id, status: "active" })), ); assert.equal( ( await fetch(origin + pathFor(deletedId) + "/get-messages", { headers, }) ).status, 404, ); await deniedSocket( origin.replace("http", "ws") + pathFor(deletedId), headers, 404, ); const recovered = await connect(firstId); await waitFor(async () => /Recovered recover complete/.test(textOf(await history(firstId))), ); assert.deepEqual(await history(secondId), stableSecondHistory); assert.equal( (await history(firstId)).filter( (m) => m.role === "user" && textOf([m]) === "recover", ).length, 1, ); assert.equal( textOf([(await history(firstId)).find((m) => m.id === cancelled.id)]), "Reply ", "cancelled turn is not resumed after restart", ); assert.ok(!JSON.stringify(recovered.frames).includes(secret.reveal())); await assertInstructions(recovered, firstId, customInstructions); await assertConfiguration(recovered, firstId, external); const secondReconnected = await connect(secondId); await assertConfiguration(secondReconnected, secondId, external); assert.deepEqual(await titleCalls(), beforeRestartTitleCalls); assert.equal(await nameOf(secondId), "Understand Cloudflare OAuth"); await owner.call("setProviderKey", ["anthropic", null]); assert.equal( (await owner.call("getModelSettings")).credentials.anthropic, "missing", ); assert.equal((await inspect()).settingsStorage[0].anthropic_key, null); assert.ok( !(await inspect()).credentialsStorage.some( (row) => row.provider === "anthropic", ), ); const removedKeyTurn = await send(recovered, firstId, "configuration"); await assert.rejects(removedKeyTurn.done, /Add an Anthropic API key/); await owner.call("updateModelSettings", [defaultSettings.configuration]); await assertConfiguration( recovered, firstId, defaultSettings.configuration, ); const stateFrames = ownerFrames .map((frame) => JSON.parse(frame)) .filter((frame) => frame.type === "cf_agent_state"); assert.ok(stateFrames.length > 0); assert.ok( !JSON.stringify(stateFrames).includes(customInstructions), "custom text is never a generic native state broadcast", ); const surfaces = JSON.stringify({ ownerFrames, logs, first: first.frames, second: second.frames, recovered: recovered.frames, firstHistory: await history(firstId), secondHistory: await history(secondId), status: await owner.call("getStatus"), }); assert.ok( logs.some((line) => line.includes("Anthropic rejected authentication")), "native error logs were captured", ); for (const sensitive of [ apiKey, replacementKey, openRouterKey, secret.reveal(), ]) assert.ok( !surfaces.includes(sensitive), "credentials absent from native protocol, snapshots, history and logs", );
const [cookieName, signedCookie] = cookie.split("="); const claims = JSON.parse( Buffer.from(signedCookie.split(".")[0], "base64url").toString(), ); const now = Math.floor(Date.now() / 1000); const shortBody = Buffer.from( JSON.stringify({ ...claims, issuedAt: now, expiresAt: now + 2 }), ).toString("base64url"); const expiryCookie = `${cookieName}=${shortBody}.${createHmac("sha256", secret.reveal()).update(shortBody).digest("base64url")}`; const expiring = new WebSocket( origin.replace("http", "ws") + pathFor(secondId), { headers: { Origin: origin, Cookie: expiryCookie }, closeTimeout: 100, }, ); const closed = await new Promise((resolve, reject) => { const timeout = setTimeout(() => { expiring.terminate(); reject(new Error("Facet session did not expire")); }, 10_000); expiring.on("error", reject); expiring.on("close", (code) => { clearTimeout(timeout); resolve(code); }); }); assert.equal(closed, 4001); } finally { for (const client of clients) client.close(); await worker?.stop(); await rm(persistence, { recursive: true, force: true }); } },);