From b7472577cc69ff8e7d2f40d5657244bde4b7cf0b Mon Sep 17 00:00:00 2001 From: jariy17 Date: Tue, 18 Aug 2026 22:33:57 +0000 Subject: [PATCH] =?UTF-8?q?feat(eval):=20batch-evaluation=20simulate=20?= =?UTF-8?q?=E2=80=94=20invokeDataset=20+=20self-running=20example=20classe?= =?UTF-8?q?s?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/core/eval.tsx | 117 ++++++++- .../__snapshots__/invokeDataset.test.ts.snap | 172 +++++++++++++ .../eval/invokeDataset/example/predefined.ts | 66 +++++ .../eval/invokeDataset/example/simulated.ts | 23 ++ src/core/eval/invokeDataset/example/types.ts | 17 ++ .../eval/invokeDataset/invokeDataset.test.ts | 235 ++++++++++++++++++ src/core/eval/invokeDataset/load.ts | 69 +++++ src/core/eval/invokeDataset/run.ts | 27 ++ src/core/invokeRuntime.ts | 221 ++++++++++++++++ src/core/runtime.tsx | 164 +----------- .../batch-evaluation.test.tsx | 2 +- src/handlers/eval/batch-evaluation/index.tsx | 2 + .../__snapshots__/simulate.test.tsx.snap | 23 ++ .../eval/batch-evaluation/simulate/index.tsx | 118 +++++++++ .../simulate/simulate.test.tsx | 182 ++++++++++++++ src/handlers/eval/types.tsx | 42 ++++ src/io/index.ts | 1 + src/io/template.test.ts | 32 +++ src/io/template.ts | 33 +++ src/testing/TestCoreClient.tsx | 23 ++ 20 files changed, 1414 insertions(+), 155 deletions(-) create mode 100644 src/core/eval/invokeDataset/__snapshots__/invokeDataset.test.ts.snap create mode 100644 src/core/eval/invokeDataset/example/predefined.ts create mode 100644 src/core/eval/invokeDataset/example/simulated.ts create mode 100644 src/core/eval/invokeDataset/example/types.ts create mode 100644 src/core/eval/invokeDataset/invokeDataset.test.ts create mode 100644 src/core/eval/invokeDataset/load.ts create mode 100644 src/core/eval/invokeDataset/run.ts create mode 100644 src/core/invokeRuntime.ts create mode 100644 src/handlers/eval/batch-evaluation/simulate/__snapshots__/simulate.test.tsx.snap create mode 100644 src/handlers/eval/batch-evaluation/simulate/index.tsx create mode 100644 src/handlers/eval/batch-evaluation/simulate/simulate.test.tsx create mode 100644 src/io/template.test.ts create mode 100644 src/io/template.ts diff --git a/src/core/eval.tsx b/src/core/eval.tsx index aef88d11d..8888b7b4f 100644 --- a/src/core/eval.tsx +++ b/src/core/eval.tsx @@ -79,6 +79,8 @@ import { } from "@aws-sdk/client-cloudwatch-logs"; import type { DocumentType } from "@smithy/types"; import { randomUUID } from "node:crypto"; +import { unlink } from "node:fs/promises"; +import { tmpdir } from "node:os"; import { basename, dirname, extname, join } from "node:path"; import { Transform } from "node:stream"; import { setTimeout as sleep } from "node:timers/promises"; @@ -108,12 +110,18 @@ import type { LlmAsAJudgeUpdate, SessionSourceValue, SessionTrace, + InvokeDatasetInput, + InvokeDatasetResult, SpanRecord, StartBatchEvaluationInput, UpdateConfigurationBundleInput, UpdateOnlineEvalInput, } from "../handlers/eval/types"; -import { atomicWrite, atomicWriteStream, readTextFile } from "../io"; +import { atomicWrite, atomicWriteStream, readTextFile, renderJsonTemplate } from "../io"; +import { accountIdFromRuntimeArn, invokeRuntime } from "./invokeRuntime"; +import { DatasetLoader } from "./eval/invokeDataset/load"; +import { runExamples } from "./eval/invokeDataset/run"; +import type { RunContext } from "./eval/invokeDataset/example/types"; import { isTerminalStatus, readEvaluationResults } from "./batchEvaluationResults"; import { applyExampleIds, diffExamples, indexRemoteById, parseJsonl } from "./datasetDiff"; import type { Addition } from "./datasetDiff"; @@ -507,6 +515,113 @@ export class EvalClient implements CoreEvalClient { }; } + async invokeDataset( + input: InvokeDatasetInput, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + const examples = DatasetLoader.load( + await this.readDatasetText(input.dataset, input.datasetVersion, options, signal), + ); + + // Resolve the runtime once, reused for every session. + const runtime = await this.clients + .control(toClientConfig(options)) + .send(new GetAgentRuntimeCommand({ agentRuntimeId: input.runtimeId }), { + abortSignal: signal, + }); + const deps = { clients: this.clients, fetch: this.fetch, logger: this.logger }; + const accountId = accountIdFromRuntimeArn(runtime.agentRuntimeArn); + + const { ok, failed, firstError } = await runExamples(examples, async (example) => { + // One session per example; the id is a client-owned input per the AgentCore docs, + // reused across turns so the conversation and its per-turn traces stay in order. + const sessionId = randomUUID(); + const ctx: RunContext = { + invokeOnce: async (payload) => { + const response = await invokeRuntime( + deps, + { + runtimeId: input.runtimeId, + accountId, + qualifier: input.qualifier ?? DEFAULT_ENDPOINT_QUALIFIER, + payload: renderJsonTemplate(input.payloadTemplate, { input: payload }), + contentType: "application/json", + accept: "application/json", + ...(input.headers?.length ? { applicationHeaders: input.headers } : {}), + ...(input.bearerToken !== undefined ? { bearerToken: input.bearerToken } : {}), + runtimeSessionId: sessionId, + runtimeUserId: input.userId, + }, + options, + signal, + ); + // Read to completion to free the socket; a scripted example ignores the text. + let text = ""; + const decoder = new TextDecoder(); + for await (const chunk of response.body) text += decoder.decode(chunk, { stream: true }); + text += decoder.decode(); + return { text }; + }, + }; + try { + const groundTruth = await example.run(ctx); + return { exampleId: example.exampleId, sessionId, groundTruth }; + } catch (error) { + this.logger.debug( + `invokeDataset: invoke failed for example "${example.exampleId}" (${example.schemaType}): ${(error as Error).message}`, + ); + throw error; + } + }); + + if (failed > 0) { + this.logger.warn(`invokeDataset: ${failed} example(s) failed to invoke and were dropped`); + } + + // AgentCore emits spans ~30s-3min after invoke; grade too early and it reads an empty + // log group and fails every session. Disabled via SIMULATE_INGESTION_WAIT_MS=0 (tests). + const waitMs = Number(process.env.SIMULATE_INGESTION_WAIT_MS ?? 180_000); + if (ok.length > 0 && waitMs > 0) { + this.logger.info( + `waiting ${Math.round(waitMs / 1000)}s for span ingestion before evaluating`, + ); + await sleep(waitMs, undefined, { signal }); + } + + return { sessions: ok, invoked: ok.length, failed, firstError }; + } + + // Resolve a dataset ref to JSONL text: a local path directly, else download the id to a + // temp file (cleaned up here). Reuses readLocalDatasetFile so replay reads like update. + private async readDatasetText( + ref: string, + version: string | undefined, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + if (await Bun.file(ref).exists()) return readLocalDatasetFile(ref, signal); + const path = await this.downloadDatasetToTemp(ref, version, options, signal); + try { + return await readLocalDatasetFile(path, signal); + } finally { + await unlink(path).catch(() => {}); + } + } + + // downloadDatasetToTemp streams a dataset version's JSONL to a temp file so + // readDatasetText can read it — reuses downloadDataset rather than re-fetching. + private async downloadDatasetToTemp( + id: string, + version: string | undefined, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + const path = join(tmpdir(), `agentcore-dataset-${randomUUID()}.jsonl`); + await this.downloadDataset(id, version, path, options, signal); + return path; + } + async createOnlineEvaluationConfig( input: CreateOnlineEvalInput, options: CoreOptions, diff --git a/src/core/eval/invokeDataset/__snapshots__/invokeDataset.test.ts.snap b/src/core/eval/invokeDataset/__snapshots__/invokeDataset.test.ts.snap new file mode 100644 index 000000000..c7a35def0 --- /dev/null +++ b/src/core/eval/invokeDataset/__snapshots__/invokeDataset.test.ts.snap @@ -0,0 +1,172 @@ +// Bun Snapshot v1, https://bun.sh/docs/test/snapshots + +exports[`EvalClient.invokeDataset golden: sessions + ground truth over representative datasets 1`] = ` +{ + "assertions + trajectory + sparse turns (full inline shape)": { + "failed": 0, + "invoked": 1, + "sessions": [ + { + "exampleId": "orders-1", + "groundTruth": { + "assertions": [ + { + "text": "stays polite", + }, + { + "text": "does not promise a date", + }, + ], + "expectedTrajectory": { + "toolNames": [ + "refund_lookup", + "refund_create", + ], + }, + "turns": [ + { + "input": { + "prompt": "I want a refund", + }, + }, + { + "expectedResponse": { + "text": "Refund started", + }, + "input": { + "prompt": "order 123", + }, + }, + ], + }, + "sessionId": "", + }, + ], + }, + "empty assertions/trajectory arrays are omitted": { + "failed": 0, + "invoked": 1, + "sessions": [ + { + "exampleId": "e4", + "groundTruth": { + "turns": [ + { + "expectedResponse": { + "text": "r1", + }, + "input": { + "prompt": "t1", + }, + }, + ], + }, + "sessionId": "", + }, + ], + }, + "empty expected_response is treated as no expectation": { + "failed": 0, + "invoked": 1, + "sessions": [ + { + "exampleId": "e3", + "groundTruth": undefined, + "sessionId": "", + }, + ], + }, + "legacy scenario_id fallback + unicode id": { + "failed": 0, + "invoked": 1, + "sessions": [ + { + "exampleId": "café-日本-🎉", + "groundTruth": { + "turns": [ + { + "expectedResponse": { + "text": "ok", + }, + "input": { + "prompt": "1", + }, + }, + ], + }, + "sessionId": "", + }, + ], + }, + "multi-turn, sparse expectation keeps its turn position": { + "failed": 0, + "invoked": 1, + "sessions": [ + { + "exampleId": "e2", + "groundTruth": { + "turns": [ + { + "input": { + "prompt": "t1", + }, + }, + { + "input": { + "prompt": "t2", + }, + }, + { + "expectedResponse": { + "text": "42", + }, + "input": { + "prompt": "t3", + }, + }, + ], + }, + "sessionId": "", + }, + ], + }, + "single turn, no ground truth": { + "failed": 0, + "invoked": 1, + "sessions": [ + { + "exampleId": "e1", + "groundTruth": undefined, + "sessionId": "", + }, + ], + }, + "tolerates blank lines and CRLF between multiple rows": { + "failed": 0, + "invoked": 2, + "sessions": [ + { + "exampleId": "a", + "groundTruth": undefined, + "sessionId": "", + }, + { + "exampleId": "b", + "groundTruth": { + "turns": [ + { + "expectedResponse": { + "text": "ok", + }, + "input": { + "prompt": "2", + }, + }, + ], + }, + "sessionId": "", + }, + ], + }, +} +`; diff --git a/src/core/eval/invokeDataset/example/predefined.ts b/src/core/eval/invokeDataset/example/predefined.ts new file mode 100644 index 000000000..240447342 --- /dev/null +++ b/src/core/eval/invokeDataset/example/predefined.ts @@ -0,0 +1,66 @@ +import type { InlineGroundTruth } from "@aws-sdk/client-bedrock-agentcore"; +import { InputValidationError } from "../../../../errors"; +import type { Example, RunContext } from "./types"; + +type Turn = { input: string; expectedResponse?: string }; + +export class PredefinedExample implements Example { + readonly schemaType = "AGENTCORE_EVALUATION_PREDEFINED_V1" as const; + readonly turns: Turn[]; + readonly assertions?: string[]; + readonly expectedTrajectory?: string[]; + + constructor( + readonly exampleId: string, + row: Record, + ) { + const turns = (Array.isArray(row.turns) ? row.turns : []).map((t: unknown, i: number) => { + // Reject a non-object entry here; dereferencing it below would throw a raw TypeError. + if (typeof t !== "object" || t === null) { + throw new InputValidationError(`example "${exampleId}" turn ${i + 1} is not an object`); + } + const turn = t as Record; + return { + input: String(turn.input ?? ""), + expectedResponse: turn.expected_response as string | undefined, + }; + }); + if (turns.length === 0) { + throw new InputValidationError(`example "${exampleId}" has no turns`); + } + this.turns = turns; + this.assertions = row.assertions as string[] | undefined; + this.expectedTrajectory = row.expected_trajectory as string[] | undefined; + } + + // Sequential and awaited: the turns share one session, so racing them would interleave + // the conversation and misalign per-turn traces with the ground truth. + async run(ctx: RunContext): Promise { + for (const turn of this.turns) await ctx.invokeOnce(turn.input); + return this.groundTruth(); + } + + // Emit every turn (carrying its prompt), not just those with an expectation: filtering + // renumbers the rest, scoring turn 3's expectation against turn 1. The service's + // alignment rule is undocumented, so the prompt keeps index and content matching both valid. + private groundTruth(): InlineGroundTruth | undefined { + // Empty expected_response means no expectation: the service rejects a zero-length + // expectedResponse.text (min 1), so treat "" the same as an omitted field. + const turns = this.turns.some((t) => t.expectedResponse) + ? this.turns.map((t) => ({ + input: { prompt: t.input }, + ...(t.expectedResponse && { expectedResponse: { text: t.expectedResponse } }), + })) + : []; + const assertions = this.assertions?.map((text) => ({ text })); + const inline: InlineGroundTruth = { + // Omit empty arrays — the service rejects zero-length assertions/turns (min-1). + ...(assertions && assertions.length > 0 && { assertions }), + ...(this.expectedTrajectory?.length && { + expectedTrajectory: { toolNames: this.expectedTrajectory }, + }), + ...(turns.length > 0 && { turns }), + }; + return Object.keys(inline).length > 0 ? inline : undefined; + } +} diff --git a/src/core/eval/invokeDataset/example/simulated.ts b/src/core/eval/invokeDataset/example/simulated.ts new file mode 100644 index 000000000..48fec52c2 --- /dev/null +++ b/src/core/eval/invokeDataset/example/simulated.ts @@ -0,0 +1,23 @@ +import type { InlineGroundTruth } from "@aws-sdk/client-bedrock-agentcore"; +import { NotImplementedError } from "../../../../errors"; +import type { Example, RunContext } from "./types"; + +// Not shipped: replaying a simulated example needs an LLM actor we don't run yet. Throw +// at construction (= load time) so the user fails early, not with a per-row misdiagnosis. +export class SimulatedExample implements Example { + readonly schemaType = "AGENTCORE_EVALUATION_SIMULATED_V1" as const; + + constructor( + readonly exampleId: string, + _row: Record, + ) { + throw new NotImplementedError( + `example "${exampleId}" is a simulated example (actor_profile), which this ` + + `command cannot replay yet — it has no scripted turns`, + ); + } + + run(_ctx: RunContext): Promise { + throw new NotImplementedError("simulated example replay is not implemented"); + } +} diff --git a/src/core/eval/invokeDataset/example/types.ts b/src/core/eval/invokeDataset/example/types.ts new file mode 100644 index 000000000..17ffce7b4 --- /dev/null +++ b/src/core/eval/invokeDataset/example/types.ts @@ -0,0 +1,17 @@ +import type { DatasetSchemaType } from "@aws-sdk/client-bedrock-agentcore-control"; +import type { InlineGroundTruth } from "@aws-sdk/client-bedrock-agentcore"; + +// A record, not a bare string, so a future tool-branching type can widen it by a field. +export type TurnResult = { text: string }; + +// The per-session transport handed to run(): one call = one turn. Session id, auth, and +// templating are bound by the caller, so an example only decides what to say next. +export type RunContext = { invokeOnce: (payload: string) => Promise }; + +// An interface, not a base class: no shared state to inherit, and the machine dispatches +// by calling run(). +export interface Example { + readonly schemaType: DatasetSchemaType; + readonly exampleId: string; + run(ctx: RunContext): Promise; +} diff --git a/src/core/eval/invokeDataset/invokeDataset.test.ts b/src/core/eval/invokeDataset/invokeDataset.test.ts new file mode 100644 index 000000000..3ad7574de --- /dev/null +++ b/src/core/eval/invokeDataset/invokeDataset.test.ts @@ -0,0 +1,235 @@ +// Disables the post-invoke span-ingestion wait so the replay returns immediately. +process.env.SIMULATE_INGESTION_WAIT_MS = "0"; + +import { afterEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, writeFileSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { GetAgentRuntimeCommand } from "@aws-sdk/client-bedrock-agentcore-control"; +import { InvokeAgentRuntimeCommand } from "@aws-sdk/client-bedrock-agentcore"; +import { EvalClient } from "../../eval"; +import type { AwsClients, CoreFetch } from "../../types"; +import type { InvokedSession } from "../../../handlers/eval/types"; + +// End-to-end coverage of EvalClient.invokeDataset over a fake AWS layer. Exercising the real +// method also exercises its consumers — DatasetLoader, the Example classes, runExamples, +// renderJsonTemplate, and invokeRuntime's IAM path — so those need no separate unit tests. + +const OPTIONS = { region: "us-west-2" }; +const RUNTIME_ARN = "arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/rt-1"; +const row = (o: object) => JSON.stringify(o); + +async function* replyBytes(text: string): AsyncGenerator { + yield new TextEncoder().encode(text); +} + +// A fake AWS layer: control resolves the runtime, data answers each invoke. Records every +// payload it was asked to send, and per `opts` can fail or delay specific invokes. +function fakeClients(opts: { fail?: (payload: string) => boolean; delayMs?: number } = {}): { + clients: AwsClients; + payloads: string[]; + peak: () => number; +} { + const payloads: string[] = []; + let inFlight = 0; + let peak = 0; + const send = async (command: unknown) => { + if (command instanceof GetAgentRuntimeCommand) return { agentRuntimeArn: RUNTIME_ARN }; + if (command instanceof InvokeAgentRuntimeCommand) { + const payload = new TextDecoder().decode(command.input.payload as Uint8Array); + payloads.push(payload); + if (opts.fail?.(payload)) throw new Error(`invoke failed for ${payload}`); + inFlight++; + peak = Math.max(peak, inFlight); + if (opts.delayMs) await new Promise((r) => setTimeout(r, opts.delayMs)); + inFlight--; + return { statusCode: 200, contentType: "application/json", response: replyBytes("ok") }; + } + throw new Error( + `unexpected command: ${(command as { constructor: { name: string } }).constructor.name}`, + ); + }; + const client = { send } as never; + return { + clients: { control: () => client, data: () => client, iam: () => client, logs: () => client }, + payloads, + peak: () => peak, + }; +} + +const dirs: string[] = []; +afterEach(async () => { + await Promise.all(dirs.splice(0).map((d) => rm(d, { recursive: true, force: true }))); +}); + +function datasetFile(jsonl: string): string { + const dir = mkdtempSync(join(tmpdir(), "agentcore-invoke-ds-")); + dirs.push(dir); + const path = join(dir, "dataset.jsonl"); + writeFileSync(path, jsonl); + return path; +} + +function invokeDataset(jsonl: string, clients: AwsClients) { + const fetch = (() => { + throw new Error("fetch is only used on the CUSTOM_JWT path, which these tests do not exercise"); + }) as unknown as CoreFetch; + return new EvalClient(clients, fetch).invokeDataset( + { runtimeId: "rt-1", payloadTemplate: '{"prompt":"{input}"}', dataset: datasetFile(jsonl) }, + OPTIONS, + ); +} + +// sessionId is a fresh UUID per example, so pin it to compare shapes; sort so completion +// order (which is nondeterministic under concurrency) doesn't churn the golden. +function normalize(sessions: InvokedSession[]) { + return [...sessions] + .sort((a, b) => a.exampleId.localeCompare(b.exampleId)) + .map((s) => ({ ...s, sessionId: "" })); +} + +const GOLDEN_FIXTURES: { name: string; jsonl: string }[] = [ + { + name: "single turn, no ground truth", + jsonl: row({ example_id: "e1", turns: [{ input: "hi" }] }), + }, + { + name: "multi-turn, sparse expectation keeps its turn position", + jsonl: row({ + example_id: "e2", + turns: [{ input: "t1" }, { input: "t2" }, { input: "t3", expected_response: "42" }], + }), + }, + { + name: "empty expected_response is treated as no expectation", + jsonl: row({ example_id: "e3", turns: [{ input: "t1", expected_response: "" }] }), + }, + { + name: "assertions + trajectory + sparse turns (full inline shape)", + jsonl: row({ + example_id: "orders-1", + turns: [ + { input: "I want a refund" }, + { input: "order 123", expected_response: "Refund started" }, + ], + assertions: ["stays polite", "does not promise a date"], + expected_trajectory: ["refund_lookup", "refund_create"], + }), + }, + { + name: "empty assertions/trajectory arrays are omitted", + jsonl: row({ + example_id: "e4", + turns: [{ input: "t1", expected_response: "r1" }], + assertions: [], + expected_trajectory: [], + }), + }, + { + name: "legacy scenario_id fallback + unicode id", + jsonl: row({ scenario_id: "café-日本-🎉", turns: [{ input: "1", expected_response: "ok" }] }), + }, + { + name: "tolerates blank lines and CRLF between multiple rows", + jsonl: + row({ example_id: "a", turns: [{ input: "1" }] }) + + "\r\n\r\n" + + row({ example_id: "b", turns: [{ input: "2", expected_response: "ok" }] }) + + "\r\n", + }, +]; + +const THROW_FIXTURES: { name: string; jsonl: string; error: RegExp }[] = [ + { + name: "both turns and actor_profile", + jsonl: row({ example_id: "x", turns: [{ input: "a" }], actor_profile: {} }), + error: /both 'turns' and 'actor_profile'/, + }, + { + name: "neither turns nor actor_profile", + jsonl: row({ example_id: "x" }), + error: /neither 'turns' nor 'actor_profile'/, + }, + { + name: "simulated example not supported yet", + jsonl: row({ example_id: "x", actor_profile: { goal: "g" } }), + error: /simulated example/, + }, + { + name: "duplicate example ids", + jsonl: [ + row({ example_id: "a", turns: [{ input: "1" }] }), + row({ example_id: "a", turns: [{ input: "2" }] }), + ].join("\n"), + error: /duplicate example_id: "a"/, + }, + { + name: "missing example id", + jsonl: row({ turns: [{ input: "hi" }] }), + error: /missing 'example_id'/, + }, + { name: "invalid JSON line", jsonl: "{not json", error: /not valid JSON/ }, + { name: "non-object row (null)", jsonl: "null", error: /not a JSON object/ }, + { name: "empty dataset", jsonl: "\n \n", error: /no examples/ }, + { name: "empty turns array", jsonl: row({ example_id: "x", turns: [] }), error: /has no turns/ }, + { + name: "non-object turn entry", + jsonl: row({ example_id: "x", turns: [null] }), + error: /turn 1 is not an object/, + }, +]; + +describe("EvalClient.invokeDataset", () => { + // One golden block over representative datasets: locks the created sessions + the exact + // inline ground-truth shape handed to the grader, across every ground-truth variation. + test("golden: sessions + ground truth over representative datasets", async () => { + const results: Record = {}; + for (const f of GOLDEN_FIXTURES) { + const r = await invokeDataset(f.jsonl, fakeClients().clients); + results[f.name] = { invoked: r.invoked, failed: r.failed, sessions: normalize(r.sessions) }; + } + expect(results).toMatchSnapshot(); + }); + + test.each(THROW_FIXTURES)("rejects and invokes nothing: $name", async ({ jsonl, error }) => { + const { clients, payloads } = fakeClients(); + await expect(invokeDataset(jsonl, clients)).rejects.toThrow(error); + expect(payloads).toEqual([]); + }); + + test("a failed invoke is counted and dropped; the rest still run", async () => { + const jsonl = [ + row({ example_id: "ok1", turns: [{ input: "hi" }] }), + row({ example_id: "bad", turns: [{ input: "FAIL" }] }), + row({ example_id: "ok2", turns: [{ input: "yo" }] }), + ].join("\n"); + const r = await invokeDataset(jsonl, fakeClients({ fail: (p) => p.includes("FAIL") }).clients); + expect(r.invoked).toBe(2); + expect(r.failed).toBe(1); + expect(r.sessions.map((s) => s.exampleId).sort()).toEqual(["ok1", "ok2"]); + expect(r.firstError?.message).toMatch(/invoke failed/); + }); + + test("invokes each turn exactly once across all examples, rendered through the template", async () => { + const jsonl = [ + row({ example_id: "a", turns: [{ input: "a1" }, { input: "a2" }] }), + row({ example_id: "b", turns: [{ input: "b1" }] }), + ].join("\n"); + const { clients, payloads } = fakeClients(); + await invokeDataset(jsonl, clients); + expect(payloads.sort()).toEqual( + ['{"prompt":"a1"}', '{"prompt":"a2"}', '{"prompt":"b1"}'].sort(), + ); + }); + + test("runs examples concurrently but never past the pool bound", async () => { + const jsonl = Array.from({ length: 12 }, (_, i) => + row({ example_id: `e${i}`, turns: [{ input: `p${i}` }] }), + ).join("\n"); + const { clients, peak } = fakeClients({ delayMs: 5 }); + await invokeDataset(jsonl, clients); + expect(peak()).toBeLessThanOrEqual(5); // runExamples default concurrency + expect(peak()).toBeGreaterThanOrEqual(2); // proves it did not run serially + }); +}); diff --git a/src/core/eval/invokeDataset/load.ts b/src/core/eval/invokeDataset/load.ts new file mode 100644 index 000000000..5fc544e24 --- /dev/null +++ b/src/core/eval/invokeDataset/load.ts @@ -0,0 +1,69 @@ +import type { DatasetSchemaType } from "@aws-sdk/client-bedrock-agentcore-control"; +import { InputValidationError } from "../../../errors"; +import type { Example } from "./example/types"; +import { PredefinedExample } from "./example/predefined"; +import { SimulatedExample } from "./example/simulated"; + +// Pure parse (no I/O) so it's testable with a plain string; the caller fetches the text. +export class DatasetLoader { + static load(text: string): Example[] { + const examples: Example[] = []; + const seen = new Set(); + for (const line of text.split("\n")) { + const trimmed = line.trim(); + if (!trimmed) continue; + let parsed: unknown; + try { + parsed = JSON.parse(trimmed); + } catch { + throw new InputValidationError("dataset contains a line that is not valid JSON"); + } + // Reject non-object rows before dereferencing — `null` is valid JSON and would throw. + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + throw new InputValidationError("dataset contains a line that is not a JSON object"); + } + const row = parsed as Record; + + // The id joins a session to its ground truth — a missing/duplicate one misassigns it. + const exampleId = String(row.example_id ?? row.scenario_id ?? ""); + if (!exampleId) { + throw new InputValidationError("dataset example is missing 'example_id'"); + } + if (seen.has(exampleId)) { + throw new InputValidationError(`dataset has a duplicate example_id: "${exampleId}"`); + } + seen.add(exampleId); + + examples.push(DatasetLoader.build(row, exampleId)); + } + if (examples.length === 0) throw new InputValidationError("dataset has no examples"); + return examples; + } + + // Classify by row shape — a local JSONL carries no schemaType, and AWS's own SDK + // dispatches this way. Refuse a both-row rather than silently dropping the actor profile. + private static build(row: Record, exampleId: string): Example { + const hasTurns = Array.isArray(row.turns); + const hasActor = row.actor_profile != null; + if (hasTurns && hasActor) { + throw new InputValidationError( + `example "${exampleId}" has both 'turns' and 'actor_profile' — one row cannot be both`, + ); + } + if (!hasTurns && !hasActor) { + throw new InputValidationError( + `example "${exampleId}" has neither 'turns' nor 'actor_profile'`, + ); + } + + const schemaType: DatasetSchemaType = hasTurns + ? "AGENTCORE_EVALUATION_PREDEFINED_V1" + : "AGENTCORE_EVALUATION_SIMULATED_V1"; + switch (schemaType) { + case "AGENTCORE_EVALUATION_PREDEFINED_V1": + return new PredefinedExample(exampleId, row); + case "AGENTCORE_EVALUATION_SIMULATED_V1": + return new SimulatedExample(exampleId, row); + } + } +} diff --git a/src/core/eval/invokeDataset/run.ts b/src/core/eval/invokeDataset/run.ts new file mode 100644 index 000000000..6f82a98b9 --- /dev/null +++ b/src/core/eval/invokeDataset/run.ts @@ -0,0 +1,27 @@ +// A failed worker is counted and dropped, not thrown — the caller reports all-failed via +// firstError. Bounded concurrency because each item invokes a live runtime. +export type ExampleRun = { ok: Result[]; failed: number; firstError?: Error }; + +export async function runExamples( + items: Item[], + worker: (item: Item) => Promise, + concurrency = 5, +): Promise> { + const ok: Result[] = []; + let failed = 0; + let firstError: Error | undefined; + let next = 0; + const run = async (): Promise => { + while (next < items.length) { + const item = items[next++]!; + try { + ok.push(await worker(item)); + } catch (error) { + failed++; + if (!firstError) firstError = error instanceof Error ? error : new Error(String(error)); + } + } + }; + await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, run)); + return { ok, failed, firstError }; +} diff --git a/src/core/invokeRuntime.ts b/src/core/invokeRuntime.ts new file mode 100644 index 000000000..c0853340e --- /dev/null +++ b/src/core/invokeRuntime.ts @@ -0,0 +1,221 @@ +import { randomUUID } from "node:crypto"; +import { InvokeAgentRuntimeCommand } from "@aws-sdk/client-bedrock-agentcore"; +import { InputValidationError } from "../errors"; +import type { Logger } from "../logging"; +import type { AwsClients, CoreFetch, CoreOptions } from "./types"; +import { abortable } from "./abortable"; +import { toClientConfig } from "./utils"; + +// Core's own invoke DTOs — duplicated from, not imported out of, the runtime handler so +// each handler keeps owning its own client-interface types (dependency inversion). Callers +// in the runtime handler pass their identically-shaped RuntimeInvokeRequest through structurally. +export type RuntimeInvokeRequest = { + runtimeId: string; + accountId: string; + qualifier: string; + payload: Uint8Array; + contentType: string; + accept?: string; + runtimeSessionId?: string; + runtimeUserId?: string; + applicationHeaders?: [string, string][]; + bearerToken?: string; + mcpSessionId?: string; + mcpProtocolVersion?: string; + mcpMethod?: string; + mcpName?: string; + traceId?: string; + traceParent?: string; + traceState?: string; + baggage?: string; +}; + +export type RuntimeInvokeResponse = { + statusCode: number; + contentType: string; + runtimeSessionId?: string; + mcpSessionId?: string; + mcpProtocolVersion?: string; + traceId?: string; + traceParent?: string; + traceState?: string; + baggage?: string; + body: AsyncIterable; +}; + +// The CUSTOM_JWT path puts the account id in the invocation URL, so a bad ARN must fail here. +export function accountIdFromRuntimeArn(arn: string | undefined): string { + const id = arn?.match(/^arn:[^:]+:bedrock-agentcore:[^:]*:(\d{12}):runtime\//)?.[1]; + if (!id) throw new InputValidationError("Runtime returned an invalid ARN"); + return id; +} + +// InvokeRuntimeDeps is the slice of a Core client an invoke needs. Passed in as a +// bag (not a sibling client) so both RuntimeClient and EvalClient.invokeDataset can call +// these free functions off their own `this.clients`/`this.fetch`/`this.logger`. +export type InvokeRuntimeDeps = { + clients: AwsClients; + fetch: CoreFetch; + logger: Logger; +}; + +async function* emptyBody(): AsyncGenerator {} + +// invokeRuntime dispatches by auth mode: a bearer token routes to the CUSTOM_JWT +// (raw fetch) path, otherwise the SigV4 SDK path. +export async function invokeRuntime( + deps: InvokeRuntimeDeps, + request: RuntimeInvokeRequest, + options: CoreOptions, + signal?: AbortSignal, +): Promise { + const { runtimeId, bearerToken } = request; + if (bearerToken !== undefined) { + const logger = deps.logger.child({ + operation: "invokeRuntime", + authMode: "CUSTOM_JWT", + runtimeId, + qualifier: request.qualifier, + region: options.region, + }); + return invokeRuntimeWithCustomJwt(deps, request, bearerToken, options, logger, signal); + } + return invokeRuntimeWithIam(deps, request, options, signal); +} + +async function invokeRuntimeWithCustomJwt( + deps: InvokeRuntimeDeps, + request: RuntimeInvokeRequest, + bearerToken: string, + options: CoreOptions, + logger: Logger, + signal?: AbortSignal, +): Promise { + const client = deps.clients.data(toClientConfig(options)); + const endpoint = client.config.endpointProvider({ + Region: options.region, + Endpoint: options.endpointUrl, + }); + const url = new URL(endpoint.url); + if (url.protocol !== "https:") { + throw new TypeError("CUSTOM_JWT requires an HTTPS endpoint"); + } + url.pathname = `${url.pathname.replace(/\/?$/, "/")}runtimes/${encodeURIComponent(request.runtimeId)}/invocations`; + url.search = new URLSearchParams({ + accountId: request.accountId, + qualifier: request.qualifier, + }).toString(); + const headers = new Headers(request.applicationHeaders); + try { + headers.set("Authorization", `Bearer ${bearerToken}`); + } catch { + throw new TypeError("Invalid bearer token"); + } + try { + for (const [name, value] of [ + ["Content-Type", request.contentType], + ["Accept", request.accept], + ["Mcp-Session-Id", request.mcpSessionId], + ["X-Amzn-Bedrock-AgentCore-Runtime-Session-Id", request.runtimeSessionId ?? randomUUID()], + ["Mcp-Protocol-Version", request.mcpProtocolVersion], + ["Mcp-Method", request.mcpMethod], + ["Mcp-Name", request.mcpName], + ["X-Amzn-Bedrock-AgentCore-Runtime-User-Id", request.runtimeUserId], + ["X-Amzn-Trace-Id", request.traceId], + ["traceparent", request.traceParent], + ["tracestate", request.traceState], + ["baggage", request.baggage], + ] as const) { + if (value !== undefined) headers.set(name, value); + } + } catch { + throw new TypeError("Invalid Runtime request header"); + } + let response: Response; + try { + response = await deps.fetch(url, { + method: "POST", + redirect: "error", + headers, + body: request.payload as RequestInit["body"], + signal, + }); + } catch (error) { + if (signal?.aborted) throw signal.reason ?? error; + logger + .child({ + errorName: + error instanceof TypeError + ? "TypeError" + : error instanceof Error + ? "Error" + : typeof error, + }) + .debug("Runtime invocation transport failed"); + throw new Error("Runtime invocation failed"); + } + if (!response.ok) { + logger + .child({ httpStatusCode: response.status }) + .debug("Runtime invocation returned a non-success response"); + await response.body?.cancel().catch(() => undefined); + throw new Error(`HTTP ${response.status}`); + } + const body = (response.body as AsyncIterable | null) ?? emptyBody(); + return { + statusCode: response.status, + contentType: response.headers.get("content-type") ?? "", + runtimeSessionId: + response.headers.get("x-amzn-bedrock-agentcore-runtime-session-id") ?? undefined, + mcpSessionId: response.headers.get("mcp-session-id") ?? undefined, + mcpProtocolVersion: response.headers.get("mcp-protocol-version") ?? undefined, + traceId: response.headers.get("x-amzn-trace-id") ?? undefined, + traceParent: response.headers.get("traceparent") ?? undefined, + traceState: response.headers.get("tracestate") ?? undefined, + baggage: response.headers.get("baggage") ?? undefined, + body: signal ? abortable(body, signal) : body, + }; +} + +async function invokeRuntimeWithIam( + deps: InvokeRuntimeDeps, + request: RuntimeInvokeRequest, + options: CoreOptions, + signal?: AbortSignal, +): Promise { + const { runtimeId, applicationHeaders, bearerToken: _bearerToken, ...input } = request; + const command = new InvokeAgentRuntimeCommand({ ...input, agentRuntimeArn: runtimeId }); + if (applicationHeaders?.length) { + command.middlewareStack.add( + (next) => async (args) => { + const sdkRequest = args.request as { headers: Record }; + for (const [name, value] of applicationHeaders) sdkRequest.headers[name] = value; + return next(args); + }, + { step: "build", name: "runtimeApplicationHeaders" }, + ); + } + let response; + try { + response = await deps.clients.data(toClientConfig(options)).send(command, { + abortSignal: signal, + }); + } catch (error) { + if (signal?.aborted) throw signal.reason ?? error; + throw error; + } + + const body = (response.response as AsyncIterable | undefined) ?? emptyBody(); + return { + statusCode: response.statusCode ?? 0, + contentType: response.contentType ?? "", + runtimeSessionId: response.runtimeSessionId, + mcpSessionId: response.mcpSessionId, + mcpProtocolVersion: response.mcpProtocolVersion, + traceId: response.traceId, + traceParent: response.traceParent, + traceState: response.traceState, + baggage: response.baggage, + body: signal ? abortable(body, signal) : body, + }; +} diff --git a/src/core/runtime.tsx b/src/core/runtime.tsx index 5de368453..4281f97d1 100644 --- a/src/core/runtime.tsx +++ b/src/core/runtime.tsx @@ -1,4 +1,3 @@ -import { randomUUID } from "node:crypto"; import { GetAgentRuntimeCommand, GetAgentRuntimeEndpointCommand, @@ -11,7 +10,6 @@ import { type ListAgentRuntimesResponse, type ListAgentRuntimeVersionsResponse, } from "@aws-sdk/client-bedrock-agentcore-control"; -import { InvokeAgentRuntimeCommand } from "@aws-sdk/client-bedrock-agentcore"; import type { CoreRuntimeClient, RuntimeInvokeRequest, @@ -19,11 +17,9 @@ import type { } from "../handlers/runtime/types"; import type { Logger } from "../logging"; import type { AwsClients, CoreFetch, CoreOptions } from "./types"; -import { abortable } from "./abortable"; +import { invokeRuntime } from "./invokeRuntime"; import { toClientConfig } from "./utils"; -async function* emptyBody(): AsyncGenerator {} - export class RuntimeClient implements CoreRuntimeClient { constructor( private readonly clients: AwsClients, @@ -31,158 +27,20 @@ export class RuntimeClient implements CoreRuntimeClient { private readonly logger: Logger, ) {} - async invokeRuntime( - request: RuntimeInvokeRequest, - options: CoreOptions, - signal?: AbortSignal, - ): Promise { - const { runtimeId, bearerToken } = request; - if (bearerToken !== undefined) { - const logger = this.logger.child({ - operation: "invokeRuntime", - authMode: "CUSTOM_JWT", - runtimeId, - qualifier: request.qualifier, - region: options.region, - }); - return this.invokeRuntimeWithCustomJwt(request, bearerToken, options, logger, signal); - } - return this.invokeRuntimeWithIam(request, options, signal); - } - - private async invokeRuntimeWithCustomJwt( + // invokeRuntime delegates to the free function so EvalClient.invokeDataset can reuse + // the same invoke logic without holding a RuntimeClient (both call it off their + // own clients/fetch/logger). + invokeRuntime( request: RuntimeInvokeRequest, - bearerToken: string, options: CoreOptions, - logger: Logger, signal?: AbortSignal, ): Promise { - const client = this.clients.data(toClientConfig(options)); - const endpoint = client.config.endpointProvider({ - Region: options.region, - Endpoint: options.endpointUrl, - }); - const url = new URL(endpoint.url); - if (url.protocol !== "https:") { - throw new TypeError("CUSTOM_JWT requires an HTTPS endpoint"); - } - url.pathname = `${url.pathname.replace(/\/?$/, "/")}runtimes/${encodeURIComponent(request.runtimeId)}/invocations`; - url.search = new URLSearchParams({ - accountId: request.accountId, - qualifier: request.qualifier, - }).toString(); - const headers = new Headers(request.applicationHeaders); - try { - headers.set("Authorization", `Bearer ${bearerToken}`); - } catch { - throw new TypeError("Invalid bearer token"); - } - try { - for (const [name, value] of [ - ["Content-Type", request.contentType], - ["Accept", request.accept], - ["Mcp-Session-Id", request.mcpSessionId], - ["X-Amzn-Bedrock-AgentCore-Runtime-Session-Id", request.runtimeSessionId ?? randomUUID()], - ["Mcp-Protocol-Version", request.mcpProtocolVersion], - ["Mcp-Method", request.mcpMethod], - ["Mcp-Name", request.mcpName], - ["X-Amzn-Bedrock-AgentCore-Runtime-User-Id", request.runtimeUserId], - ["X-Amzn-Trace-Id", request.traceId], - ["traceparent", request.traceParent], - ["tracestate", request.traceState], - ["baggage", request.baggage], - ] as const) { - if (value !== undefined) headers.set(name, value); - } - } catch { - throw new TypeError("Invalid Runtime request header"); - } - let response: Response; - try { - response = await this.fetch(url, { - method: "POST", - redirect: "error", - headers, - body: request.payload as RequestInit["body"], - signal, - }); - } catch (error) { - if (signal?.aborted) throw signal.reason ?? error; - logger - .child({ - errorName: - error instanceof TypeError - ? "TypeError" - : error instanceof Error - ? "Error" - : typeof error, - }) - .debug("Runtime invocation transport failed"); - throw new Error("Runtime invocation failed"); - } - if (!response.ok) { - logger - .child({ httpStatusCode: response.status }) - .debug("Runtime invocation returned a non-success response"); - await response.body?.cancel().catch(() => undefined); - throw new Error(`HTTP ${response.status}`); - } - const body = (response.body as AsyncIterable | null) ?? emptyBody(); - return { - statusCode: response.status, - contentType: response.headers.get("content-type") ?? "", - runtimeSessionId: - response.headers.get("x-amzn-bedrock-agentcore-runtime-session-id") ?? undefined, - mcpSessionId: response.headers.get("mcp-session-id") ?? undefined, - mcpProtocolVersion: response.headers.get("mcp-protocol-version") ?? undefined, - traceId: response.headers.get("x-amzn-trace-id") ?? undefined, - traceParent: response.headers.get("traceparent") ?? undefined, - traceState: response.headers.get("tracestate") ?? undefined, - baggage: response.headers.get("baggage") ?? undefined, - body: signal ? abortable(body, signal) : body, - }; - } - - private async invokeRuntimeWithIam( - request: RuntimeInvokeRequest, - options: CoreOptions, - signal?: AbortSignal, - ): Promise { - const { runtimeId, applicationHeaders, bearerToken: _bearerToken, ...input } = request; - const command = new InvokeAgentRuntimeCommand({ ...input, agentRuntimeArn: runtimeId }); - if (applicationHeaders?.length) { - command.middlewareStack.add( - (next) => async (args) => { - const sdkRequest = args.request as { headers: Record }; - for (const [name, value] of applicationHeaders) sdkRequest.headers[name] = value; - return next(args); - }, - { step: "build", name: "runtimeApplicationHeaders" }, - ); - } - let response; - try { - response = await this.clients.data(toClientConfig(options)).send(command, { - abortSignal: signal, - }); - } catch (error) { - if (signal?.aborted) throw signal.reason ?? error; - throw error; - } - - const body = (response.response as AsyncIterable | undefined) ?? emptyBody(); - return { - statusCode: response.statusCode ?? 0, - contentType: response.contentType ?? "", - runtimeSessionId: response.runtimeSessionId, - mcpSessionId: response.mcpSessionId, - mcpProtocolVersion: response.mcpProtocolVersion, - traceId: response.traceId, - traceParent: response.traceParent, - traceState: response.traceState, - baggage: response.baggage, - body: signal ? abortable(body, signal) : body, - }; + return invokeRuntime( + { clients: this.clients, fetch: this.fetch, logger: this.logger }, + request, + options, + signal, + ); } async getRuntime( diff --git a/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx b/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx index 9eb348b21..b710f529c 100644 --- a/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx +++ b/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx @@ -60,7 +60,7 @@ describe("eval batch-evaluation command hierarchy", () => { .find((c) => c.name() === "eval") ?.children() .find((c) => c.name() === "batch-evaluation"); - expect(group?.children().map((c) => c.name())).toEqual(["evaluate", "get", "list"]); + expect(group?.children().map((c) => c.name())).toEqual(["evaluate", "simulate", "get", "list"]); }); test("prints help for `eval batch-evaluation --json` without an SDK call", async () => { diff --git a/src/handlers/eval/batch-evaluation/index.tsx b/src/handlers/eval/batch-evaluation/index.tsx index da96049ad..3d865cb7c 100644 --- a/src/handlers/eval/batch-evaluation/index.tsx +++ b/src/handlers/eval/batch-evaluation/index.tsx @@ -6,6 +6,7 @@ import type { Core } from "../../types"; import { createGetBatchEvaluationHandler } from "./get"; import { createListBatchEvaluationsHandler } from "./list"; import { createEvaluateBatchEvaluationHandler } from "./evaluate"; +import { createSimulateBatchEvaluationHandler } from "./simulate"; // batch-evaluation supports evaluate (start an async job) plus get + list. A bare // invocation opens the interactive TUI (list → get), matching evaluator and @@ -15,6 +16,7 @@ export function createBatchEvaluationHandler(core: Core, io: AppIO): Router { .use(withTuiOnEmptyFlagsAndArgs(core, io)) .default(renderTui(core, io)) .handler(createEvaluateBatchEvaluationHandler(core, io)) + .handler(createSimulateBatchEvaluationHandler(core, io)) .handler(createGetBatchEvaluationHandler(core, io)) .handler(createListBatchEvaluationsHandler(core)); } diff --git a/src/handlers/eval/batch-evaluation/simulate/__snapshots__/simulate.test.tsx.snap b/src/handlers/eval/batch-evaluation/simulate/__snapshots__/simulate.test.tsx.snap new file mode 100644 index 000000000..ddc632555 --- /dev/null +++ b/src/handlers/eval/batch-evaluation/simulate/__snapshots__/simulate.test.tsx.snap @@ -0,0 +1,23 @@ +// Bun Snapshot v1, https://bun.sh/docs/test/snapshots + +exports[`eval batch-evaluation simulate builds the sessionMetadata ground-truth shape [golden] 1`] = ` +[ + { + "groundTruth": { + "inline": { + "assertions": [ + { + "text": "polite", + }, + ], + }, + }, + "sessionId": "s1", + "testScenarioId": "e1", + }, + { + "sessionId": "s2", + "testScenarioId": "e2", + }, +] +`; diff --git a/src/handlers/eval/batch-evaluation/simulate/index.tsx b/src/handlers/eval/batch-evaluation/simulate/index.tsx new file mode 100644 index 000000000..ad6f5c4d0 --- /dev/null +++ b/src/handlers/eval/batch-evaluation/simulate/index.tsx @@ -0,0 +1,118 @@ +import z from "zod"; +import { createHandler, flag } from "../../../../router"; +import { InputValidationError } from "../../../../errors"; +import { JsonRendererKey } from "../../../../tui"; +import type { AppIO } from "../../../../io"; +import type { Core } from "../../../types"; +import { coreOptsFromCtx } from "../../../utils"; +import { parseRuntimeInvokeHeaders } from "../../../runtime/invoke/request"; + +// Composes invokeDataset (replay) → startBatchEvaluation (grade). Invoke flags mirror +// `runtime invoke`. +export const createSimulateBatchEvaluationHandler = (core: Core, _io: AppIO) => + createHandler({ + name: "simulate", + description: "replay a dataset against a runtime, then batch-evaluate the resulting sessions", + flags: [ + flag("runtime-id", "runtime id to invoke per scenario", z.string().optional()), + flag("qualifier", "runtime endpoint qualifier (default DEFAULT)", z.string().optional()), + flag( + "payload-template", + 'JSON payload template; {input} is the scenario input, e.g. {"prompt":"{input}"}', + z.string().optional(), + ), + flag("header", "an ordered application header (repeatable)", z.array(z.string()).optional()), + flag( + "bearer-token", + "CUSTOM_JWT bearer token (for JWT-auth runtimes)", + z.string().optional(), + ), + flag("user-id", "runtime user id", z.string().optional()), + flag("dataset", "dataset source: local JSONL path or a dataset id", z.string().optional()), + flag("dataset-version", "dataset version (with a dataset id)", z.string().optional()), + flag("evaluator", "evaluator id(s) to apply", z.array(z.string()).optional()), + flag("name", "batch evaluation name (unique in the account)", z.string().optional()), + flag("description", "optional description", z.string().optional()), + flag("kms-key-arn", "KMS key to encrypt evaluation data at rest", z.string().optional()), + ], + handle: async (ctx, flags) => { + if (!flags["runtime-id"]) + throw new InputValidationError("required option '--runtime-id' not specified"); + if (!flags["payload-template"]) { + throw new InputValidationError("required option '--payload-template' not specified"); + } + if (!flags["dataset"]) + throw new InputValidationError("required option '--dataset' not specified"); + if (!flags["evaluator"]?.length) { + throw new InputValidationError( + "required option '--evaluator ' not specified", + ); + } + if (!flags["name"]) + throw new InputValidationError("required option '--name ' not specified"); + + // Ctrl-C aborts the run (invokes, the ingestion wait, the dataset download). + const controller = new AbortController(); + const interrupt = () => controller.abort(); + process.once("SIGINT", interrupt); + try { + const opts = coreOptsFromCtx(ctx); + + const r = await core.eval.invokeDataset( + { + runtimeId: flags["runtime-id"], + qualifier: flags["qualifier"], + payloadTemplate: flags["payload-template"], + headers: parseRuntimeInvokeHeaders(flags["header"]), + bearerToken: flags["bearer-token"], + userId: flags["user-id"], + dataset: flags["dataset"], + datasetVersion: flags["dataset-version"], + }, + opts, + controller.signal, + ); + if (r.invoked === 0) { + const detail = r.firstError ? `; first error: ${r.firstError.message}` : ""; + throw new InputValidationError( + `no examples could be invoked (${r.failed} failed) — nothing to evaluate${detail}`, + ); + } + + // The example's neutral ground truth crosses over as sessionMetadata (inline arm). + const job = await core.eval.startBatchEvaluation( + { + name: flags["name"], + description: flags["description"], + evaluatorIds: flags["evaluator"], + source: { + origin: "agent", + agent: flags["runtime-id"], + endpoint: flags["qualifier"], + sessionIds: r.sessions.map((s) => s.sessionId), + }, + groundTruth: r.sessions.map((s) => ({ + sessionId: s.sessionId, + testScenarioId: s.exampleId, + ...(s.groundTruth && { groundTruth: { inline: s.groundTruth } }), + })), + kmsKeyArn: flags["kms-key-arn"], + }, + opts, + ); + + ctx.require(JsonRendererKey).renderJson({ + batchEvaluationId: job.batchEvaluationId, + status: job.status, + examplesInvoked: r.invoked, + examplesFailed: r.failed, + }); + } catch (error) { + // A Ctrl-C exits quietly; the half-created sessions grade nothing. + if (controller.signal.aborted) return; + throw error; + } finally { + process.off("SIGINT", interrupt); + } + }, + }); diff --git a/src/handlers/eval/batch-evaluation/simulate/simulate.test.tsx b/src/handlers/eval/batch-evaluation/simulate/simulate.test.tsx new file mode 100644 index 000000000..2375a3b3f --- /dev/null +++ b/src/handlers/eval/batch-evaluation/simulate/simulate.test.tsx @@ -0,0 +1,182 @@ +import { test, expect, describe } from "bun:test"; +import { createRootHandler } from "../../../index"; +import { + createSilentLogger, + TestCoreClient, + testIO, + TestGlobalConfigAccessor, +} from "../../../../testing"; +import type { InvokeDatasetResult } from "../../types"; + +// Two invoked sessions; the handler feeds these into startBatchEvaluation and renders +// the job it returns (DEFAULT_START_BATCH_EVAL_RESPONSE: batch-eval-test / RUNNING). +const INVOKE_RESULT: InvokeDatasetResult = { + sessions: [ + { exampleId: "e1", sessionId: "s1", groundTruth: { assertions: [{ text: "polite" }] } }, + { exampleId: "e2", sessionId: "s2" }, + ], + invoked: 2, + failed: 0, +}; + +async function run(args: string[], configure?: (core: TestCoreClient) => void) { + const core = new TestCoreClient(); + core.eval.setInvokeDatasetResponse(INVOKE_RESULT); + configure?.(core); + const io = testIO(); + const root = createRootHandler(core, { + io: io.io, + logger: createSilentLogger(), + globalConfigAccessor: new TestGlobalConfigAccessor(), + }); + await root.route(["node", "agentcore", ...args, "--region", "us-west-2"]); + return { core, stdout: io.stdout() }; +} + +const BASE = [ + "eval", + "batch-evaluation", + "simulate", + "--runtime-id", + "r-1", + "--payload-template", + '{"prompt":"{input}"}', + "--dataset", + "/tmp/ds.jsonl", + "--evaluator", + "Builtin.Helpfulness", + "--name", + "sim-1", +]; + +describe("eval batch-evaluation simulate", () => { + test("registered under batch-evaluation", () => { + const io = testIO(); + const root = createRootHandler(new TestCoreClient(), { + io: io.io, + logger: createSilentLogger(), + globalConfigAccessor: new TestGlobalConfigAccessor(), + }); + const group = root + .children() + .find((c) => c.name() === "eval") + ?.children() + .find((c) => c.name() === "batch-evaluation"); + expect(group?.children().map((c) => c.name())).toContain("simulate"); + }); + + test.each([ + [ + [ + "--payload-template", + '{"prompt":"{input}"}', + "--dataset", + "/tmp/ds.jsonl", + "--evaluator", + "E", + "--name", + "n", + ], + /--runtime-id/, + ], + [ + ["--runtime-id", "r-1", "--dataset", "/tmp/ds.jsonl", "--evaluator", "E", "--name", "n"], + /--payload-template/, + ], + [ + ["--runtime-id", "r-1", "--payload-template", "{}", "--evaluator", "E", "--name", "n"], + /--dataset/, + ], + [ + [ + "--runtime-id", + "r-1", + "--payload-template", + "{}", + "--dataset", + "/tmp/ds.jsonl", + "--name", + "n", + ], + /--evaluator/, + ], + [ + [ + "--runtime-id", + "r-1", + "--payload-template", + "{}", + "--dataset", + "/tmp/ds.jsonl", + "--evaluator", + "E", + ], + /--name/, + ], + ])("rejects missing required flag", async (args, expected) => { + await expect(run(["eval", "batch-evaluation", "simulate", ...args])).rejects.toThrow(expected); + }); + + test("passes runtime-level flags to invokeDataset (no evaluator/name leak)", async () => { + const { core } = await run([...BASE, "--qualifier", "PROD", "--header", "x-a:1"]); + const call = core.eval.calls.find((c) => c.method === "invokeDataset"); + expect(call?.args[0]).toMatchObject({ + runtimeId: "r-1", + qualifier: "PROD", + payloadTemplate: '{"prompt":"{input}"}', + headers: [["x-a", "1"]], + dataset: "/tmp/ds.jsonl", + }); + // Grader-only flags are NOT part of the invokeDataset (runtime-level) input. + expect(call?.args[0]).not.toHaveProperty("evaluatorIds"); + expect(call?.args[0]).not.toHaveProperty("name"); + // Handler wires an AbortSignal (Ctrl-C) through to Core. + expect(call?.args[2]).toBeInstanceOf(AbortSignal); + }); + + test("composes startBatchEvaluation over the created sessions + wrapped ground truth", async () => { + const { core, stdout } = await run(BASE); + + // Rendered output is the batch job + invoked/failed counts. + expect(JSON.parse(stdout)).toEqual({ + batchEvaluationId: "batch-eval-test", + status: "RUNNING", + examplesInvoked: 2, + examplesFailed: 0, + }); + + const start = core.eval.calls.find((c) => c.method === "startBatchEvaluation"); + expect(start?.args[0]).toMatchObject({ + name: "sim-1", + evaluatorIds: ["Builtin.Helpfulness"], + source: { origin: "agent", agent: "r-1", sessionIds: ["s1", "s2"] }, + // e1's inline GT is wrapped; e2 (no GT) omits the member. + groundTruth: [ + { + sessionId: "s1", + testScenarioId: "e1", + groundTruth: { inline: { assertions: [{ text: "polite" }] } }, + }, + { sessionId: "s2", testScenarioId: "e2" }, + ], + }); + }); + + // Golden: the exact evaluationMetadata (sessionMetadata) the handler builds from the + // invoked sessions. Locks the `{ inline: gt }` wrapping and the omitted-member case for + // a session with no ground truth — the wire shape the batch service reads. + test("builds the sessionMetadata ground-truth shape [golden]", async () => { + const { core } = await run(BASE); + const start = core.eval.calls.find((c) => c.method === "startBatchEvaluation"); + const input = start!.args[0] as { groundTruth: unknown }; + expect(input.groundTruth).toMatchSnapshot(); + }); + + test("refuses to grade when nothing was invoked", async () => { + await expect( + run(BASE, (core) => + core.eval.setInvokeDatasetResponse({ sessions: [], invoked: 0, failed: 3 }), + ), + ).rejects.toThrow(/no examples could be invoked \(3 failed\)/); + }); +}); diff --git a/src/handlers/eval/types.tsx b/src/handlers/eval/types.tsx index 278f8f34c..9ed9870ff 100644 --- a/src/handlers/eval/types.tsx +++ b/src/handlers/eval/types.tsx @@ -34,6 +34,7 @@ import type { ListBatchEvaluationsResponse, StartBatchEvaluationResponse, SessionMetadataShape, + InlineGroundTruth, EvaluationReferenceInput, EvaluationResultContent, DataSourceConfig as DataPlaneDataSourceConfig, @@ -205,6 +206,39 @@ export type StartBatchEvaluationInput = { kmsKeyArn?: string; }; +// InvokeDatasetInput is the runtime-level shape for replaying a dataset: invoke each +// example against the runtime, one client-generated session per example. Runtime fields +// only — no evaluator/name/kms (those belong to the grader the handler composes on top, +// e.g. startBatchEvaluation). Invoke fields mirror `runtime invoke`. +export type InvokeDatasetInput = { + runtimeId: string; + qualifier?: string; + payloadTemplate: string; // e.g. {"prompt":"{input}"} — {input} is the example's turn input + headers?: [string, string][]; + bearerToken?: string; + userId?: string; + dataset: string; // local JSONL path or a dataset id + datasetVersion?: string; +}; + +// InvokedSession is one replayed example: the session created for it plus its neutral +// ground truth. Grader-agnostic — the batch handler wraps `groundTruth` as +// SessionMetadataShape; a future ondemand handler adapts it to EvaluationReferenceInput. +export type InvokedSession = { + exampleId: string; + sessionId: string; + groundTruth?: InlineGroundTruth; +}; + +// InvokeDatasetResult reports the created sessions plus how many examples were invoked +// vs dropped (a failed invoke is skipped, not fatal). firstError explains a total failure. +export type InvokeDatasetResult = { + sessions: InvokedSession[]; + invoked: number; + failed: number; + firstError?: Error; +}; + export type SpanRecord = Record; export type SessionTrace = { @@ -310,6 +344,14 @@ export interface CoreEvalClient { // Evaluate API and returns per-session scores. No job, no CloudWatch — the // trace read happened in getTracesForAgent. evaluate(input: EvaluateInput, options: CoreOptions): Promise; + // invokeDataset replays a dataset against the runtime (invoke per example, client-side, + // one session each) and returns the created sessions + neutral ground truth. Grader- + // agnostic: the handler composes it with startBatchEvaluation (or, later, evaluate). + invokeDataset( + input: InvokeDatasetInput, + options: CoreOptions, + signal?: AbortSignal, + ): Promise; createOnlineEvaluationConfig( input: CreateOnlineEvalInput, diff --git a/src/io/index.ts b/src/io/index.ts index fb0da0b0e..bcd632f3e 100644 --- a/src/io/index.ts +++ b/src/io/index.ts @@ -26,6 +26,7 @@ export { type JsonValue, } from "./jsonl"; export { SourceResolver, type SourceResolverConfig } from "./source"; +export { renderJsonTemplate } from "./template"; export { classifyStreamingResponse, writeStreamingResponse, diff --git a/src/io/template.test.ts b/src/io/template.test.ts new file mode 100644 index 000000000..6be6da2de --- /dev/null +++ b/src/io/template.test.ts @@ -0,0 +1,32 @@ +import { test, expect, describe } from "bun:test"; +import { renderJsonTemplate } from "./template"; + +const decode = (bytes: Uint8Array) => new TextDecoder().decode(bytes); + +describe("renderJsonTemplate", () => { + test("substitutes {input} inside a string value", () => { + const out = decode(renderJsonTemplate('{"prompt":"{input}"}', { input: "hello" })); + expect(JSON.parse(out)).toEqual({ prompt: "hello" }); + }); + + test("JSON-escapes substituted values (quotes, newlines don't break the payload)", () => { + const out = renderJsonTemplate('{"prompt":"{input}"}', { input: 'a "quote"\nline' }); + expect(JSON.parse(decode(out))).toEqual({ prompt: 'a "quote"\nline' }); + }); + + test("substitutes nested + array positions", () => { + const out = renderJsonTemplate('{"messages":[{"role":"user","content":"{input}"}]}', { + input: "hi", + }); + expect(JSON.parse(decode(out))).toEqual({ messages: [{ role: "user", content: "hi" }] }); + }); + + test("supports arbitrary keys, leaves unknown placeholders intact", () => { + const out = renderJsonTemplate('{"m":"{model}","p":"{input}"}', { input: "x" }); + expect(JSON.parse(decode(out))).toEqual({ m: "{model}", p: "x" }); + }); + + test("rejects invalid JSON template", () => { + expect(() => renderJsonTemplate("{not json", { input: "x" })).toThrow(/valid JSON/); + }); +}); diff --git a/src/io/template.ts b/src/io/template.ts new file mode 100644 index 000000000..e7d12cca8 --- /dev/null +++ b/src/io/template.ts @@ -0,0 +1,33 @@ +import { InputValidationError } from "../errors"; + +// renderJsonTemplate substitutes `{key}` placeholders inside the string values of a +// JSON template and returns the encoded bytes. General on purpose: `simulate` uses +// `{ input }`, but any `{model}`/`{sessionId}` a future caller adds works the same. +// Parsing the template first (rather than string-replacing raw) keeps the result +// valid JSON regardless of quotes/newlines in the substituted values. +export function renderJsonTemplate( + template: string, + values: Record, + flagName = "payload-template", +): Uint8Array { + let parsed: unknown; + try { + parsed = JSON.parse(template); + } catch { + throw new InputValidationError(`--${flagName} must be valid JSON`); + } + return new TextEncoder().encode(JSON.stringify(substitute(parsed, values))); +} + +function substitute(value: unknown, values: Record): unknown { + if (typeof value === "string") { + return value.replace(/\{(\w+)\}/g, (match, key) => values[key] ?? match); + } + if (Array.isArray(value)) return value.map((item) => substitute(item, values)); + if (value && typeof value === "object") { + return Object.fromEntries( + Object.entries(value).map(([key, item]) => [key, substitute(item, values)]), + ); + } + return value; +} diff --git a/src/testing/TestCoreClient.tsx b/src/testing/TestCoreClient.tsx index 50047db83..043536bd9 100644 --- a/src/testing/TestCoreClient.tsx +++ b/src/testing/TestCoreClient.tsx @@ -133,6 +133,8 @@ import type { DatasetUpdateProgressEvent, EvaluateInput, EvaluateResult, + InvokeDatasetInput, + InvokeDatasetResult, GetBatchEvaluationResult, GetTracesInput, LlmAsAJudgeUpdate, @@ -1405,6 +1407,11 @@ export class TestEvalClient implements CoreEvalClient { sessionsEvaluated: 0, results: [], }; + private invokeDatasetResponse: InvokeDatasetResult = { + sessions: [], + invoked: 0, + failed: 0, + }; private error?: Error; // setListResponse sets what listEvaluators resolves to (when not erroring). @@ -1724,6 +1731,22 @@ export class TestEvalClient implements CoreEvalClient { return this.evaluateResponse; } + // setInvokeDatasetResponse sets what invokeDataset resolves to (when not erroring). + setInvokeDatasetResponse(response: InvokeDatasetResult): this { + this.invokeDatasetResponse = response; + return this; + } + + async invokeDataset( + input: InvokeDatasetInput, + options: CoreOptions, + signal?: AbortSignal, + ): Promise { + this.calls.push({ method: "invokeDataset", args: [input, options, signal] }); + if (this.error) throw this.error; + return this.invokeDatasetResponse; + } + async createOnlineEvaluationConfig( input: CreateOnlineEvalInput, options: CoreOptions,