Skip to content

test(e2e): Port cloudflare-agent to span streaming #23862

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
JPeer264 wants to merge 1 commit into from
Draft

test(e2e): Port cloudflare-agent to span streaming #23862

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 15 additions & 11 deletions dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/ai-streaming.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import { expect, test } from '@playwright/test';
import { getSpanOp, waitForStreamedSpan, waitForTransaction } from '@sentry-internal/test-utils';
import { getSpanOp, waitForStreamedSpan } from '@sentry-internal/test-utils';
import type { SerializedStreamedSpan } from '@sentry/core';

// Drives Workers AI through the real Cloudflare Agents SDK + Vercel AI SDK + `workers-ai-provider`
Expand All @@ -26,32 +26,36 @@ function assertGenAiStreamingSpan(span: SerializedStreamedSpan): void {
expect(span.attributes['gen_ai.conversation.id']?.value).toMatch(/^[0-9a-f]{32}$/);
}

// With span streaming, URL-sourced `http.server` spans are named by method only, so the request
// segment is identified by its `url.path` attribute.
function isRequestSegment(span: SerializedStreamedSpan, urlPath: string): boolean {
return getSpanOp(span) === 'http.server' && span.is_segment && span.attributes['url.path']?.value === urlPath;
}

test('captures Workers AI streaming output when driven via an Agent', async ({ request, baseURL }) => {
const spanPromise = waitForStreamedSpan('cloudflare-agent', span => getSpanOp(span) === 'gen_ai.chat');
const transactionPromise = waitForTransaction(
'cloudflare-agent',
transactionEvent => transactionEvent.transaction === 'GET /agents/my-agent/test',
const requestSpanPromise = waitForStreamedSpan('cloudflare-agent', span =>
isRequestSegment(span, '/agents/my-agent/test'),
);

const response = await request.get(`${baseURL}/agents/my-agent/test`);
expect(response.ok()).toBe(true);

const [span, transaction] = await Promise.all([spanPromise, transactionPromise]);
expect(span.trace_id).toBe(transaction.contexts?.trace?.trace_id);
const [span, requestSpan] = await Promise.all([spanPromise, requestSpanPromise]);
expect(span.trace_id).toBe(requestSpan.trace_id);
assertGenAiStreamingSpan(span);
});

test('captures Workers AI streaming output when driven via an AIChatAgent', async ({ request, baseURL }) => {
const spanPromise = waitForStreamedSpan('cloudflare-agent', span => getSpanOp(span) === 'gen_ai.chat');
const transactionPromise = waitForTransaction(
'cloudflare-agent',
transactionEvent => transactionEvent.transaction === 'GET /agents/my-chat-agent/test',
const requestSpanPromise = waitForStreamedSpan('cloudflare-agent', span =>
isRequestSegment(span, '/agents/my-chat-agent/test'),
);

const response = await request.get(`${baseURL}/agents/my-chat-agent/test`);
expect(response.ok()).toBe(true);

const [span, transaction] = await Promise.all([spanPromise, transactionPromise]);
expect(span.trace_id).toBe(transaction.contexts?.trace?.trace_id);
const [span, requestSpan] = await Promise.all([spanPromise, requestSpanPromise]);
expect(span.trace_id).toBe(requestSpan.trace_id);
assertGenAiStreamingSpan(span);
});
226 changes: 101 additions & 125 deletions dev-packages/e2e-tests/test-applications/cloudflare-agent/tests/callable.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,22 +1,37 @@
import { expect, test } from '@playwright/test';
import { waitForTransaction } from '@sentry-internal/test-utils';
import { collectStreamedSpans, getSpanOp, waitForStreamedSpan } from '@sentry-internal/test-utils';
import type { SerializedStreamedSpan } from '@sentry/core';

// The agent request segment is the Durable Object's `http.server` span. It has a parent because
// the worker propagates its trace over the RPC binding; the worker's own segment for the same URL
// does not. With span streaming the name is the method only, so the segment is picked by `url.path`.
function isAgentRequestSegment(span: SerializedStreamedSpan): boolean {
return (
getSpanOp(span) === 'http.server' &&
span.is_segment &&
span.attributes['url.path']?.value === '/agents/my-agent/user-123' &&
span.parent_span_id !== undefined
);
}

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' &&
transactionEvent.contexts?.trace?.parent_span_id !== undefined
);
});
function isGreetRpcSpan(span: SerializedStreamedSpan): boolean {
return getSpanOp(span) === 'rpc' && span.name === 'greet';
}

function hasWebSocketMessageSegment(spans: SerializedStreamedSpan[], traceId: string): boolean {
return spans.some(span => span.is_segment && span.name === 'webSocketMessage' && span.trace_id === traceId);
}

