From 528f5bd15bf4aa4b8bf8bae78ab06b58389a4ad9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Peer=20St=C3=B6cklmair?= Date: Tue, 28 Jul 2026 20:15:12 +0300 Subject: [PATCH] feat(v10/cloudflare): Add instrumentAgentWithSentry for Cloudflare Agents Backport of: #22718 --- .../cloudflare-agent/package.json | 34 +- .../cloudflare-agent/tests/agent-socket.ts | 90 ++++ .../tests/ai-streaming.test.ts | 60 +++ .../cloudflare-agent/tests/callable.test.ts | 62 ++- .../tests/chat-conversation.test.ts | 32 ++ .../cloudflare-agent/tests/chat-rpc.test.ts | 30 ++ .../worker-configuration.d.ts | 3 +- .../cloudflare-agent/worker/index.ts | 106 ++-- .../cloudflare-agent/worker/mocks.ts | 44 ++ .../cloudflare-agent/wrangler.jsonc | 7 +- packages/cloudflare/src/durableobject.ts | 456 ++++++++++++------ packages/cloudflare/src/index.ts | 2 +- .../src/instrumentations/agents/index.ts | 34 ++ .../agents/instrumentAgentCallableRpc.ts | 69 +++ .../agents/instrumentChatAgentConversation.ts | 29 ++ .../src/instrumentations/agents/types.ts | 62 +++ packages/cloudflare/test/agents.test.ts | 69 +++ .../test/instrumentCloudflareAgent.test.ts | 161 +++++++ 18 files changed, 1141 insertions(+), 209 deletions(-) create mode 100644 dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/agent-socket.ts create mode 100644 dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/ai-streaming.test.ts create mode 100644 dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-conversation.test.ts create mode 100644 dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-rpc.test.ts create mode 100644 dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/mocks.ts create mode 100644 packages/cloudflare/src/instrumentations/agents/index.ts create mode 100644 packages/cloudflare/src/instrumentations/agents/instrumentAgentCallableRpc.ts create mode 100644 packages/cloudflare/src/instrumentations/agents/instrumentChatAgentConversation.ts create mode 100644 packages/cloudflare/src/instrumentations/agents/types.ts create mode 100644 packages/cloudflare/test/agents.test.ts create mode 100644 packages/cloudflare/test/instrumentCloudflareAgent.test.ts diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/package.json b/dev-packages/e2e-tests/test-applications/cloudflare-agent/package.json index 6c834c1d9f47..21ed3f3a9b8b 100644 --- a/dev-packages/e2e-tests/test-applications/cloudflare-agent/package.json +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/package.json @@ -14,26 +14,32 @@ "test:dev": "TEST_ENV=development playwright test" }, "dependencies": { - "@cloudflare/ai-chat": "^0.7.1", - "@sentry/cloudflare": "^10.53.1", - "agents": "^0.13.1", - "react": "^19.2.6", - "react-dom": "^19.2.6" + "@cloudflare/ai-chat": "^0.10.0", + "@sentry/cloudflare": "^10.68.0", + "@sentry/core": "^10.68.0", + "agents": "^0.20.0", + "ai": "^6.0.235", + "react": "^19.2.8", + "react-dom": "^19.2.8", + "workers-ai-provider": "^3.3.1" }, "devDependencies": { "@playwright/test": "~1.56.0", - "@cloudflare/vite-plugin": "^1.37.2", - "@cloudflare/workers-types": "^4.20260520.1", + "@cloudflare/vite-plugin": "^1.47.0", + "@cloudflare/workers-types": "^5.20260727.1", "@sentry-internal/test-utils": "link:../../../test-utils", - "@types/node": "^24.12.4", - "@types/react": "^19.2.15", + "@types/node": "^26.1.2", + "@types/react": "^19.2.17", "@types/react-dom": "^19.2.3", - "@vitejs/plugin-react": "^6.0.2", - "globals": "^17.6.0", + "@types/ws": "^8.18.1", + "@babel/core": "^8.0.1", + "@babel/plugin-proposal-decorators": "^8.0.2", + "@vitejs/plugin-react": "^6.0.4", + "globals": "^17.8.0", "typescript": "~6.0.3", - "vite": "^8.0.14", - "wrangler": "^4.93.0", - "ws": "^8.20.1" + "vite": "^8.1.5", + "wrangler": "^4.114.0", + "ws": "^8.21.1" }, "volta": { "node": "24.15.0", diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/agent-socket.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/agent-socket.ts new file mode 100644 index 000000000000..96437eb47738 --- /dev/null +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/agent-socket.ts @@ -0,0 +1,90 @@ +import WebSocket from 'ws'; + +type AgentReply = { type?: string; id?: string; done?: boolean }; + +/** Sends a single frame over a WS to the given agent instance and resolves once `isDone` matches a reply. */ +function driveAgentSocket( + baseURL: string, + binding: string, + instance: string, + frame: unknown, + isDone: (reply: AgentReply) => boolean, + timeoutLabel: string, +): Promise { + const wsUrl = `${baseURL.replace(/^http/, 'ws')}/agents/${binding}/${instance}`; + + return new Promise((resolveSocket, rejectSocket) => { + const socket = new WebSocket(wsUrl); + const timeout = setTimeout(() => { + socket.close(); + rejectSocket(new Error(`Timed out waiting for ${timeoutLabel}`)); + }, 15_000); + + socket.on('open', () => { + socket.send(JSON.stringify(frame)); + }); + + socket.on('message', data => { + try { + const parsed = JSON.parse(data.toString()) as AgentReply; + if (isDone(parsed)) { + clearTimeout(timeout); + socket.close(); + resolveSocket(); + } + } catch { + // Ignore non-JSON / unrelated frames. + } + }); + + socket.on('error', err => { + clearTimeout(timeout); + rejectSocket(err); + }); + }); +} + +/** Opens a chat WebSocket to `/agents//`, sends one `cf_agent_use_chat_request` frame. */ +export function sendChatMessage( + baseURL: string, + options: { binding: string; instance: string; prompt: string }, +): Promise { + const id = `chat-${options.instance}`; + const frame = { + type: 'cf_agent_use_chat_request', + id, + init: { + method: 'POST', + body: JSON.stringify({ + messages: [{ id: 'msg-1', role: 'user', parts: [{ type: 'text', text: options.prompt }] }], + }), + }, + }; + + return driveAgentSocket( + baseURL, + options.binding, + options.instance, + frame, + reply => reply.type === 'cf_agent_use_chat_response' && reply.id === id && !!reply.done, + 'chat response', + ); +} + +/** Opens a WebSocket to `/agents//`, sends one RPC frame, resolves on the reply. */ +export function callRpc( + baseURL: string, + options: { binding: string; instance: string; method: string; args: unknown[] }, +): Promise { + const id = `rpc-${options.method}`; + const frame = { type: 'rpc', id, method: options.method, args: options.args }; + + return driveAgentSocket( + baseURL, + options.binding, + options.instance, + frame, + reply => reply.type === 'rpc' && reply.id === id && !!reply.done, + `RPC reply to "${options.method}"`, + ); +} diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/ai-streaming.test.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/ai-streaming.test.ts new file mode 100644 index 000000000000..de64bff1dbfd --- /dev/null +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/ai-streaming.test.ts @@ -0,0 +1,60 @@ +import { expect, test } from '@playwright/test'; +import { waitForTransaction } from '@sentry-internal/test-utils'; + +// Drives Workers AI through the real Cloudflare Agents SDK + Vercel AI SDK + `workers-ai-provider` +// stack (the OpenAI-compatible SSE shape, `choices[].delta.content`) from an Agent's `onRequest`. +// Asserts the streaming response text is captured on the gen_ai span — the regression seen in +// production where only input + usage survived. The model output is read from +// `gen_ai.output.messages`, so the streaming instrumentation must emit it alongside the +// deprecated `gen_ai.response.text`. +function assertGenAiStreamingSpan(spans: Array> | undefined): void { + const genAiSpan = (spans ?? []).find(span => span.op === 'gen_ai.chat'); + + expect(genAiSpan).toBeDefined(); + expect(genAiSpan.origin).toBe('auto.ai.cloudflare.workers_ai'); + expect(genAiSpan.data).toEqual( + expect.objectContaining({ + 'sentry.origin': 'auto.ai.cloudflare.workers_ai', + 'gen_ai.operation.name': 'chat', + 'gen_ai.request.model': '@cf/meta/llama-3.1-8b-instruct', + 'gen_ai.response.streaming': true, + 'gen_ai.response.text': 'The capital of France is Paris.', + 'gen_ai.output.messages': JSON.stringify([ + { role: 'assistant', parts: [{ type: 'text', content: 'The capital of France is Paris.' }] }, + ]), + 'gen_ai.usage.input_tokens': 15, + 'gen_ai.usage.output_tokens': 8, + 'gen_ai.usage.total_tokens': 23, + }), + ); +} + +test('captures Workers AI streaming output when driven via an Agent', async ({ request, baseURL }) => { + const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => { + return ( + transactionEvent.transaction === 'GET /agents/my-agent/test' && + (transactionEvent.spans ?? []).some(span => span.op === 'gen_ai.chat') + ); + }); + + const response = await request.get(`${baseURL}/agents/my-agent/test`); + expect(response.ok()).toBe(true); + + const transaction = await transactionPromise; + assertGenAiStreamingSpan(transaction.spans); +}); + +test('captures Workers AI streaming output when driven via an AIChatAgent', async ({ request, baseURL }) => { + const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => { + return ( + transactionEvent.transaction === 'GET /agents/my-chat-agent/test' && + (transactionEvent.spans ?? []).some(span => span.op === 'gen_ai.chat') + ); + }); + + const response = await request.get(`${baseURL}/agents/my-chat-agent/test`); + expect(response.ok()).toBe(true); + + const transaction = await transactionPromise; + assertGenAiStreamingSpan(transaction.spans); +}); diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/callable.test.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/callable.test.ts index d23b9f4c3870..b41347148fb6 100644 --- a/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/callable.test.ts +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/callable.test.ts @@ -1,7 +1,7 @@ import { expect, test } from '@playwright/test'; import { waitForTransaction } from '@sentry-internal/test-utils'; -test('@callable() methods work correctly with Sentry instrumentDurableObjectWithSentry', async ({ page, baseURL }) => { +test('@callable() methods work correctly with Sentry instrumentAgentWithSentry', async ({ page, baseURL }) => { const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => { return ( transactionEvent.transaction === 'GET /agents/my-agent/user-123' && @@ -127,3 +127,63 @@ test('@callable() methods work correctly with Sentry instrumentDurableObjectWith }), ]); }); + +test('does not emit db.query spans for the agents runtime `cf_`-prefixed internal tables', async ({ + page, + baseURL, +}) => { + const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => { + return ( + transactionEvent.transaction === 'GET /agents/my-agent/user-123' && + transactionEvent.contexts?.trace?.parent_span_id !== undefined + ); + }); + + await page.goto(baseURL!); + + await expect(page.getByText('Connected')).toBeVisible(); + await page.getByRole('button', { name: 'Call Agent' }).click(); + await expect(page.getByText('Hello, World!')).toBeVisible(); + + const transaction = await transactionPromise; + + // The agents runtime constantly queries its own `cf_agents_*` / `cf_agent_*` bookkeeping tables. + // These are framework internals and are filtered out by default, so no such span should leak. + const internalTableSpans = (transaction.spans ?? []).filter( + span => span.op === 'db.query' && /\bcf_/.test((span.data?.['db.query.summary'] as string) ?? ''), + ); + + expect(internalTableSpans).toEqual([]); +}); + +test('creates an rpc span named after the @callable() method', async ({ page, baseURL }) => { + const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => { + return ( + transactionEvent.transaction === 'webSocketMessage' && + (transactionEvent.spans ?? []).some(span => span.op === 'rpc' && span.description === 'greet') + ); + }); + + await page.goto(baseURL!); + + await expect(page.getByText('Connected')).toBeVisible(); + await page.getByRole('button', { name: 'Call Agent' }).click(); + await expect(page.getByText('Hello, World!')).toBeVisible(); + + const transaction = await transactionPromise; + + const rpcSpans = (transaction.spans ?? []).filter(span => span.op === 'rpc'); + expect(rpcSpans).toHaveLength(1); + + expect(rpcSpans[0]).toEqual( + expect.objectContaining({ + op: 'rpc', + description: 'greet', + origin: 'auto.faas.cloudflare.agents', + data: expect.objectContaining({ + 'cloudflare.agent.class': 'MyBaseAgent', + 'cloudflare.agent.name': 'user-123', + }), + }), + ); +}); diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-conversation.test.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-conversation.test.ts new file mode 100644 index 000000000000..5d7c6c2fbdd3 --- /dev/null +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-conversation.test.ts @@ -0,0 +1,32 @@ +import { expect, test } from '@playwright/test'; +import { waitForTransaction } from '@sentry-internal/test-utils'; +import { sendChatMessage } from './agent-socket'; + +// In the Agents model one agent instance is one conversation, so the instance name is the +// conversation id the SDK correlates the turn's gen_ai spans with. +const CONVERSATION_ID = 'chat-conv-instance'; + +test('stamps the conversation id on gen_ai spans created inside a chat turn', async ({ baseURL }) => { + const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => { + return ( + transactionEvent.transaction === 'webSocketMessage' && + (transactionEvent.spans ?? []).some(span => span.op === 'gen_ai.chat') + ); + }); + + await sendChatMessage(baseURL!, { + binding: 'my-chat-agent', + instance: CONVERSATION_ID, + prompt: 'What is the capital of France?', + }); + + const transaction = await transactionPromise; + + const genAiSpan = (transaction.spans ?? []).find(span => span.op === 'gen_ai.chat'); + expect(genAiSpan).toBeDefined(); + expect(genAiSpan?.data).toEqual( + expect.objectContaining({ + 'gen_ai.conversation.id': CONVERSATION_ID, + }), + ); +}); diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-rpc.test.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-rpc.test.ts new file mode 100644 index 000000000000..7b3250810c45 --- /dev/null +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/chat-rpc.test.ts @@ -0,0 +1,30 @@ +import { expect, test } from '@playwright/test'; +import { waitForTransaction } from '@sentry-internal/test-utils'; +import { callRpc } from './agent-socket'; + +const AGENT_INSTANCE = 'chat-rpc-instance'; + +test('creates an rpc span for a @callable() invocation on an AIChatAgent', async ({ baseURL }) => { + const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => { + return ( + transactionEvent.transaction === 'webSocketMessage' && + (transactionEvent.spans ?? []).some(span => span.op === 'rpc' && span.description === 'greet') + ); + }); + + await callRpc(baseURL!, { binding: 'my-chat-agent', instance: AGENT_INSTANCE, method: 'greet', args: ['World'] }); + + const transaction = await transactionPromise; + + const rpcSpan = (transaction.spans ?? []).find(span => span.op === 'rpc' && span.description === 'greet'); + expect(rpcSpan).toEqual( + expect.objectContaining({ + op: 'rpc', + description: 'greet', + origin: 'auto.faas.cloudflare.agents', + data: expect.objectContaining({ + 'cloudflare.agent.name': AGENT_INSTANCE, + }), + }), + ); +}); diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker-configuration.d.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker-configuration.d.ts index 7e95ce232f43..693fee686ebd 100644 --- a/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker-configuration.d.ts +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker-configuration.d.ts @@ -4,11 +4,12 @@ interface __BaseEnv_Env { CF_VERSION_METADATA: WorkerVersionMetadata; E2E_TEST_DSN: string; MyAgent: DurableObjectNamespace; + MyChatAgent: DurableObjectNamespace; } declare namespace Cloudflare { interface GlobalProps { mainModule: typeof import('./worker/index'); - durableNamespaces: 'MyAgent'; + durableNamespaces: 'MyAgent' | 'MyChatAgent'; } interface Env extends __BaseEnv_Env {} } diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/index.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/index.ts index a38b6fa1248e..c07ffe75652d 100644 --- a/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/index.ts +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/index.ts @@ -1,7 +1,43 @@ import * as Sentry from '@sentry/cloudflare'; -import { routeAgentRequest, Agent, callable } from 'agents'; +import { instrumentWorkersAiClient } from '@sentry/core'; +import { AIChatAgent } from '@cloudflare/ai-chat'; +import { Agent, callable, routeAgentRequest } from 'agents'; +import { streamText } from 'ai'; +import { createWorkersAI } from 'workers-ai-provider'; +import { MockAi } from './mocks'; -class MyBaseAgent extends Agent { +const MODEL = '@cf/meta/llama-3.1-8b-instruct'; + +const sentryOptions = (env: Env) => ({ + traceLifecycle: 'static' as const, + dsn: env.E2E_TEST_DSN, + tunnel: `http://localhost:3031/`, + tracesSampleRate: 1, + enableRpcTracePropagation: true, + // Keep gen_ai spans embedded in the transaction (instead of streamed as a separate envelope + // container) so they can be asserted on `transaction.spans`. + streamGenAiSpans: false, + durableObjectStorageSpanAllowlist: ['cf_user_key'], +}); + +/** + * In production `env.AI` is auto-instrumented by `@sentry/cloudflare`. There is no real AI + * binding offline, so we instrument the mock binding manually and drive it through the real + * Vercel AI SDK + `workers-ai-provider` stack (the OpenAI-compatible SSE shape). + */ +function streamWorkersAi(): Response { + const ai = instrumentWorkersAiClient(new MockAi(), { recordInputs: true, recordOutputs: true }); + const workersai = createWorkersAI({ binding: ai as unknown as Ai }); + + const result = streamText({ + model: workersai(MODEL), + prompt: 'What is the capital of France?', + }); + + return result.toTextStreamResponse(); +} + +class MyBaseAgent extends Agent { @callable() async greet(name: string): Promise { // User keys — instrumented, spans expected @@ -18,35 +54,41 @@ class MyBaseAgent extends Agent { return `Hello, ${name}!`; } + + async onRequest(): Promise { + return streamWorkersAi(); + } } -export const MyAgent = Sentry.instrumentDurableObjectWithSentry( - (env: Env) => ({ - dsn: env.E2E_TEST_DSN, - tunnel: `http://localhost:3031/`, - tracesSampleRate: 1, - enableRpcTracePropagation: true, - durableObjectStorageSpanAllowlist: ['cf_user_key'], - }), - MyBaseAgent, -); - -export default Sentry.withSentry( - (env: Env) => ({ - dsn: env.E2E_TEST_DSN, - tunnel: `http://localhost:3031/`, - tracesSampleRate: 1, - enableRpcTracePropagation: true, - }), - { - async fetch(request: Request, env: Env): Promise { - const agentResponse = await routeAgentRequest(request, env); - - if (agentResponse) { - return agentResponse; - } - - return new Response(null, { status: 404 }); - }, - } satisfies ExportedHandler, -); +class MyChatAgentBase extends AIChatAgent { + @callable() + async greet(name: string): Promise { + return `Hello, ${name}!`; + } + + async onRequest(): Promise { + return streamWorkersAi(); + } + + async onChatMessage(): Promise { + // The gen_ai turn must run inside `onChatMessage` (not `onRequest`) so it happens while the + // SDK has set the conversation id on the scope for this chat turn — that is what + // `conversationIdIntegration` reads to stamp `gen_ai.conversation.id` onto the span. + return streamWorkersAi(); + } +} + +export const MyAgent = Sentry.instrumentAgentWithSentry(sentryOptions, MyBaseAgent); +export const MyChatAgent = Sentry.instrumentAgentWithSentry(sentryOptions, MyChatAgentBase); + +export default Sentry.withSentry(sentryOptions, { + async fetch(request: Request, env: Env): Promise { + const agentResponse = await routeAgentRequest(request, env); + + if (agentResponse) { + return agentResponse; + } + + return new Response(null, { status: 404 }); + }, +} satisfies ExportedHandler); diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/mocks.ts b/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/mocks.ts new file mode 100644 index 000000000000..4fba65c6a777 --- /dev/null +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/worker/mocks.ts @@ -0,0 +1,44 @@ +import { simulateReadableStream } from 'ai'; + +function createSseStream(events: string[]): ReadableStream { + const encoder = new TextEncoder(); + return simulateReadableStream({ + initialDelayInMs: 0, + chunkDelayInMs: 0, + chunks: events.map(event => encoder.encode(`data: ${event}\n\n`)), + }); +} + +/** + * Minimal mock of the Cloudflare Workers AI binding (`env.AI`) that emits the + * OpenAI-compatible streaming shape (`choices[].delta.content`) — the format models + * routed through `workers-ai-provider` stream when driven by an Agent / `AIChatAgent`. + * + * This is the format `workers-ai-provider` receives from `binding.run(..., { stream: true })` + * for models routed through the OpenAI-compatible endpoint (which the Agents SDK uses). + * The native `{ response }` shape is covered separately. + */ +export class MockAi { + public async run(_model: string, inputs: Record): Promise { + await new Promise(resolve => setTimeout(resolve, 10)); + + if (inputs?.stream === true) { + return createSseStream([ + '{"choices":[{"index":0,"delta":{"content":"The capital "},"finish_reason":null}]}', + '{"choices":[{"index":0,"delta":{"content":"of France "},"finish_reason":null}]}', + '{"choices":[{"index":0,"delta":{"content":"is Paris."},"finish_reason":"stop"}]}', + '{"usage":{"prompt_tokens":15,"completion_tokens":8,"total_tokens":23}}', + '[DONE]', + ]); + } + + return { + response: 'The capital of France is Paris.', + usage: { + prompt_tokens: 15, + completion_tokens: 8, + total_tokens: 23, + }, + }; + } +} diff --git a/dev-packages/e2e-tests/test-applications/cloudflare-agent/wrangler.jsonc b/dev-packages/e2e-tests/test-applications/cloudflare-agent/wrangler.jsonc index de8b5998eac4..0e32ad2bd13c 100644 --- a/dev-packages/e2e-tests/test-applications/cloudflare-agent/wrangler.jsonc +++ b/dev-packages/e2e-tests/test-applications/cloudflare-agent/wrangler.jsonc @@ -23,10 +23,13 @@ "compatibility_flags": ["nodejs_compat"], "durable_objects": { - "bindings": [{ "name": "MyAgent", "class_name": "MyAgent" }], + "bindings": [ + { "name": "MyAgent", "class_name": "MyAgent" }, + { "name": "MyChatAgent", "class_name": "MyChatAgent" }, + ], }, - "migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyAgent"] }], + "migrations": [{ "tag": "v1", "new_sqlite_classes": ["MyAgent", "MyChatAgent"] }], "version_metadata": { "binding": "CF_VERSION_METADATA", diff --git a/packages/cloudflare/src/durableobject.ts b/packages/cloudflare/src/durableobject.ts index 42bfcab834cf..1e18e18ac059 100644 --- a/packages/cloudflare/src/durableobject.ts +++ b/packages/cloudflare/src/durableobject.ts @@ -10,8 +10,237 @@ import { wrapRequestHandler } from './request'; import { instrumentContext } from './utils/instrumentContext'; import { extractRpcMeta } from './utils/rpcMeta'; import { getEffectiveRpcPropagation } from './utils/rpcOptions'; +import { instrumentCloudflareAgent } from './instrumentations/agents'; import { type UncheckedMethod, wrapMethodWithSentry } from './wrapMethodWithSentry'; +/** + * The instrumented context passed between the shared construction helpers. + * + * This is intentionally `any` rather than `ReturnType>`. + * A concrete `DurableObjectState` here forces `tsc` to structurally relate its `storage: SqlStorage` + * graph against the `ExecutionContext | InstrumentedDurableObjectState` parameter of + * `wrapMethodWithSentry`, while the RPC-branded `DurableObject` from `cloudflare:workers` is also in + * scope. That union comparison explodes (1296×1296) and hangs the type build. The original inline + * implementation avoided this only incidentally, because `ctx` reached it as `any` through the Proxy + * `construct` trap. Keeping the shared context `any` preserves that behavior. + */ +// eslint-disable-next-line @typescript-eslint/no-explicit-any +type InstrumentedDurableObjectContext = any; + +/** + * Constructs a Durable Object instance and instruments its built-in handler methods + * (`fetch`, `alarm`, `webSocketMessage`, `webSocketClose`, `webSocketError`). + * + * This is the shared construction path used by both {@link instrumentDurableObjectWithSentry} + * and {@link instrumentAgentWithSentry}. It intentionally does NOT apply the RPC prototype-method + * proxy — callers apply that last via {@link finalizeWithRpcInstrumentation}, after any additional + * per-instance instrumentation has been layered onto the returned object. + * + * @internal + */ +export function constructInstrumentedDurableObject>( + target: new (state: DurableObjectState, env: E) => T, + ctx: DurableObjectState, + env: E, + newTarget: NewableFunction, + optionsCallback: (env: E) => CloudflareOptions, +): { obj: T; options: CloudflareOptions; context: InstrumentedDurableObjectContext } { + setAsyncLocalStorageAsyncContextStrategy(); + const options = getFinalOptions(optionsCallback(env), env); + // See InstrumentedDurableObjectContext — `ctx` is widened to `any` so the concrete + // `DurableObjectState` type never enters the checker's relation graph in this module. + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const context = instrumentContext(ctx as any); + const instrumentedEnv = instrumentEnv(env as Record, options); + + // Pass `newTarget` so that subclasses of the instrumented class (e.g. the wrapper classes + // created by wrangler's local dev tooling or `@cloudflare/vitest-pool-workers`) keep their + // own prototype — otherwise subclass methods disappear and `instanceof` checks break. + const obj = Reflect.construct(target, [context, instrumentedEnv], newTarget) as T; + + instrumentDurableObjectHandlers(obj, options, context); + + return { obj, options, context }; +} + +/** + * Instruments the built-in Durable Object handler methods on a constructed instance. + * + * These are the methods that are available on a Durable Object + * ref: https://developers.cloudflare.com/durable-objects/api/base/ + * - obj.alarm + * - obj.fetch + * - obj.webSocketError + * - obj.webSocketClose + * - obj.webSocketMessage + * + * Any other public methods on the Durable Object instance are RPC calls. + */ +function instrumentDurableObjectHandlers>( + obj: T, + options: CloudflareOptions, + context: InstrumentedDurableObjectContext, +): void { + // Bind each built-in handler to this instance before wrapping. + // See https://github.com/getsentry/sentry-javascript/issues/22328 + if (obj.fetch && typeof obj.fetch === 'function') { + obj.fetch = ensureInstrumented( + obj.fetch.bind(obj), + original => + new Proxy(original, { + apply(target, thisArg, args) { + return wrapRequestHandler({ options, request: args[0], context }, () => { + return Reflect.apply(target, thisArg, args); + }); + }, + }), + ); + } + + if (obj.alarm && typeof obj.alarm === 'function') { + // Alarms are independent invocations, so we start a new trace and link to the previous alarm + obj.alarm = wrapMethodWithSentry( + { + options, + context, + spanName: 'alarm', + spanOp: 'function', + startNewTrace: true, + origin: 'auto.faas.cloudflare.durable_object', + }, + obj.alarm.bind(obj), + ); + } + + if (obj.webSocketMessage && typeof obj.webSocketMessage === 'function') { + obj.webSocketMessage = wrapMethodWithSentry( + { options, context, spanName: 'webSocketMessage', origin: 'auto.faas.cloudflare.durable_object' }, + obj.webSocketMessage.bind(obj), + ); + } + + if (obj.webSocketClose && typeof obj.webSocketClose === 'function') { + obj.webSocketClose = wrapMethodWithSentry( + { options, context, spanName: 'webSocketClose', origin: 'auto.faas.cloudflare.durable_object' }, + obj.webSocketClose.bind(obj), + ); + } + + if (obj.webSocketError && typeof obj.webSocketError === 'function') { + obj.webSocketError = wrapMethodWithSentry( + { options, context, spanName: 'webSocketError', origin: 'auto.faas.cloudflare.durable_object' }, + obj.webSocketError.bind(obj), + (_, error) => + captureException(error, { + mechanism: { + type: 'auto.faas.cloudflare.durable_object_websocket', + handled: false, + }, + }), + ); + } +} + +/** + * Wraps a constructed (and already handler-instrumented) Durable Object instance with the RPC + * prototype-method proxy, when RPC trace propagation is enabled. Returns the object unchanged when + * RPC instrumentation is disabled. + * + * This must be applied last, so that any per-instance instrumentation (own properties such as + * `fetch`, `alarm`, or Agent-specific handlers) is excluded from RPC method tracing. + * + * @internal + */ +export function finalizeWithRpcInstrumentation( + obj: T, + options: CloudflareOptions, + context: InstrumentedDurableObjectContext, +): T { + // Get effective RPC propagation setting (handles deprecation of instrumentPrototypeMethods) + const rpcPropagation = getEffectiveRpcPropagation(options); + + // Skip RPC instrumentation if not enabled + if (!rpcPropagation) { + return obj; + } + + // If `instrumentPrototypeMethods` was passed as an array (deprecated), + // only the listed method names should be instrumented. + // eslint-disable-next-line typescript/no-deprecated + const instrumentPrototypeMethods = Array.isArray(options.instrumentPrototypeMethods) + ? // eslint-disable-next-line typescript/no-deprecated + options.instrumentPrototypeMethods + : undefined; + const allowSet = instrumentPrototypeMethods ? new Set(instrumentPrototypeMethods) : null; + + // When using the deprecated `instrumentPrototypeMethods` option, always create spans. + // When using the new `enableRpcTracePropagation`, only create spans when RPC metadata is present. + const alwaysTrace = options.enableRpcTracePropagation === undefined; + + // Return a Proxy that binds all methods to the original object and creates spans + // for RPC calls that have Sentry trace context propagated. + // Binding is required because frameworks may use private fields (babel WeakMap pattern), + // which fail if `this` is the Proxy instead of the original object. + const methodCache = new Map(); + + return new Proxy(obj, { + get(proxyTarget, prop, receiver) { + const value = Reflect.get(proxyTarget, prop, receiver); + + if (typeof prop !== 'string' || typeof value !== 'function' || prop === 'constructor') { + return value; + } + + const cached = methodCache.get(prop); + + if (cached) { + return cached; + } + + const boundMethod = (value as UncheckedMethod).bind(proxyTarget); + + if ( + prop in Object.prototype || + Object.prototype.hasOwnProperty.call(proxyTarget, prop) || + (allowSet && !allowSet.has(prop)) + ) { + methodCache.set(prop, boundMethod); + + return boundMethod; + } + + // Pre-create the traced version + const tracedMethod = wrapMethodWithSentry( + { options, context, spanName: prop, spanOp: 'rpc', origin: 'auto.faas.cloudflare.durable_object' }, + boundMethod, + undefined, + true, + ); + + // For deprecated `instrumentPrototypeMethods`, always trace. + // For new `enableRpcTracePropagation`, only trace when RPC metadata is present. + if (alwaysTrace) { + methodCache.set(prop, tracedMethod); + + return tracedMethod; + } + + // Wrapper that checks for Sentry RPC metadata at call time + const wrappedMethod = ((...args: unknown[]) => { + const { rpcMeta } = extractRpcMeta(args); + + // If Sentry RPC metadata is present, use the traced version (creates span) + // Otherwise, call the bound method directly (no span) + return rpcMeta ? tracedMethod(...args) : boundMethod(...args); + }) as UncheckedMethod; + + methodCache.set(prop, wrappedMethod); + + return wrappedMethod; + }, + }); +} + /** * Instruments a Durable Object class to capture errors and performance data. * @@ -52,168 +281,79 @@ export function instrumentDurableObjectWithSentry< >(optionsCallback: (env: E) => CloudflareOptions, DurableObjectClass: C): C { return new Proxy(DurableObjectClass, { construct(target, [ctx, env], newTarget) { - setAsyncLocalStorageAsyncContextStrategy(); - const context = instrumentContext(ctx); - const options = getFinalOptions(optionsCallback(env), env); - const instrumentedEnv = instrumentEnv(env, options); - - // Pass `newTarget` so that subclasses of the instrumented class (e.g. the wrapper classes - // created by wrangler's local dev tooling or `@cloudflare/vitest-pool-workers`) keep their - // own prototype — otherwise subclass methods disappear and `instanceof` checks break. - const obj = Reflect.construct(target, [context, instrumentedEnv], newTarget) as T; - - // These are the methods that are available on a Durable Object - // ref: https://developers.cloudflare.com/durable-objects/api/base/ - // obj.alarm - // obj.fetch - // obj.webSocketError - // obj.webSocketClose - // obj.webSocketMessage - - // Any other public methods on the Durable Object instance are RPC calls. - - // Bind each built-in handler to this instance before wrapping. - // See https://github.com/getsentry/sentry-javascript/issues/22328 - if (obj.fetch && typeof obj.fetch === 'function') { - obj.fetch = ensureInstrumented( - obj.fetch.bind(obj), - original => - new Proxy(original, { - apply(target, thisArg, args) { - return wrapRequestHandler({ options, request: args[0], context }, () => { - return Reflect.apply(target, thisArg, args); - }); - }, - }), - ); - } + const { obj, options, context } = constructInstrumentedDurableObject( + target, + ctx, + env, + newTarget, + optionsCallback, + ); - if (obj.alarm && typeof obj.alarm === 'function') { - // Alarms are independent invocations, so we start a new trace and link to the previous alarm - obj.alarm = wrapMethodWithSentry( - { - options, - context, - spanName: 'alarm', - spanOp: 'function', - startNewTrace: true, - origin: 'auto.faas.cloudflare.durable_object', - }, - obj.alarm.bind(obj), - ); - } - - if (obj.webSocketMessage && typeof obj.webSocketMessage === 'function') { - obj.webSocketMessage = wrapMethodWithSentry( - { options, context, spanName: 'webSocketMessage', origin: 'auto.faas.cloudflare.durable_object' }, - obj.webSocketMessage.bind(obj), - ); - } - - if (obj.webSocketClose && typeof obj.webSocketClose === 'function') { - obj.webSocketClose = wrapMethodWithSentry( - { options, context, spanName: 'webSocketClose', origin: 'auto.faas.cloudflare.durable_object' }, - obj.webSocketClose.bind(obj), - ); - } - - if (obj.webSocketError && typeof obj.webSocketError === 'function') { - obj.webSocketError = wrapMethodWithSentry( - { options, context, spanName: 'webSocketError', origin: 'auto.faas.cloudflare.durable_object' }, - obj.webSocketError.bind(obj), - (_, error) => - captureException(error, { - mechanism: { - type: 'auto.faas.cloudflare.durable_object_websocket', - handled: false, - }, - }), - ); - } + return finalizeWithRpcInstrumentation(obj, options, context); + }, + }); +} - // Get effective RPC propagation setting (handles deprecation of instrumentPrototypeMethods) - const rpcPropagation = getEffectiveRpcPropagation(options); +/** + * Instruments a Cloudflare [`agents`](https://www.npmjs.com/package/agents) Agent class with Sentry. + * + * An `Agent` is a Durable Object under the hood, so this applies the same instrumentation as + * {@link instrumentDurableObjectWithSentry} (request transactions, `alarm`, WebSocket handlers, RPC + * trace propagation, SQL spans) and additionally captures Agent-specific telemetry via + * `instrumentCloudflareAgent`: + * + * - **Callable RPC spans** — a span (op `rpc`) for each `@callable()` method invoked over WebSocket. + * - **Breadcrumbs** — for every Agent observability event (`rpc`, `state:update`, `connect`, + * `disconnect`, `schedule:*`, `queue:*`, `workflow:*`, `email:*`, `mcp:*`, ...). + * + * Cloudflare Workers cannot auto-instrument, so the Agent class must be wrapped manually. + * + * @param optionsCallback Function that returns the options for the SDK initialization. + * @param AgentClass The Agent class to instrument. + * @returns The instrumented Agent class. + * + * @example + * ```ts + * import { Agent, callable, routeAgentRequest } from 'agents'; + * import * as Sentry from '@sentry/cloudflare'; + * + * class MyAgentBase extends Agent { + * @callable() + * async greet(name: string): Promise { + * return `Hello, ${name}!`; + * } + * } + * + * export const MyAgent = Sentry.instrumentAgentWithSentry( + * env => ({ + * dsn: env.SENTRY_DSN, + * tracesSampleRate: 1.0, + * enableRpcTracePropagation: true, + * }), + * MyAgentBase, + * ); + * ``` + */ +export function instrumentAgentWithSentry< + E, + T extends DurableObject, + C extends new (state: DurableObjectState, env: E) => T, +>(optionsCallback: (env: E) => CloudflareOptions, AgentClass: C): C { + return new Proxy(AgentClass, { + construct(target, [ctx, env], newTarget) { + const { obj, options, context } = constructInstrumentedDurableObject( + target, + ctx, + env, + newTarget, + optionsCallback, + ); - // Skip RPC instrumentation if not enabled - if (!rpcPropagation) { - return obj; - } + instrumentCloudflareAgent(obj); - // If `instrumentPrototypeMethods` was passed as an array (deprecated), - // only the listed method names should be instrumented. - // eslint-disable-next-line typescript/no-deprecated - const instrumentPrototypeMethods = Array.isArray(options.instrumentPrototypeMethods) - ? // eslint-disable-next-line typescript/no-deprecated - options.instrumentPrototypeMethods - : undefined; - const allowSet = instrumentPrototypeMethods ? new Set(instrumentPrototypeMethods) : null; - - // When using the deprecated `instrumentPrototypeMethods` option, always create spans. - // When using the new `enableRpcTracePropagation`, only create spans when RPC metadata is present. - const alwaysTrace = options.enableRpcTracePropagation === undefined; - - // Return a Proxy that binds all methods to the original object and creates spans - // for RPC calls that have Sentry trace context propagated. - // Binding is required because frameworks may use private fields (babel WeakMap pattern), - // which fail if `this` is the Proxy instead of the original object. - const methodCache = new Map(); - - return new Proxy(obj, { - get(proxyTarget, prop, receiver) { - const value = Reflect.get(proxyTarget, prop, receiver); - - if (typeof prop !== 'string' || typeof value !== 'function' || prop === 'constructor') { - return value; - } - - const cached = methodCache.get(prop); - - if (cached) { - return cached; - } - - const boundMethod = (value as UncheckedMethod).bind(proxyTarget); - - if ( - prop in Object.prototype || - Object.prototype.hasOwnProperty.call(proxyTarget, prop) || - (allowSet && !allowSet.has(prop)) - ) { - methodCache.set(prop, boundMethod); - - return boundMethod; - } - - // Pre-create the traced version - const tracedMethod = wrapMethodWithSentry( - { options, context, spanName: prop, spanOp: 'rpc', origin: 'auto.faas.cloudflare.durable_object' }, - boundMethod, - undefined, - true, - ); - - // For deprecated `instrumentPrototypeMethods`, always trace. - // For new `enableRpcTracePropagation`, only trace when RPC metadata is present. - if (alwaysTrace) { - methodCache.set(prop, tracedMethod); - - return tracedMethod; - } - - // Wrapper that checks for Sentry RPC metadata at call time - const wrappedMethod = ((...args: unknown[]) => { - const { rpcMeta } = extractRpcMeta(args); - - // If Sentry RPC metadata is present, use the traced version (creates span) - // Otherwise, call the bound method directly (no span) - return rpcMeta ? tracedMethod(...args) : boundMethod(...args); - }) as UncheckedMethod; - - methodCache.set(prop, wrappedMethod); - - return wrappedMethod; - }, - }); + // Apply RPC prototype-method instrumentation last, so the Agent-specific own-property + // handlers we just installed are excluded from RPC method tracing. + return finalizeWithRpcInstrumentation(obj, options, context); }, }); } diff --git a/packages/cloudflare/src/index.ts b/packages/cloudflare/src/index.ts index 127c61eaeb39..4a718f10f5ce 100644 --- a/packages/cloudflare/src/index.ts +++ b/packages/cloudflare/src/index.ts @@ -120,7 +120,7 @@ export { export { withSentry } from './withSentry'; export { defineCloudflareOptions } from './defineCloudflareOptions'; -export { instrumentDurableObjectWithSentry } from './durableobject'; +export { instrumentAgentWithSentry, instrumentDurableObjectWithSentry } from './durableobject'; export { sentryPagesPlugin } from './pages-plugin'; export { wrapRequestHandler } from './request'; diff --git a/packages/cloudflare/src/instrumentations/agents/index.ts b/packages/cloudflare/src/instrumentations/agents/index.ts new file mode 100644 index 000000000000..89546446d93b --- /dev/null +++ b/packages/cloudflare/src/instrumentations/agents/index.ts @@ -0,0 +1,34 @@ +import { instrumentAgentCallableRpc } from './instrumentAgentCallableRpc'; +import { instrumentChatAgentConversation } from './instrumentChatAgentConversation'; +import type { AgentInternals } from './types'; + +/** + * Instruments an instance of a Cloudflare [`agents`](https://www.npmjs.com/package/agents) `Agent` + * with Sentry, adding telemetry that is specific to the Agent's runtime behavior: + * + * - **Callable RPC spans** — a span (op `rpc`) for each `@callable()` method invoked over WebSocket. + * - **Conversation correlation** — sets the conversation id on the scope for each unit of agent + * work — chat turn or callable RPC call — so `gen_ai` spans created within it are correlated, for + * chat and plain agents alike. Defaults to the instance `name` and is rotated when the chat is + * cleared (`cf_agent_chat_clear`). + * + * It only hooks the `agents` package internals and uses Sentry's tracing primitives. On Cloudflare + * Workers, prefer `instrumentAgentWithSentry`, which additionally instruments the Durable Object + * handlers and initializes the SDK; that function calls this one internally. + * + * The hooks replace own-properties on the instance and defer to the original implementation, so + * they compose with any other instrumentation applied to the same instance. + * + * @param agent The `agents` `Agent` instance to instrument. + * @returns The same instance, instrumented. + * + * @internal Use `instrumentAgentWithSentry` instead — this is called internally by it. + */ +export function instrumentCloudflareAgent(agent: T): T { + const internals = agent as T & AgentInternals; + + instrumentAgentCallableRpc(internals); + instrumentChatAgentConversation(internals); + + return agent; +} diff --git a/packages/cloudflare/src/instrumentations/agents/instrumentAgentCallableRpc.ts b/packages/cloudflare/src/instrumentations/agents/instrumentAgentCallableRpc.ts new file mode 100644 index 000000000000..a0a214e5f2c9 --- /dev/null +++ b/packages/cloudflare/src/instrumentations/agents/instrumentAgentCallableRpc.ts @@ -0,0 +1,69 @@ +import { debug, SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startSpan } from '@sentry/core'; +import { DEBUG_BUILD } from '../../debug-build'; +import { AGENT_SPAN_ORIGIN, type AgentInternals, getAgentAttributes, setAgentConversationId } from './types'; + +/** + * Wraps the Agent's `onMessage` handler to create a span for each `@callable()` RPC invocation. + * RPC requests arrive as WebSocket messages, so this span nests under the active transaction for + * the WebSocket message (on Cloudflare, the instrumented Durable Object `webSocketMessage` hook). + * + * Also sets the conversation id on the scope for the duration of the call: callable methods are the + * unit of work for plain (non-chat) agents, which run LLM calls just like chat turns do. + */ +export function instrumentAgentCallableRpc(obj: AgentInternals): void { + const original = obj.onMessage; + if (typeof original !== 'function') { + DEBUG_BUILD && debug.log('[Sentry] Agent `onMessage` not found — callable RPC span instrumentation skipped.'); + return; + } + + obj.onMessage = new Proxy(original, { + apply(target, thisArg: AgentInternals, args: unknown[]): unknown { + const method = extractCallableMethod(args[1]); + + if (!method) { + return Reflect.apply(target, thisArg, args); + } + + return startSpan( + { + name: method, + attributes: { + [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'rpc', + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: AGENT_SPAN_ORIGIN, + ...getAgentAttributes(thisArg), + }, + }, + () => { + setAgentConversationId(thisArg); + return Reflect.apply(target, thisArg, args); + }, + ); + }, + }); +} + +/** Extracts the RPC method name from a WebSocket message, mirroring the SDK's `isRPCRequest`. */ +function extractCallableMethod(message: unknown): string | undefined { + const text = + typeof message === 'string' + ? message + : message instanceof ArrayBuffer + ? new TextDecoder().decode(message) + : undefined; + + if (!text) { + return undefined; + } + + try { + const parsed = JSON.parse(text) as { type?: unknown; method?: unknown; args?: unknown }; + if (parsed.type === 'rpc' && typeof parsed.method === 'string' && Array.isArray(parsed.args)) { + return parsed.method; + } + } catch { + // Not JSON, or not an RPC request — no span. + } + + return undefined; +} diff --git a/packages/cloudflare/src/instrumentations/agents/instrumentChatAgentConversation.ts b/packages/cloudflare/src/instrumentations/agents/instrumentChatAgentConversation.ts new file mode 100644 index 000000000000..7ea6613ece19 --- /dev/null +++ b/packages/cloudflare/src/instrumentations/agents/instrumentChatAgentConversation.ts @@ -0,0 +1,29 @@ +import { type AgentInternals, setAgentConversationId } from './types'; + +/** + * For chat agents (`AIChatAgent` from `@cloudflare/ai-chat`), sets the conversation id on the active + * scope for the duration of a chat turn. In the Agents model one agent instance is one conversation, + * so the instance `name` is the natural conversation id. + * + * The id itself is not attached to spans here — the SDK's `conversationIdIntegration` reads it off + * the scope at `spanStart` and stamps `gen_ai.conversation.id` onto the AI spans created inside the + * turn (e.g. by the Workers AI instrumentation), which correlates a turn's model and tool calls. + * + * Plain (non-chat) `Agent`s do not define `onChatMessage`, so they are skipped here — their unit + * of work is the callable RPC method, where `instrumentAgentCallableRpc` sets the conversation id + * instead. + */ +export function instrumentChatAgentConversation(obj: AgentInternals): void { + const original = obj.onChatMessage; + if (typeof original !== 'function') { + return; + } + + obj.onChatMessage = new Proxy(original, { + apply(target, thisArg: AgentInternals, args: unknown[]): unknown { + setAgentConversationId(thisArg); + + return Reflect.apply(target, thisArg, args); + }, + }); +} diff --git a/packages/cloudflare/src/instrumentations/agents/types.ts b/packages/cloudflare/src/instrumentations/agents/types.ts new file mode 100644 index 000000000000..7f84405670f1 --- /dev/null +++ b/packages/cloudflare/src/instrumentations/agents/types.ts @@ -0,0 +1,62 @@ +import { getCurrentScope } from '@sentry/core'; + +export const AGENT_SPAN_ORIGIN = 'auto.faas.cloudflare.agents'; +export const AGENT_CLASS_ATTRIBUTE = 'cloudflare.agent.class'; +export const AGENT_NAME_ATTRIBUTE = 'cloudflare.agent.name'; + +/** + * The subset of the `agents` `Agent` instance internals that we instrument. These are runtime + * implementation details of the `agents` package (v0.13.x) rather than part of its public type + * surface, so every access is guarded and wrapping degrades gracefully if a name changes upstream. + */ +export interface AgentInternals { + /** Central choke-point through which all `agents:*` observability events are published. */ + _emit?: (type: string, payload?: Record) => void; + /** WebSocket message handler; dispatches `@callable()` RPC requests. */ + onMessage?: (...args: unknown[]) => unknown; + /** + * Chat-turn handler. Only `AIChatAgent` (from `@cloudflare/ai-chat`) defines this; a plain `Agent` + * does not, so its presence discriminates a chat agent. + */ + onChatMessage?: (...args: unknown[]) => unknown; + /** The user's Agent class (used by the SDK for the observability event `agent` field). */ + _ParentClass?: { name?: string }; + /** The Agent instance name, which in the Agents model identifies the conversation/thread. */ + name?: string; +} + +/** Reads best-effort agent identity attributes from the instance, tolerating missing internals. */ +export function getAgentAttributes(instance: AgentInternals): Record { + const attributes: Record = {}; + + const agentClass = instance._ParentClass?.name; + if (typeof agentClass === 'string' && agentClass) { + attributes[AGENT_CLASS_ATTRIBUTE] = agentClass; + } + + const agentName = instance.name; + if (typeof agentName === 'string' && agentName) { + attributes[AGENT_NAME_ATTRIBUTE] = agentName; + } + + return attributes; +} + +/** + * Sets the agent instance's conversation id on the current scope for the duration of the + * surrounding unit of work (chat turn, callable RPC call). In the Agents model one instance is one + * conversation, so the instance `name` is the natural conversation id — for chat and plain agents + * alike, since plain agents run LLM calls too (e.g. inside `@callable()` methods). + * + * `conversationIdIntegration` reads the id off the scope at `spanStart` and stamps + * `gen_ai.conversation.id` onto AI spans created within the unit of work, correlating its model + * and tool calls. Callers run inside a per-event forked scope (`wrapMethodWithSentry`), so the id + * does not leak into unrelated events. + */ +export function setAgentConversationId(instance: AgentInternals): void { + const conversationId = instance.name; + + if (typeof conversationId === 'string' && conversationId) { + getCurrentScope().setConversationId(conversationId); + } +} diff --git a/packages/cloudflare/test/agents.test.ts b/packages/cloudflare/test/agents.test.ts new file mode 100644 index 000000000000..93c6abda81f0 --- /dev/null +++ b/packages/cloudflare/test/agents.test.ts @@ -0,0 +1,69 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { instrumentAgentWithSentry } from '../src'; +import { getInstrumented } from '../src/instrument'; + +describe('instrumentAgentWithSentry', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('instruments the built-in Durable Object handlers (Agent extends DurableObject)', () => { + const testClass = class { + fetch() {} + alarm() {} + webSocketMessage() {} + webSocketClose() {} + webSocketError() {} + }; + + const instrumented = instrumentAgentWithSentry(vi.fn().mockReturnValue({}), testClass as any); + const obj = Reflect.construct(instrumented, []); + + for (const methodName of ['fetch', 'alarm', 'webSocketMessage', 'webSocketClose', 'webSocketError']) { + expect(getInstrumented((obj as any)[methodName]), `Method ${methodName} is instrumented`).toBeTruthy(); + } + }); + + it('wraps the Agent-specific handlers as own-properties on the constructed instance', () => { + const testClass = class { + fetch() {} + onMessage() {} + }; + const proto = testClass.prototype as any; + + const instrumented = instrumentAgentWithSentry(vi.fn().mockReturnValue({}), testClass as any); + const obj = Reflect.construct(instrumented, []) as any; + + // `instrumentCloudflareAgent` replaces each handler with a wrapper stored as an own-property, + // so the instance's copy is a distinct function from the untouched prototype method. + for (const methodName of ['onMessage']) { + expect(Object.prototype.hasOwnProperty.call(obj, methodName), `${methodName} is an own-property`).toBe(true); + expect(obj[methodName], `${methodName} differs from the prototype original`).not.toBe(proto[methodName]); + } + }); + + it('keeps RPC methods on the prototype callable while wrapping Agent handlers as own-properties', () => { + const testClass = class { + fetch() {} + onMessage() {} + onChatMessage() {} + rpcMethod() { + return 'rpc'; + } + }; + + const instrumented = instrumentAgentWithSentry( + vi.fn().mockReturnValue({ enableRpcTracePropagation: true }), + testClass as any, + ); + const obj = Reflect.construct(instrumented, []); + + // Agent-specific handlers become own properties, so they are excluded from RPC method tracing. + expect(Object.prototype.hasOwnProperty.call(obj, 'onMessage')).toBe(true); + expect(Object.prototype.hasOwnProperty.call(obj, 'onChatMessage')).toBe(true); + + // RPC methods remain on the prototype and still work through the proxy. + expect(Object.prototype.hasOwnProperty.call(obj, 'rpcMethod')).toBe(false); + expect((obj as any).rpcMethod()).toBe('rpc'); + }); +}); diff --git a/packages/cloudflare/test/instrumentCloudflareAgent.test.ts b/packages/cloudflare/test/instrumentCloudflareAgent.test.ts new file mode 100644 index 000000000000..04fe3fa7500e --- /dev/null +++ b/packages/cloudflare/test/instrumentCloudflareAgent.test.ts @@ -0,0 +1,161 @@ +import type { Event } from '@sentry/core'; +import { getCurrentScope, setCurrentClient } from '@sentry/core'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { setAsyncLocalStorageAsyncContextStrategy } from '../src/async'; +import { CloudflareClient, type CloudflareClientOptions } from '../src/client'; +import { instrumentCloudflareAgent } from '../src/instrumentations/agents'; +import { resetSdk } from './testUtils'; + +const dsn = 'https://123@sentry.io/42'; + +/** Minimal stand-in for an `agents` Agent instance exposing the internals we hook. */ +function createFakeAgent(overrides: Record = {}): Record { + return { + _ParentClass: { name: 'MyAgent' }, + name: 'instance-1', + messages: [] as unknown[], + onMessage(this: any, _connection: unknown, message: unknown) { + this.messages.push(message); + return 'handled'; + }, + ...overrides, + }; +} + +describe('instrumentCloudflareAgent', () => { + let transactions: Event[]; + let client: CloudflareClient; + + beforeEach(() => { + resetSdk(); + setAsyncLocalStorageAsyncContextStrategy(); + + transactions = []; + + const options: CloudflareClientOptions = { + dsn, + tracesSampleRate: 1, + traceLifecycle: 'static', + stackParser: () => [], + integrations: [], + transport: () => ({ + send: vi.fn().mockResolvedValue({}), + flush: vi.fn().mockResolvedValue(true), + }), + beforeSendTransaction: event => { + transactions.push(event); + return event; + }, + }; + + client = new CloudflareClient(options); + setCurrentClient(client); + client.init(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('returns the same instance', () => { + const agent = createFakeAgent(); + expect(instrumentCloudflareAgent(agent)).toBe(agent); + }); + + it('does not throw when the Agent internals are missing', () => { + const agent = { name: 'nope' } as Record; + expect(() => instrumentCloudflareAgent(agent)).not.toThrow(); + }); + + describe('onMessage → callable RPC spans', () => { + it('creates an rpc span named after the method for RPC messages', async () => { + const agent = createFakeAgent(); + instrumentCloudflareAgent(agent); + + const result = agent.onMessage({}, JSON.stringify({ type: 'rpc', id: '1', method: 'greet', args: ['World'] })); + + expect(result).toBe('handled'); + expect(agent.messages).toHaveLength(1); + + await client.flush(); + + expect(transactions).toHaveLength(1); + expect(transactions[0]?.transaction).toBe('greet'); + expect(transactions[0]?.contexts?.trace).toEqual( + expect.objectContaining({ + op: 'rpc', + origin: 'auto.faas.cloudflare.agents', + data: expect.objectContaining({ + 'cloudflare.agent.class': 'MyAgent', + 'cloudflare.agent.name': 'instance-1', + }), + }), + ); + }); + + it('does not create a span for non-RPC messages', async () => { + const agent = createFakeAgent(); + instrumentCloudflareAgent(agent); + + agent.onMessage({}, JSON.stringify({ type: 'cf_agent_state', state: {} })); + agent.onMessage({}, 'not json'); + + await client.flush(); + + expect(agent.messages).toHaveLength(2); + expect(transactions).toHaveLength(0); + }); + + it('does not set the conversation id for non-RPC messages', () => { + const agent = createFakeAgent({ + onMessage(this: any, _connection: unknown, message: unknown) { + this.seenConversationId = getCurrentScope().getScopeData().conversationId; + this.messages.push(message); + return 'handled'; + }, + }); + instrumentCloudflareAgent(agent); + + agent.onMessage({}, JSON.stringify({ type: 'cf_agent_state', state: {} })); + + expect(agent.seenConversationId).toBeUndefined(); + }); + }); + + describe('conversation id', () => { + it('sets the conversation id from the instance name during a chat turn', () => { + const agent = createFakeAgent({ + name: 'thread-abc', + onChatMessage(this: any) { + // Capture what the scope sees while the turn is running. + this.seenConversationId = getCurrentScope().getScopeData().conversationId; + return 'response'; + }, + }); + instrumentCloudflareAgent(agent); + + const result = agent.onChatMessage(() => {}, {}); + + expect(result).toBe('response'); + expect(agent.seenConversationId).toBe('thread-abc'); + }); + + it('sets the conversation id during callable RPC execution on plain (non-chat) agents', () => { + const agent = createFakeAgent({ + onMessage(this: any, _connection: unknown, message: unknown) { + // Capture what the scope sees while the RPC method is running. + this.seenConversationId = getCurrentScope().getScopeData().conversationId; + this.messages.push(message); + return 'handled'; + }, + }); + instrumentCloudflareAgent(agent); + + // A plain Agent has no `onChatMessage`; the RPC call is its unit of work. + agent.onMessage({}, JSON.stringify({ type: 'rpc', id: '1', method: 'greet', args: [] })); + + expect(agent.seenConversationId).toBe('instance-1'); + expect('onChatMessage' in agent).toBe(false); + }); + }); +});