diff --git a/src/core/eval.tsx b/src/core/eval.tsx index 849054732..fcf61ecb3 100644 --- a/src/core/eval.tsx +++ b/src/core/eval.tsx @@ -142,6 +142,7 @@ import { } from "./onlineEvalExecutionRole"; const DEFAULT_ENDPOINT_QUALIFIER = "DEFAULT"; +const DEFAULT_INGESTION_WAIT_MS = 180_000; const DATASET_EXAMPLES_BATCH_LIMIT = 1000; const DATASET_MUTATION_PAYLOAD_LIMIT_BYTES = 5 * 1024 * 1024; const DATASET_ACTIVE_TIMEOUT_MS = 60_000; @@ -188,6 +189,7 @@ export class EvalClient implements CoreEvalClient { private readonly fetch: CoreFetch = globalThis.fetch, // logger for batch-evaluation result-log diagnostics private readonly logger: Logger = noopLogger, + private readonly newSessionId: () => string = randomUUID, ) {} async createEvaluator( @@ -624,10 +626,10 @@ export class EvalClient implements CoreEvalClient { 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) => { + const { ok, failures } = 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 sessionId = this.newSessionId(); const ctx: RunContext = { invokeOnce: async (payload) => { const response = await invokeRuntime( @@ -655,27 +657,22 @@ export class EvalClient implements CoreEvalClient { return { text }; }, }; - try { - const groundTruth = await example.run(ctx); - return { exampleId: example.exampleId, sessionId, groundTruth }; - } catch (error) { - // Enrich with the example identity so the dropped-invoke reason is self-describing - // in firstError, instead of a bare transport message logged separately. - const cause = error instanceof Error ? error : new Error(String(error)); - throw new Error( - `example "${example.exampleId}" (${example.schemaType}) failed to invoke: ${cause.message}`, - { cause }, - ); - } + const groundTruth = await example.run(ctx); + return { exampleId: example.exampleId, sessionId, groundTruth }; }); - if (failed > 0) { - this.logger.warn(`invokeDataset: ${failed} example(s) failed to invoke and were dropped`); + const invokeFailures = failures.map((f) => ({ + exampleId: f.item.exampleId, + error: f.error.message, + })); + if (invokeFailures.length > 0) { + this.logger.warn( + `invokeDataset: ${invokeFailures.length} example(s) failed to invoke and were dropped` + + `; first: ${invokeFailures[0]!.exampleId} — ${invokeFailures[0]!.error}`, + ); } - // 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); + const waitMs = input.waitIngestionMs ?? DEFAULT_INGESTION_WAIT_MS; if (ok.length > 0 && waitMs > 0) { this.logger.info( `waiting ${Math.round(waitMs / 1000)}s for span ingestion before evaluating`, @@ -683,7 +680,12 @@ export class EvalClient implements CoreEvalClient { await sleep(waitMs, undefined, { signal }); } - return { sessions: ok, invoked: ok.length, failed, firstError }; + return { + sessions: ok, + invoked: ok.length, + failed: invokeFailures.length, + failures: invokeFailures, + }; } // Resolve a dataset ref to JSONL text: a local path directly, else download the id to a diff --git a/src/core/eval/invokeDataset/__snapshots__/invokeDataset.test.ts.snap b/src/core/eval/invokeDataset/__snapshots__/invokeDataset.test.ts.snap deleted file mode 100644 index c7a35def0..000000000 --- a/src/core/eval/invokeDataset/__snapshots__/invokeDataset.test.ts.snap +++ /dev/null @@ -1,172 +0,0 @@ -// 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/invokeDataset.test.ts b/src/core/eval/invokeDataset/invokeDataset.test.ts deleted file mode 100644 index 5247ed431..000000000 --- a/src/core/eval/invokeDataset/invokeDataset.test.ts +++ /dev/null @@ -1,263 +0,0 @@ -// 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/, - }, - { - name: "non-array assertions", - jsonl: row({ example_id: "x", turns: [{ input: "a" }], assertions: "nope" }), - error: /assertions must be an array of strings/, - }, - { - name: "non-array expected_trajectory", - jsonl: row({ example_id: "x", turns: [{ input: "a" }], expected_trajectory: "nope" }), - error: /expected_trajectory must be an array of strings/, - }, -]; - -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("rejects a payload-template without the {input} placeholder before invoking", async () => { - const { clients, payloads } = fakeClients(); - const fetch = (() => { - throw new Error("unused"); - }) as unknown as CoreFetch; - await expect( - new EvalClient(clients, fetch).invokeDataset( - { - runtimeId: "rt-1", - payloadTemplate: '{"prompt":"static"}', - dataset: datasetFile(row({ example_id: "x", turns: [{ input: "hi" }] })), - }, - OPTIONS, - ), - ).rejects.toThrow(/\{input\} placeholder/); - expect(payloads).toEqual([]); - }); - - 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/run.ts b/src/core/eval/invokeDataset/run.ts index 6f82a98b9..5304d09f0 100644 --- a/src/core/eval/invokeDataset/run.ts +++ b/src/core/eval/invokeDataset/run.ts @@ -1,15 +1,15 @@ -// 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 type ExampleRun = { + ok: Result[]; + failures: { item: Item; error: Error }[]; +}; export async function runExamples( items: Item[], worker: (item: Item) => Promise, concurrency = 5, -): Promise> { +): Promise> { const ok: Result[] = []; - let failed = 0; - let firstError: Error | undefined; + const failures: { item: Item; error: Error }[] = []; let next = 0; const run = async (): Promise => { while (next < items.length) { @@ -17,11 +17,10 @@ export async function runExamples( try { ok.push(await worker(item)); } catch (error) { - failed++; - if (!firstError) firstError = error instanceof Error ? error : new Error(String(error)); + failures.push({ item, error: error instanceof Error ? error : new Error(String(error)) }); } } }; await Promise.all(Array.from({ length: Math.min(concurrency, items.length) }, run)); - return { ok, failed, firstError }; + return { ok, failures }; } diff --git a/src/core/index.tsx b/src/core/index.tsx index 34b9b4795..d7275b912 100644 --- a/src/core/index.tsx +++ b/src/core/index.tsx @@ -38,6 +38,7 @@ type CoreClientConfig = { createLogsClient: CreateLogsClient; logger: Logger; fetch?: CoreFetch; + newSessionId?: () => string; }; // CoreClient is the single entry point to the Bedrock AgentCore APIs. It owns the @@ -78,7 +79,12 @@ export class CoreClient implements AwsClients { // EvalClient shares the injected fetch: dataset content is served from a // presigned S3 URL, outside the SDK seam the other operations use. The logger // is used for batch-evaluation result-log diagnostics. - this.eval = new EvalClient(this, fetch, this.logger.child({ module: "eval" })); + this.eval = new EvalClient( + this, + fetch, + this.logger.child({ module: "eval" }), + config.newSessionId, + ); this.projectManager = new FsProjectManager({ logger: this.logger.child({ module: "projectManager" }), diff --git a/src/handlers/eval/batch-evaluation/__fixtures__/InvokeAgentRuntimeCommand.46bef739bc7ae9c1.json b/src/handlers/eval/batch-evaluation/__fixtures__/InvokeAgentRuntimeCommand.46bef739bc7ae9c1.json new file mode 100644 index 000000000..aa033fb90 --- /dev/null +++ b/src/handlers/eval/batch-evaluation/__fixtures__/InvokeAgentRuntimeCommand.46bef739bc7ae9c1.json @@ -0,0 +1,8 @@ +{ + "contentType": "text/event-stream; charset=utf-8", + "runtimeSessionId": "00000000-0000-4000-8000-000000000001", + "response": { + "$stream": "data: \"Hello! How can I help you today\"\n\ndata: \"?\"\n\n" + }, + "statusCode": 200 +} \ No newline at end of file diff --git a/src/handlers/eval/batch-evaluation/__fixtures__/StartBatchEvaluationCommand.9c92cd5b227a329e.json b/src/handlers/eval/batch-evaluation/__fixtures__/StartBatchEvaluationCommand.9c92cd5b227a329e.json new file mode 100644 index 000000000..1aaf75008 --- /dev/null +++ b/src/handlers/eval/batch-evaluation/__fixtures__/StartBatchEvaluationCommand.9c92cd5b227a329e.json @@ -0,0 +1,20 @@ +{ + "batchEvaluationId": "golden_batch_simulate_fixture1-86183b0ccc", + "batchEvaluationArn": "arn:aws:bedrock-agentcore:us-west-2:685197708687:batch-evaluate/golden_batch_simulate_fixture1-86183b0ccc", + "batchEvaluationName": "golden_batch_simulate_fixture1", + "status": "PENDING", + "createdAt": { + "$date": "2026-08-26T22:07:26.049Z" + }, + "evaluators": [ + { + "evaluatorId": "Builtin.Helpfulness" + } + ], + "outputConfig": { + "cloudWatchConfig": { + "logGroupName": "/aws/bedrock-agentcore/evaluations/batch-evaluations/results/default", + "logStreamName": "run-golden_batch_simulate_fixture1-86183b0ccc" + } + } +} \ No newline at end of file diff --git a/src/handlers/eval/batch-evaluation/__fixtures__/simulate-ds.jsonl b/src/handlers/eval/batch-evaluation/__fixtures__/simulate-ds.jsonl new file mode 100644 index 000000000..39fe36263 --- /dev/null +++ b/src/handlers/eval/batch-evaluation/__fixtures__/simulate-ds.jsonl @@ -0,0 +1 @@ +{"example_id":"e1","turns":[{"input":"hi"}],"assertions":["stays polite"]} diff --git a/src/handlers/eval/batch-evaluation/__fixtures__/simulate.golden.json b/src/handlers/eval/batch-evaluation/__fixtures__/simulate.golden.json new file mode 100644 index 000000000..841535721 --- /dev/null +++ b/src/handlers/eval/batch-evaluation/__fixtures__/simulate.golden.json @@ -0,0 +1,13 @@ +{ + "batchEvaluationId": "golden_batch_simulate_fixture1-86183b0ccc", + "status": "PENDING", + "examplesInvoked": 1, + "examplesFailed": 0, + "sessions": [ + { + "exampleId": "e1", + "sessionId": "00000000-0000-4000-8000-000000000001" + } + ], + "failures": [] +} \ No newline at end of file diff --git a/src/handlers/eval/batch-evaluation/batch-evaluation.fixture.test.tsx b/src/handlers/eval/batch-evaluation/batch-evaluation.fixture.test.tsx index 8494e9a53..034e3ac4d 100644 --- a/src/handlers/eval/batch-evaluation/batch-evaluation.fixture.test.tsx +++ b/src/handlers/eval/batch-evaluation/batch-evaluation.fixture.test.tsx @@ -42,6 +42,9 @@ const MISSING_JOB_ID = "missing-batch-eval-0000000000"; const FIXTURE_EVAL_AGENT = "asdf_MyAgent-3s5axvBC6Q"; const FIXTURE_EVAL_NAME = "golden_batch_evaluate_fixture685"; +const FIXTURE_SIMULATE_NAME = "golden_batch_simulate_fixture1"; +const FIXTURE_SIMULATE_DATASET = join(FIXTURES, "simulate-ds.jsonl"); + function createFixtureCore(): CoreClient { const { createControlClient, createDataClient, createIamClient, createLogsClient } = fixtureFactories(FIXTURES); @@ -136,4 +139,49 @@ describe("eval batch-evaluation (fixture-backed)", () => { expect(job.batchEvaluationId).toBeTruthy(); expect(job.status).toBeTruthy(); }); + + test("simulate replays a dataset, then submits a batch job over the created sessions", async () => { + let n = 0; + const { createControlClient, createDataClient, createIamClient, createLogsClient } = + fixtureFactories(FIXTURES); + const core = new CoreClient({ + createControlClient, + createDataClient, + createIamClient, + createLogsClient, + logger: createSilentLogger(), + newSessionId: () => `00000000-0000-4000-8000-${String(++n).padStart(12, "0")}`, + }); + const io = testIO(); + const root = createRootHandler(core, { + io: io.io, + logger: createSilentLogger(), + globalConfigAccessor: new TestGlobalConfigAccessor(), + }); + + await root.route([ + "node", + "agentcore", + "eval", + "batch-evaluation", + "simulate", + "--runtime-id", + FIXTURE_EVAL_AGENT, + "--payload-template", + '{"prompt":"{input}"}', + "--dataset", + FIXTURE_SIMULATE_DATASET, + "--evaluator", + "Builtin.Helpfulness", + "--name", + FIXTURE_SIMULATE_NAME, + "--ingestion-wait-ms", + "0", + "--json", + "--region", + REGION, + ]); + + matchGolden(FIXTURES, "simulate.golden.json", io.stdout()); + }, 180_000); }); diff --git a/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx b/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx index b710f529c..e76da4395 100644 --- a/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx +++ b/src/handlers/eval/batch-evaluation/batch-evaluation.test.tsx @@ -160,3 +160,96 @@ describe("eval batch-evaluation list", () => { expect(core.eval.calls[0]?.args).toEqual([undefined, 10, { region: "us-west-2" }]); }); }); + +describe("eval batch-evaluation simulate", () => { + const BASE = [ + "eval", + "batch-evaluation", + "simulate", + "--runtime-id", + "r-1", + "--payload-template", + '{"prompt":"{input}"}', + "--dataset", + "/tmp/ds.jsonl", + "--evaluator", + "Builtin.Helpfulness", + "--name", + "sim-1", + ]; + + test.each<[RegExp, string[]]>([ + [ + /--runtime-id/, + ["--payload-template", "{}", "--dataset", "/tmp/ds.jsonl", "--evaluator", "E", "--name", "n"], + ], + [ + /--payload-template/, + ["--runtime-id", "r-1", "--dataset", "/tmp/ds.jsonl", "--evaluator", "E", "--name", "n"], + ], + [ + /--dataset/, + ["--runtime-id", "r-1", "--payload-template", "{}", "--evaluator", "E", "--name", "n"], + ], + [ + /--evaluator/, + [ + "--runtime-id", + "r-1", + "--payload-template", + "{}", + "--dataset", + "/tmp/ds.jsonl", + "--name", + "n", + ], + ], + [ + /--name/, + [ + "--runtime-id", + "r-1", + "--payload-template", + "{}", + "--dataset", + "/tmp/ds.jsonl", + "--evaluator", + "E", + ], + ], + ])("rejects when a required flag is missing (%s)", async (expected, args) => { + await expect(run(["eval", "batch-evaluation", "simulate", ...args])).rejects.toThrow(expected); + }); + + test("refuses to grade when nothing was invoked, naming the first failure", async () => { + await expect( + run(BASE, (c) => + c.eval.setInvokeDatasetResponse({ + sessions: [], + invoked: 0, + failed: 3, + failures: [ + { exampleId: "e1", error: "HTTP 500" }, + { exampleId: "e2", error: "HTTP 500" }, + { exampleId: "e3", error: "HTTP 500" }, + ], + }), + ), + ).rejects.toThrow(/no examples could be invoked \(3 failed\).*first error: e1 — HTTP 500/); + }); + + test("passes --ingestion-wait-ms through to invokeDataset and renders failures", async () => { + const { core, stdout } = await run([...BASE, "--ingestion-wait-ms", "0"], (c) => + c.eval.setInvokeDatasetResponse({ + sessions: [{ exampleId: "ok1", sessionId: "s1" }], + invoked: 1, + failed: 1, + failures: [{ exampleId: "bad", error: "HTTP 500" }], + }), + ); + const invoke = core.eval.calls.find((c) => c.method === "invokeDataset"); + expect(invoke).toBeDefined(); + expect((invoke!.args[0] as { waitIngestionMs?: number }).waitIngestionMs).toBe(0); + expect(JSON.parse(stdout).failures).toEqual([{ exampleId: "bad", error: "HTTP 500" }]); + }); +}); 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 deleted file mode 100644 index ddc632555..000000000 --- a/src/handlers/eval/batch-evaluation/simulate/__snapshots__/simulate.test.tsx.snap +++ /dev/null @@ -1,23 +0,0 @@ -// 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 index 72dddd6d0..4246021fe 100644 --- a/src/handlers/eval/batch-evaluation/simulate/index.tsx +++ b/src/handlers/eval/batch-evaluation/simulate/index.tsx @@ -34,6 +34,11 @@ export const createSimulateBatchEvaluationHandler = (core: Core, _io: AppIO) => flag("name", "batch evaluation name (unique in the account)", z.string().optional()), flag("description", "description for the batch evaluation", z.string().optional()), flag("kms-key-arn", "KMS key to encrypt evaluation data at rest", z.string().optional()), + flag( + "ingestion-wait-ms", + "ms to wait for span ingestion before grading (default 180000; 0 to skip)", + z.coerce.number().int().nonnegative().optional(), + ), ], handle: async (ctx, flags) => { if (!flags["runtime-id"]) @@ -69,12 +74,14 @@ export const createSimulateBatchEvaluationHandler = (core: Core, _io: AppIO) => userId: flags["user-id"], dataset: flags["dataset"], datasetVersion: flags["dataset-version"], + waitIngestionMs: flags["ingestion-wait-ms"], }, opts, controller.signal, ); if (r.invoked === 0) { - const detail = r.firstError ? `; first error: ${r.firstError.message}` : ""; + const first = r.failures[0]; + const detail = first ? `; first error: ${first.exampleId} — ${first.error}` : ""; throw new InputValidationError( `no examples could be invoked (${r.failed} failed) — nothing to evaluate${detail}`, ); @@ -107,6 +114,8 @@ export const createSimulateBatchEvaluationHandler = (core: Core, _io: AppIO) => status: job.status, examplesInvoked: r.invoked, examplesFailed: r.failed, + sessions: r.sessions.map((s) => ({ exampleId: s.exampleId, sessionId: s.sessionId })), + failures: r.failures, }); } 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 deleted file mode 100644 index 8c6215fd9..000000000 --- a/src/handlers/eval/batch-evaluation/simulate/simulate.test.tsx +++ /dev/null @@ -1,169 +0,0 @@ -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("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" }, - ], - }); - - // Handler threads the Ctrl-C AbortSignal into the replay (invokeDataset) call. - const invoke = core.eval.calls.find((c) => c.method === "invokeDataset"); - expect(invoke?.args[2]).toBeInstanceOf(AbortSignal); - }); - - // 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 f34d4a081..14eaefd23 100644 --- a/src/handlers/eval/types.tsx +++ b/src/handlers/eval/types.tsx @@ -230,8 +230,11 @@ export type InvokeDatasetInput = { userId?: string; dataset: string; // local JSONL path or a dataset id datasetVersion?: string; + waitIngestionMs?: number; }; +export type InvokeFailure = { exampleId: string; error: 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. @@ -241,13 +244,11 @@ export type InvokedSession = { 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; + failures: InvokeFailure[]; }; export type SpanRecord = Record; diff --git a/src/testing/TestCoreClient.tsx b/src/testing/TestCoreClient.tsx index b835ded7f..2e2252f66 100644 --- a/src/testing/TestCoreClient.tsx +++ b/src/testing/TestCoreClient.tsx @@ -1416,6 +1416,7 @@ export class TestEvalClient implements CoreEvalClient { sessions: [], invoked: 0, failed: 0, + failures: [], }; private error?: Error; diff --git a/src/testing/fixtures.tsx b/src/testing/fixtures.tsx index 51b6a47a7..a6a4d39c6 100644 --- a/src/testing/fixtures.tsx +++ b/src/testing/fixtures.tsx @@ -145,6 +145,31 @@ function reviveError(tagged: TaggedError): Error { return error; } +const STREAM_TAG = "$stream"; + +async function freezeStream(response: unknown): Promise { + const stream = (response as { response?: { transformToString?: () => Promise } }) + ?.response; + if (typeof stream?.transformToString !== "function") return response; + return { + ...(response as Record), + response: { [STREAM_TAG]: await stream.transformToString() }, + }; +} + +async function* streamOf(text: string): AsyncGenerator { + yield new TextEncoder().encode(text); +} + +function reviveStream(recorded: unknown): unknown { + const stream = (recorded as { response?: Record })?.response; + if (!stream || typeof stream !== "object" || !(STREAM_TAG in stream)) return recorded; + return { + ...(recorded as Record), + response: streamOf(stream[STREAM_TAG] as string), + }; +} + // makeRecordingSend returns a `.send()` that records to / replays from `dir`. // In record mode it delegates to the real client, saves the response (or the // service error), and propagates it; otherwise it reads the fixture, failing @@ -168,8 +193,9 @@ function makeRecordingSend Promise }>( writeFileSync(path, stringify(sanitizePresignedUrls(tagged))); throw error; } - writeFileSync(path, stringify(sanitizePresignedUrls(response))); - return response; + const frozen = await freezeStream(response); + writeFileSync(path, stringify(sanitizePresignedUrls(frozen))); + return reviveStream(frozen); } if (!existsSync(path)) { @@ -180,7 +206,7 @@ function makeRecordingSend Promise }>( } const recorded = parse(readFileSync(path, "utf8")); if (isTaggedError(recorded)) throw reviveError(recorded); - return recorded; + return reviveStream(recorded); }; }