// The greet() call goes over the websocket, so its storage spans land in a webSocketMessage
// transaction. Filter for the one carrying our put span — control messages produce their own
// webSocketMessage transactions without storage spans.
const storageTransactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => {
return (
transactionEvent.transaction === 'webSocketMessage' &&
(transactionEvent.spans ?? []).some(span => span.description === 'durable_object_storage_put')
);
test('@callable() methods work correctly with Sentry instrumentAgentWithSentry', async ({ page, baseURL }) => {
const requestSpanPromise = waitForStreamedSpan('cloudflare-agent', isAgentRequestSegment);

// The greet() call goes over the websocket, so its storage spans are children of the `greet` rpc
// span inside a webSocketMessage segment. Control messages produce their own webSocketMessage
// segments without storage spans. Streamed children arrive before their segment, so collect until
// the segment that closes the trace of a `greet` call has arrived.
const storageSpansPromise = collectStreamedSpans('cloudflare-agent', spans => {
const rpcSpan = spans.find(isGreetRpcSpan);
return !!rpcSpan && hasWebSocketMessageSegment(spans, rpcSpan.trace_id);
});

await page.goto(baseURL!);
Expand All @@ -25,109 +40,77 @@ test('@callable() methods work correctly with Sentry instrumentAgentWithSentry',
await page.getByRole('button', { name: 'Call Agent' }).click();
await expect(page.getByText('Hello, World!')).toBeVisible();

const transaction = await transactionPromise;

expect(transaction).toEqual({
contexts: {
trace: {
parent_span_id: expect.stringMatching(/[a-f0-9]{16}/),
span_id: expect.stringMatching(/[a-f0-9]{16}/),
trace_id: expect.stringMatching(/[a-f0-9]{32}/),
data: expect.any(Object),
op: 'http.server',
status: 'ok',
origin: 'auto.http.cloudflare',
},
cloud_resource: { 'cloud.provider': 'cloudflare' },
culture: { timezone: expect.any(String) },
runtime: { name: 'cloudflare' },
},
spans: [],
const requestSpan = await requestSpanPromise;

expect(requestSpan).toEqual({
trace_id: expect.stringMatching(/[a-f0-9]{32}/),
span_id: expect.stringMatching(/[a-f0-9]{16}/),
parent_span_id: expect.stringMatching(/[a-f0-9]{16}/),
name: 'GET',
start_timestamp: expect.any(Number),
timestamp: expect.any(Number),
transaction: 'GET /agents/my-agent/user-123',
type: 'transaction',
request: {
cookies: {},
headers: expect.any(Object),
method: 'GET',
url: expect.stringContaining('/agents/my-agent/user-123'),
query_string: expect.any(String),
},
transaction_info: { source: 'url' },
platform: 'javascript',
event_id: expect.stringMatching(/[a-f0-9]{32}/),
environment: expect.any(String),
user: {
ip_address: '127.0.0.1',
},
release: expect.any(String),
sdk: {
integrations: expect.any(Array),
name: 'sentry.javascript.cloudflare',
version: expect.any(String),
packages: expect.any(Array),
},
end_timestamp: expect.any(Number),
status: 'ok',
is_segment: true,
attributes: expect.objectContaining({
'sentry.op': { value: 'http.server', type: 'string' },
'sentry.origin': { value: 'auto.http.cloudflare', type: 'string' },
'sentry.segment.name.source': { value: 'url', type: 'string' },
'http.request.method': { value: 'GET', type: 'string' },
'url.path': { value: '/agents/my-agent/user-123', type: 'string' },
'sentry.environment': { value: expect.any(String), type: 'string' },
}),
});

// greet() touches 6 storage keys: 2 user ops + 3 framework-internal keys (cf_, __ps_, /) that
// must be filtered + 1 allowlisted cf_ key. Spans carry no key attribute, so filtering can only
// be verified by count — exactly these 3 storage spans (in execution order) should survive, and
// any framework-internal span leaking through shows up as an extra entry here.
const storageTransaction = await storageTransactionPromise;

const storageSpans = (storageTransaction.spans ?? []).filter(
span => span.origin === 'auto.db.cloudflare.durable_object',
);
const spans = await storageSpansPromise;
const rpcSpan = spans.find(isGreetRpcSpan)!;

const storageSpans = spans
.filter(
span =>
span.parent_span_id === rpcSpan.span_id &&
span.attributes['sentry.origin']?.value === 'auto.db.cloudflare.durable_object',
)
.sort((a, b) => a.start_timestamp - b.start_timestamp);

const storageSpanAttributes = (operation: string) =>
expect.objectContaining({
'db.operation.name': { value: operation, type: 'string' },
'db.system.name': { value: 'cloudflare.durable_object.storage', type: 'string' },
'sentry.op': { value: 'db', type: 'string' },
'sentry.origin': { value: 'auto.db.cloudflare.durable_object', type: 'string' },
});

expect(storageSpans).toEqual([
expect.objectContaining({
data: {
'db.operation.name': 'put',
'db.system.name': 'cloudflare.durable_object.storage',
'sentry.op': 'db',
'sentry.origin': 'auto.db.cloudflare.durable_object',
},
description: 'durable_object_storage_put',
op: 'db',
origin: 'auto.db.cloudflare.durable_object',
parent_span_id: expect.stringMatching(/[a-f0-9]{16}/),
name: 'durable_object_storage_put',
attributes: storageSpanAttributes('put'),
parent_span_id: rpcSpan.span_id,
span_id: expect.stringMatching(/[a-f0-9]{16}/),
start_timestamp: expect.any(Number),
timestamp: expect.any(Number),
trace_id: expect.stringMatching(/[a-f0-9]{32}/),
end_timestamp: expect.any(Number),
trace_id: rpcSpan.trace_id,
}),
expect.objectContaining({
data: {
'db.operation.name': 'get',
'db.system.name': 'cloudflare.durable_object.storage',
'sentry.op': 'db',
'sentry.origin': 'auto.db.cloudflare.durable_object',
},
description: 'durable_object_storage_get',
op: 'db',
origin: 'auto.db.cloudflare.durable_object',
parent_span_id: expect.stringMatching(/[a-f0-9]{16}/),
name: 'durable_object_storage_get',
attributes: storageSpanAttributes('get'),
parent_span_id: rpcSpan.span_id,
span_id: expect.stringMatching(/[a-f0-9]{16}/),
start_timestamp: expect.any(Number),
timestamp: expect.any(Number),
trace_id: expect.stringMatching(/[a-f0-9]{32}/),
end_timestamp: expect.any(Number),
trace_id: rpcSpan.trace_id,
}),
expect.objectContaining({
data: {
'db.operation.name': 'get',
'db.system.name': 'cloudflare.durable_object.storage',
'sentry.op': 'db',
'sentry.origin': 'auto.db.cloudflare.durable_object',
},
description: 'durable_object_storage_get',
op: 'db',
origin: 'auto.db.cloudflare.durable_object',
parent_span_id: expect.stringMatching(/[a-f0-9]{16}/),
name: 'durable_object_storage_get',
attributes: storageSpanAttributes('get'),
parent_span_id: rpcSpan.span_id,
span_id: expect.stringMatching(/[a-f0-9]{16}/),
start_timestamp: expect.any(Number),
timestamp: expect.any(Number),
trace_id: expect.stringMatching(/[a-f0-9]{32}/),
end_timestamp: expect.any(Number),
trace_id: rpcSpan.trace_id,
}),
]);
});
Expand All @@ -136,36 +119,33 @@ test('does not emit db.query spans for the agents runtime `cf_`-prefixed interna
page,
baseURL,
}) => {
const transactionPromise = waitForTransaction('cloudflare-agent', transactionEvent => {
return (
transactionEvent.transaction === 'GET /agents/my-agent/user-123' &&
transactionEvent.contexts?.trace?.parent_span_id !== undefined
);
});
const spansPromise = collectStreamedSpans('cloudflare-agent', spans => spans.some(isAgentRequestSegment));

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 spans = await spansPromise;
const requestSpan = spans.find(isAgentRequestSegment)!;

// 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) ?? ''),
const internalTableSpans = spans.filter(
span =>
span.trace_id === requestSpan.trace_id &&
getSpanOp(span) === 'db.query' &&
/\bcf_/.test(String(span.attributes['db.query.summary']?.value ?? '')),
);

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')
);
const spansPromise = collectStreamedSpans('cloudflare-agent', spans => {
const rpcSpan = spans.find(isGreetRpcSpan);
return !!rpcSpan && hasWebSocketMessageSegment(spans, rpcSpan.trace_id);
});

await page.goto(baseURL!);
Expand All @@ -174,19 +154,15 @@ test('creates an rpc span named after the @callable() method', async ({ page, ba
await page.getByRole('button', { name: 'Call Agent' }).click();
await expect(page.getByText('Hello, World!')).toBeVisible();

const transaction = await transactionPromise;
const spans = await spansPromise;
const rpcSpan = spans.find(isGreetRpcSpan)!;

const rpcSpans = (transaction.spans ?? []).filter(span => span.op === 'rpc');
const rpcSpans = spans.filter(span => getSpanOp(span) === 'rpc' && span.trace_id === rpcSpan.trace_id);
expect(rpcSpans).toHaveLength(1);

expect(rpcSpans[0]).toEqual(
expect.objectContaining({
op: 'rpc',
description: 'greet',
origin: 'auto.faas.cloudflare.agents',
data: expect.objectContaining({
'gen_ai.agent.name': 'MyBaseAgent',
}),
}),
);
expect(rpcSpan.attributes).toMatchObject({
'sentry.op': { value: 'rpc', type: 'string' },
'sentry.origin': { value: 'auto.faas.cloudflare.agents', type: 'string' },
'gen_ai.agent.name': { value: 'MyBaseAgent', type: 'string' },
});
});
Loading
Loading