From 22971c517738c0782074180962b755ee09529682 Mon Sep 17 00:00:00 2001 From: jariy17 Date: Fri, 14 Aug 2026 20:06:01 +0000 Subject: [PATCH 1/2] refactor(eval): address on-demand review feedback --- src/core/eval.tsx | 31 ++-- src/core/evalOnDemand.test.ts | 138 ++++++++++++++++++ src/errors/errors.tsx | 14 ++ src/errors/index.tsx | 2 + .../eval/ondemand/ondemand.fixture.test.tsx | 15 -- src/handlers/eval/ondemand/ondemand.test.tsx | 73 ++++----- src/handlers/eval/types.tsx | 27 +--- 7 files changed, 205 insertions(+), 95 deletions(-) create mode 100644 src/core/evalOnDemand.test.ts diff --git a/src/core/eval.tsx b/src/core/eval.tsx index ba9c0a2a7..aef88d11d 100644 --- a/src/core/eval.tsx +++ b/src/core/eval.tsx @@ -84,10 +84,12 @@ import { Transform } from "node:stream"; import { setTimeout as sleep } from "node:timers/promises"; import { AgentCoreCLIError, + CloudWatchQueryError, ERROR_SOURCE, FileWriteError, InputValidationError, NetworkingError, + ResourceNotFoundError, } from "../errors"; import type { BatchEvaluationDetail, @@ -426,8 +428,6 @@ export class EvalClient implements CoreEvalClient { const logGroupName = runtimeLogGroup(runtimeId, qualifier); const serviceName = runtimeServiceName(runtimeName, qualifier); - // CloudWatch Insights takes epoch seconds. Discovery defaults to now-7d when - // no explicit window is given (matches the batch service's default). const endMs = input.window ? +input.window.endTime : Date.now(); const startMs = input.window ? +input.window.startTime : endMs - SEVEN_DAYS_MS; const startSec = Math.floor(startMs / 1000); @@ -441,10 +441,10 @@ export class EvalClient implements CoreEvalClient { const [runtimeRows, sharedRows] = await Promise.all([ runInsightsQuery(logs, [logGroupName], queryString, startSec, endSec).catch((error) => { if (error instanceof ResourceNotFoundException) { - throw new InputValidationError( + throw new ResourceNotFoundError( `No telemetry found for agent "${input.agent}": its runtime log group ${logGroupName} ` + `does not exist. Ensure the agent has been invoked and emits traces.`, - { meta: { agent: input.agent, logGroupName } }, + { cause: error, meta: { agent: input.agent, logGroupName } }, ); } throw error; @@ -454,7 +454,7 @@ export class EvalClient implements CoreEvalClient { throw error; }), ]); - const traces = groupSpansBySession([...sharedRows, ...runtimeRows]); + const traces = groupSpansBySession([...sharedRows, ...runtimeRows], this.logger); // Warn when explicitly requested sessions never showed up in the logs (aged // out, wrong id, or never emitted) so a caller isn't misled by a partial run. @@ -1390,12 +1390,6 @@ function sanitizeQueryValue(value: string): string { return value.replace(/'/g, ""); } -// buildSpanQuery is the single-phase Insights query: scope to one runtime by its -// OTel service.name, optionally narrow to specific sessions and/or one trace, and -// select the full span JSON (@message) plus the session id to group by. It does -// NOT over-filter on ispresent(kind) — that span-only predicate is what forced the -// old CLI's second query for log records; the looser scope returns everything for -// the session in one pass. function buildSpanQuery(serviceName: string, sessionIds?: string[], traceId?: string): string { let query = `fields @message, attributes.session.id as sessionId, traceId, spanId | filter resource.attributes.service.name in ['${sanitizeQueryValue(serviceName)}']`; @@ -1433,15 +1427,15 @@ async function runInsightsQuery( const result = await logs.send(new GetQueryResultsCommand({ queryId })); status = result.status ?? "Unknown"; if (status === "Failed" || status === "Cancelled" || status === "Timeout") { - throw new NetworkingError(`CloudWatch Logs Insights query ${status.toLowerCase()}`, { - meta: { queryId }, + throw new CloudWatchQueryError(`CloudWatch Logs Insights query ${status.toLowerCase()}`, { + meta: { queryId, status }, }); } if (status !== "Complete") await new Promise((resolve) => setTimeout(resolve, 1000)); } if (status !== "Complete") { - throw new NetworkingError("CloudWatch Logs Insights query did not finish in time", { - meta: { queryId }, + throw new CloudWatchQueryError("CloudWatch Logs Insights query did not finish in time", { + meta: { queryId, status }, }); } @@ -1466,9 +1460,10 @@ async function runInsightsQuery( // Group parsed @message docs by session, keeping only sessions with >=1 span // (Evaluate rejects log-only sessions), and derive each session's trace/tool ids. -function groupSpansBySession(rows: ResultField[][]): SessionTrace[] { +function groupSpansBySession(rows: ResultField[][], logger: Logger): SessionTrace[] { const docsBySession = new Map(); const sessionsWithSpans = new Set(); + let warnedAboutMalformedTelemetry = false; for (const row of rows) { const message = row.find((f) => f.field === "@message")?.value; const sessionId = row.find((f) => f.field === "sessionId")?.value; @@ -1479,6 +1474,10 @@ function groupSpansBySession(rows: ResultField[][]): SessionTrace[] { try { doc = JSON.parse(message) as SpanRecord; } catch { + if (!warnedAboutMalformedTelemetry) { + logger.warn("skipping malformed telemetry records"); + warnedAboutMalformedTelemetry = true; + } continue; } const list = docsBySession.get(sessionId); diff --git a/src/core/evalOnDemand.test.ts b/src/core/evalOnDemand.test.ts new file mode 100644 index 000000000..c3200b0c2 --- /dev/null +++ b/src/core/evalOnDemand.test.ts @@ -0,0 +1,138 @@ +import { describe, expect, test } from "bun:test"; +import { + GetAgentRuntimeCommand, + type BedrockAgentCoreControlClient, +} from "@aws-sdk/client-bedrock-agentcore-control"; +import type { BedrockAgentCoreClient } from "@aws-sdk/client-bedrock-agentcore"; +import { + GetQueryResultsCommand, + ResourceNotFoundException, + StartQueryCommand, + type CloudWatchLogsClient, + type ResultField, +} from "@aws-sdk/client-cloudwatch-logs"; +import type { IAMClient } from "@aws-sdk/client-iam"; +import { CloudWatchQueryError, ResourceNotFoundError } from "../errors"; +import type { Logger } from "../logging"; +import { EvalClient } from "./eval"; +import type { AwsClients } from "./types"; + +const OPTIONS = { region: "us-west-2" }; +const RUNTIME_ID = "runtime-1"; +const RUNTIME_LOG_GROUP = "/aws/bedrock-agentcore/runtimes/runtime-1-DEFAULT"; + +type QueryFailureStatus = "Failed" | "Cancelled" | "Timeout"; + +type LogsOptions = { + malformedRows?: ResultField[][]; + missingRuntimeLogGroup?: boolean; + status?: QueryFailureStatus; +}; + +function subject(options: LogsOptions = {}, logger?: Logger): EvalClient { + const control = { + send: async (command: unknown) => { + if (command instanceof GetAgentRuntimeCommand) { + return { agentRuntimeId: "runtime-1", agentRuntimeName: "agent-1" }; + } + throw new Error(`unexpected control command: ${(command as object).constructor.name}`); + }, + } as unknown as BedrockAgentCoreControlClient; + + const logs = { + send: async (command: unknown) => { + if (command instanceof StartQueryCommand) { + const logGroup = command.input.logGroupNames?.[0]; + if (options.missingRuntimeLogGroup && logGroup !== "aws/spans") { + throw new ResourceNotFoundException({ + $metadata: {}, + message: "log group does not exist", + }); + } + return { queryId: logGroup === "aws/spans" ? "shared-query" : "runtime-query" }; + } + if (command instanceof GetQueryResultsCommand) { + return { + status: options.status ?? "Complete", + results: command.input.queryId === "runtime-query" ? (options.malformedRows ?? []) : [], + }; + } + throw new Error(`unexpected logs command: ${(command as object).constructor.name}`); + }, + } as unknown as CloudWatchLogsClient; + + const clients: AwsClients = { + control: () => control, + data: () => ({}) as BedrockAgentCoreClient, + iam: () => ({}) as IAMClient, + logs: () => logs, + }; + return new EvalClient(clients, globalThis.fetch, logger); +} + +function telemetryRow(sessionId: string, message: string): ResultField[] { + return [ + { field: "@message", value: message }, + { field: "sessionId", value: sessionId }, + ]; +} + +describe("EvalClient on-demand trace collection", () => { + test("reports a missing runtime log group as ResourceNotFoundError", async () => { + const error = await subject({ missingRuntimeLogGroup: true }) + .getTracesForAgent({ agent: RUNTIME_ID, sessionIds: ["session-1"] }, OPTIONS) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(ResourceNotFoundError); + expect(error).toMatchObject({ + source: "user", + meta: { + agent: RUNTIME_ID, + logGroupName: RUNTIME_LOG_GROUP, + }, + }); + expect((error as Error).cause).toBeInstanceOf(ResourceNotFoundException); + }); + + test.each(["Failed", "Cancelled", "Timeout"] as const)( + "reports CloudWatch query status %s as CloudWatchQueryError", + async (status) => { + const error = await subject({ status }) + .getTracesForAgent({ agent: RUNTIME_ID, sessionIds: ["session-1"] }, OPTIONS) + .catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(CloudWatchQueryError); + expect(error).toMatchObject({ + source: "service", + meta: { status }, + }); + }, + ); + + test("warns once when malformed telemetry records are skipped", async () => { + const warnings: string[] = []; + const logger: Logger = { + debug: () => {}, + info: () => {}, + warn: (...messages) => warnings.push(messages.join(" ")), + error: () => {}, + child: () => logger, + }; + const rows = [ + telemetryRow( + "session-1", + JSON.stringify({ kind: "SERVER", traceId: "trace-1", spanId: "span-1" }), + ), + telemetryRow("session-1", "{"), + telemetryRow("session-1", "not-json"), + ]; + + const traces = await subject({ malformedRows: rows }, logger).getTracesForAgent( + { agent: RUNTIME_ID, sessionIds: ["session-1"] }, + OPTIONS, + ); + + expect(traces).toHaveLength(1); + expect(warnings).toEqual(["skipping malformed telemetry records"]); + }); +}); diff --git a/src/errors/errors.tsx b/src/errors/errors.tsx index e671c1996..dda3ac697 100644 --- a/src/errors/errors.tsx +++ b/src/errors/errors.tsx @@ -69,6 +69,13 @@ export class InputValidationError extends AgentCoreCLIError { } } +/** Error raised when a requested resource does not exist. */ +export class ResourceNotFoundError extends AgentCoreCLIError { + constructor(message?: string, options?: Omit) { + super(message, { ...options, source: ERROR_SOURCE.USER }); + } +} + /** Error raised when a command or operation has not been implemented yet. */ export class NotImplementedError extends AgentCoreCLIError { constructor(message?: string, options?: Omit) { @@ -178,6 +185,13 @@ export class NetworkingError extends AgentCoreCLIError { } } +/** A CloudWatch Logs Insights query reached a terminal failure state. */ +export class CloudWatchQueryError extends AgentCoreCLIError { + constructor(message: string, options?: Omit) { + super(message, { ...options, source: ERROR_SOURCE.SERVICE }); + } +} + /** Service data was returned successfully, but did not match the expected contract. */ export class MalformedServiceResponseError extends AgentCoreCLIError { constructor(message: string, options?: Omit) { diff --git a/src/errors/index.tsx b/src/errors/index.tsx index 9b3249fa2..feb51f0a4 100644 --- a/src/errors/index.tsx +++ b/src/errors/index.tsx @@ -1,5 +1,6 @@ export { AgentCoreCLIError, + CloudWatchQueryError, DeserializationError, EmbeddedAssetNotFoundError, FileWriteError, @@ -12,6 +13,7 @@ export { NetworkingError, NotImplementedError, ProjectFileExistsError, + ResourceNotFoundError, ResultTruncationError, RuntimeInvokeInterruptedError, RuntimeInvokeResponseError, diff --git a/src/handlers/eval/ondemand/ondemand.fixture.test.tsx b/src/handlers/eval/ondemand/ondemand.fixture.test.tsx index f34081c7d..39bf5767f 100644 --- a/src/handlers/eval/ondemand/ondemand.fixture.test.tsx +++ b/src/handlers/eval/ondemand/ondemand.fixture.test.tsx @@ -13,21 +13,6 @@ import { createRootHandler } from "../../index"; const REGION = "us-west-2"; const FIXTURES = join(import.meta.dir, "__fixtures__"); -// Record with: RECORD=1 bun test src/handlers/eval/ondemand/ondemand.fixture.test.tsx -// -// This exercises the real seam end to end: parsing → handler → CoreClient → -// getTracesForAgent (GetAgentRuntime + CloudWatch Logs Insights StartQuery / -// GetQueryResults, read from aws/spans and the runtime group) → evaluate (the -// Evaluate data-plane API) → rendered scores. -// -// Determinism: the window is PINNED (not --lookback-days) so the StartQuery input — -// which embeds startTime/endTime epoch seconds — hashes to the same fixture on -// record and replay. --session-ids bounds the fetch to the two sessions recorded -// against the live agent below. -// -// Re-recording needs the agent to still exist AND those sessions' spans to still be -// within CloudWatch retention (they age out). If they've aged out, invoke the agent -// to create fresh sessions, then repoint FIXTURE_SESSION_IDS + the window at them. const FIXTURE_AGENT = "asdf_MyAgent-3s5axvBC6Q"; const FIXTURE_SESSION_IDS = [ "67ebf93b-65e3-4127-9e13-483b239f256a", diff --git a/src/handlers/eval/ondemand/ondemand.test.tsx b/src/handlers/eval/ondemand/ondemand.test.tsx index 6d7173444..b8f9d14ee 100644 --- a/src/handlers/eval/ondemand/ondemand.test.tsx +++ b/src/handlers/eval/ondemand/ondemand.test.tsx @@ -24,7 +24,9 @@ const TRACE: SessionTrace = { const RESULT: EvaluateResult = { sessionsRequested: 1, sessionsEvaluated: 1, - results: [{ evaluatorId: "Builtin.Helpfulness", value: 0.9 } as EvaluateResult["results"][number]], + results: [ + { evaluatorId: "Builtin.Helpfulness", value: 0.9 } as EvaluateResult["results"][number], + ], }; async function run(args: string[], configure?: (core: TestCoreClient) => void) { @@ -69,33 +71,21 @@ describe("eval ondemand command hierarchy", () => { }); describe("eval ondemand evaluate validation", () => { - test("requires --agent", async () => { - await expect( - run([ - "eval", - "ondemand", - "evaluate", - "--evaluator", - "Builtin.Helpfulness", - "--session-ids", - "s1", - ]), - ).rejects.toThrow(/--agent/); - }); - - test("requires --evaluator", async () => { - await expect( - run(["eval", "ondemand", "evaluate", "--agent", "a-1", "--session-ids", "s1"]), - ).rejects.toThrow(/--evaluator/); - }); - - test("rejects an empty session source", async () => { - await expect(run(BASE)).rejects.toThrow(/session source/); - }); - - test("rejects --lookback-days combined with an explicit window", async () => { - await expect( - run([ + test.each<[string, string[], RegExp]>([ + [ + "requires --agent", + ["eval", "ondemand", "evaluate", "--evaluator", "Builtin.Helpfulness", "--session-ids", "s1"], + /--agent/, + ], + [ + "requires --evaluator", + ["eval", "ondemand", "evaluate", "--agent", "a-1", "--session-ids", "s1"], + /--evaluator/, + ], + ["rejects an empty session source", BASE, /session source/], + [ + "rejects --lookback-days combined with an explicit window", + [ ...BASE, "--lookback-days", "7", @@ -103,20 +93,21 @@ describe("eval ondemand evaluate validation", () => { "2026-01-01T00:00:00Z", "--end-time", "2026-01-02T00:00:00Z", - ]), - ).rejects.toThrow(/cannot be combined/); - }); - - test("rejects a half-open explicit window", async () => { - await expect(run([...BASE, "--start-time", "2026-01-01T00:00:00Z"])).rejects.toThrow( + ], + /cannot be combined/, + ], + [ + "rejects a half-open explicit window", + [...BASE, "--start-time", "2026-01-01T00:00:00Z"], /together/, - ); - }); - - test("rejects start-time not before end-time", async () => { - await expect( - run([...BASE, "--start-time", "2026-01-02T00:00:00Z", "--end-time", "2026-01-01T00:00:00Z"]), - ).rejects.toThrow(/before/); + ], + [ + "rejects start-time not before end-time", + [...BASE, "--start-time", "2026-01-02T00:00:00Z", "--end-time", "2026-01-01T00:00:00Z"], + /before/, + ], + ])("%s", async (_name, args, expectedError) => { + await expect(run(args)).rejects.toThrow(expectedError); }); }); diff --git a/src/handlers/eval/types.tsx b/src/handlers/eval/types.tsx index 6e1875b18..278f8f34c 100644 --- a/src/handlers/eval/types.tsx +++ b/src/handlers/eval/types.tsx @@ -205,25 +205,15 @@ export type StartBatchEvaluationInput = { kmsKeyArn?: string; }; -// SpanRecord is one OTel span/log document — the parsed `@message` JSON of a -// CloudWatch Logs Insights result row. Left open (arbitrary JSON) because it is -// handed to the Evaluate API's `sessionSpans` verbatim; the CLI only reads a few -// well-known fields off it (traceId, spanId, attributes) to route evaluators. export type SpanRecord = Record; -// SessionTrace is one session's gathered telemetry, grouped client-side. Neutral -// by design — no Evaluate coupling — so getTracesForAgent stays reusable and -// EvalClient.evaluate owns the mapping into the Evaluate request shape. export type SessionTrace = { - sessionId: string; // read from attributes.session.id - spans: SpanRecord[]; // full OTel span JSON (@message) - traceIds: string[]; // for TRACE-level evaluators (evaluationTarget.traceIds) - toolCallSpanIds: string[]; // for TOOL_CALL-level evaluators (evaluationTarget.spanIds) + sessionId: string; + spans: SpanRecord[]; + traceIds: string[]; + toolCallSpanIds: string[]; }; -// GetTracesInput selects which sessions' traces to read for one agent. `sessionIds` -// and `traceId` are independent, optional, AND-ed query filters; with neither, the -// `window` bounds discovery. `window` unset ⇒ the client defaults to now−7d. export type GetTracesInput = { agent: string; endpoint?: string; @@ -232,21 +222,12 @@ export type GetTracesInput = { traceId?: string; }; -// EvaluateInput is the CLI-facing shape for the synchronous Evaluate path. -// `groundTruth` is the SDK-native array, passed verbatim; EvalClient groups it by -// session (context.spanContext.sessionId) and attaches per session. export type EvaluateInput = { traces: SessionTrace[]; evaluatorIds: string[]; groundTruth?: EvaluationReferenceInput[]; }; -// EvaluateResult returns the raw Evaluate API results across all evaluators and -// sessions (each carries its own evaluatorId + span context). No aggregation — the -// caller renders the raw scores. The two counts are distinct on purpose: -// `sessionsRequested` is how many gathered sessions were handed to Evaluate; -// `sessionsEvaluated` is how many actually produced results (a TRACE/TOOL_CALL -// session with no matching ids is requested but not evaluated). export type EvaluateResult = { sessionsRequested: number; sessionsEvaluated: number; From 1779d5321ce3400565eae107a8bdcad2b5bd5b7d Mon Sep 17 00:00:00 2001 From: jariy17 Date: Fri, 14 Aug 2026 20:27:01 +0000 Subject: [PATCH 2/2] test(eval): consolidate on-demand regression coverage --- src/core/evalOnDemand.test.ts | 138 ---------------- src/handlers/eval/ondemand/ondemand.test.tsx | 161 +++++++++++++++++++ 2 files changed, 161 insertions(+), 138 deletions(-) delete mode 100644 src/core/evalOnDemand.test.ts diff --git a/src/core/evalOnDemand.test.ts b/src/core/evalOnDemand.test.ts deleted file mode 100644 index c3200b0c2..000000000 --- a/src/core/evalOnDemand.test.ts +++ /dev/null @@ -1,138 +0,0 @@ -import { describe, expect, test } from "bun:test"; -import { - GetAgentRuntimeCommand, - type BedrockAgentCoreControlClient, -} from "@aws-sdk/client-bedrock-agentcore-control"; -import type { BedrockAgentCoreClient } from "@aws-sdk/client-bedrock-agentcore"; -import { - GetQueryResultsCommand, - ResourceNotFoundException, - StartQueryCommand, - type CloudWatchLogsClient, - type ResultField, -} from "@aws-sdk/client-cloudwatch-logs"; -import type { IAMClient } from "@aws-sdk/client-iam"; -import { CloudWatchQueryError, ResourceNotFoundError } from "../errors"; -import type { Logger } from "../logging"; -import { EvalClient } from "./eval"; -import type { AwsClients } from "./types"; - -const OPTIONS = { region: "us-west-2" }; -const RUNTIME_ID = "runtime-1"; -const RUNTIME_LOG_GROUP = "/aws/bedrock-agentcore/runtimes/runtime-1-DEFAULT"; - -type QueryFailureStatus = "Failed" | "Cancelled" | "Timeout"; - -type LogsOptions = { - malformedRows?: ResultField[][]; - missingRuntimeLogGroup?: boolean; - status?: QueryFailureStatus; -}; - -function subject(options: LogsOptions = {}, logger?: Logger): EvalClient { - const control = { - send: async (command: unknown) => { - if (command instanceof GetAgentRuntimeCommand) { - return { agentRuntimeId: "runtime-1", agentRuntimeName: "agent-1" }; - } - throw new Error(`unexpected control command: ${(command as object).constructor.name}`); - }, - } as unknown as BedrockAgentCoreControlClient; - - const logs = { - send: async (command: unknown) => { - if (command instanceof StartQueryCommand) { - const logGroup = command.input.logGroupNames?.[0]; - if (options.missingRuntimeLogGroup && logGroup !== "aws/spans") { - throw new ResourceNotFoundException({ - $metadata: {}, - message: "log group does not exist", - }); - } - return { queryId: logGroup === "aws/spans" ? "shared-query" : "runtime-query" }; - } - if (command instanceof GetQueryResultsCommand) { - return { - status: options.status ?? "Complete", - results: command.input.queryId === "runtime-query" ? (options.malformedRows ?? []) : [], - }; - } - throw new Error(`unexpected logs command: ${(command as object).constructor.name}`); - }, - } as unknown as CloudWatchLogsClient; - - const clients: AwsClients = { - control: () => control, - data: () => ({}) as BedrockAgentCoreClient, - iam: () => ({}) as IAMClient, - logs: () => logs, - }; - return new EvalClient(clients, globalThis.fetch, logger); -} - -function telemetryRow(sessionId: string, message: string): ResultField[] { - return [ - { field: "@message", value: message }, - { field: "sessionId", value: sessionId }, - ]; -} - -describe("EvalClient on-demand trace collection", () => { - test("reports a missing runtime log group as ResourceNotFoundError", async () => { - const error = await subject({ missingRuntimeLogGroup: true }) - .getTracesForAgent({ agent: RUNTIME_ID, sessionIds: ["session-1"] }, OPTIONS) - .catch((caught: unknown) => caught); - - expect(error).toBeInstanceOf(ResourceNotFoundError); - expect(error).toMatchObject({ - source: "user", - meta: { - agent: RUNTIME_ID, - logGroupName: RUNTIME_LOG_GROUP, - }, - }); - expect((error as Error).cause).toBeInstanceOf(ResourceNotFoundException); - }); - - test.each(["Failed", "Cancelled", "Timeout"] as const)( - "reports CloudWatch query status %s as CloudWatchQueryError", - async (status) => { - const error = await subject({ status }) - .getTracesForAgent({ agent: RUNTIME_ID, sessionIds: ["session-1"] }, OPTIONS) - .catch((caught: unknown) => caught); - - expect(error).toBeInstanceOf(CloudWatchQueryError); - expect(error).toMatchObject({ - source: "service", - meta: { status }, - }); - }, - ); - - test("warns once when malformed telemetry records are skipped", async () => { - const warnings: string[] = []; - const logger: Logger = { - debug: () => {}, - info: () => {}, - warn: (...messages) => warnings.push(messages.join(" ")), - error: () => {}, - child: () => logger, - }; - const rows = [ - telemetryRow( - "session-1", - JSON.stringify({ kind: "SERVER", traceId: "trace-1", spanId: "span-1" }), - ), - telemetryRow("session-1", "{"), - telemetryRow("session-1", "not-json"), - ]; - - const traces = await subject({ malformedRows: rows }, logger).getTracesForAgent( - { agent: RUNTIME_ID, sessionIds: ["session-1"] }, - OPTIONS, - ); - - expect(traces).toHaveLength(1); - expect(warnings).toEqual(["skipping malformed telemetry records"]); - }); -}); diff --git a/src/handlers/eval/ondemand/ondemand.test.tsx b/src/handlers/eval/ondemand/ondemand.test.tsx index b8f9d14ee..9c5d9082b 100644 --- a/src/handlers/eval/ondemand/ondemand.test.tsx +++ b/src/handlers/eval/ondemand/ondemand.test.tsx @@ -1,4 +1,21 @@ import { test, expect, describe } from "bun:test"; +import { + GetAgentRuntimeCommand, + GetEvaluatorCommand, + type BedrockAgentCoreControlClient, +} from "@aws-sdk/client-bedrock-agentcore-control"; +import { EvaluateCommand, type BedrockAgentCoreClient } from "@aws-sdk/client-bedrock-agentcore"; +import { + GetQueryResultsCommand, + ResourceNotFoundException, + StartQueryCommand, + type CloudWatchLogsClient, + type ResultField, +} from "@aws-sdk/client-cloudwatch-logs"; +import type { IAMClient } from "@aws-sdk/client-iam"; +import { CoreClient } from "../../../core"; +import { CloudWatchQueryError, ResourceNotFoundError } from "../../../errors"; +import type { Logger } from "../../../logging"; import { createRootHandler } from "../../index"; import { createSilentLogger, @@ -29,6 +46,17 @@ const RESULT: EvaluateResult = { ], }; +const RUNTIME_ID = "runtime-1"; +const RUNTIME_LOG_GROUP = "/aws/bedrock-agentcore/runtimes/runtime-1-DEFAULT"; + +type QueryFailureStatus = "Failed" | "Cancelled" | "Timeout"; + +type LogsOptions = { + malformedRows?: ResultField[][]; + missingRuntimeLogGroup?: boolean; + status?: QueryFailureStatus; +}; + async function run(args: string[], configure?: (core: TestCoreClient) => void) { const core = new TestCoreClient(); core.eval.setGetTracesResponse([TRACE]).setEvaluateResponse(RESULT); @@ -43,6 +71,85 @@ async function run(args: string[], configure?: (core: TestCoreClient) => void) { return { core, stdout: io.stdout(), stderr: io.stderr() }; } +async function runWithRealCore(options: LogsOptions, logger = createSilentLogger()) { + const control = { + send: async (command: unknown) => { + if (command instanceof GetAgentRuntimeCommand) { + return { agentRuntimeId: RUNTIME_ID, agentRuntimeName: "agent-1" }; + } + if (command instanceof GetEvaluatorCommand) { + return { evaluatorId: "Builtin.Helpfulness", level: "SESSION" }; + } + throw new Error(`unexpected control command: ${(command as object).constructor.name}`); + }, + } as unknown as BedrockAgentCoreControlClient; + + const data = { + send: async (command: unknown) => { + if (command instanceof EvaluateCommand) return { evaluationResults: [] }; + throw new Error(`unexpected data command: ${(command as object).constructor.name}`); + }, + } as unknown as BedrockAgentCoreClient; + + const logs = { + send: async (command: unknown) => { + if (command instanceof StartQueryCommand) { + const logGroup = command.input.logGroupNames?.[0]; + if (options.missingRuntimeLogGroup && logGroup !== "aws/spans") { + throw new ResourceNotFoundException({ + $metadata: {}, + message: "log group does not exist", + }); + } + return { queryId: logGroup === "aws/spans" ? "shared-query" : "runtime-query" }; + } + if (command instanceof GetQueryResultsCommand) { + return { + status: options.status ?? "Complete", + results: command.input.queryId === "runtime-query" ? (options.malformedRows ?? []) : [], + }; + } + throw new Error(`unexpected logs command: ${(command as object).constructor.name}`); + }, + } as unknown as CloudWatchLogsClient; + + const core = new CoreClient({ + createControlClient: () => control, + createDataClient: () => data, + createIamClient: () => ({}) as IAMClient, + createLogsClient: () => logs, + logger, + }); + const io = testIO(); + const root = createRootHandler(core, { + io: io.io, + logger, + globalConfigAccessor: new TestGlobalConfigAccessor(), + }); + await root.route([ + "node", + "agentcore", + "eval", + "ondemand", + "evaluate", + "--agent", + RUNTIME_ID, + "--evaluator", + "Builtin.Helpfulness", + "--session-ids", + "session-1", + "--region", + "us-west-2", + ]); +} + +function telemetryRow(sessionId: string, message: string): ResultField[] { + return [ + { field: "@message", value: message }, + { field: "sessionId", value: sessionId }, + ]; +} + const BASE = [ "eval", "ondemand", @@ -167,3 +274,57 @@ describe("eval ondemand evaluate orchestration", () => { expect(evaluate?.args[0]).toMatchObject({ groundTruth }); }); }); + +describe("eval ondemand evaluate telemetry failures", () => { + test("reports a missing runtime log group as ResourceNotFoundError", async () => { + const error = await runWithRealCore({ missingRuntimeLogGroup: true }).catch( + (caught: unknown) => caught, + ); + + expect(error).toBeInstanceOf(ResourceNotFoundError); + expect(error).toMatchObject({ + source: "user", + meta: { + agent: RUNTIME_ID, + logGroupName: RUNTIME_LOG_GROUP, + }, + }); + expect((error as Error).cause).toBeInstanceOf(ResourceNotFoundException); + }); + + test.each(["Failed", "Cancelled", "Timeout"] as const)( + "reports CloudWatch query status %s as CloudWatchQueryError", + async (status) => { + const error = await runWithRealCore({ status }).catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(CloudWatchQueryError); + expect(error).toMatchObject({ + source: "service", + meta: { status }, + }); + }, + ); + + test("warns once when malformed telemetry records are skipped", async () => { + const warnings: string[] = []; + const logger: Logger = { + debug: () => {}, + info: () => {}, + warn: (...messages) => warnings.push(messages.join(" ")), + error: () => {}, + child: () => logger, + }; + const rows = [ + telemetryRow( + "session-1", + JSON.stringify({ kind: "SERVER", traceId: "trace-1", spanId: "span-1" }), + ), + telemetryRow("session-1", "{"), + telemetryRow("session-1", "not-json"), + ]; + + await runWithRealCore({ malformedRows: rows }, logger); + + expect(warnings).toEqual(["skipping malformed telemetry records"]); + }); +});