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/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..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, @@ -24,7 +41,20 @@ 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], + ], +}; + +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) { @@ -41,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", @@ -69,33 +178,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 +200,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); }); }); @@ -176,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"]); + }); +}); 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;