feat: wire observability client to harness and evals

src/core/batchEvaluationResults.test.ts

@@ -1,55 +1,29 @@
 import { test, expect } from "bun:test";
-import type { CloudWatchLogsClient, OutputLogEvent } from "@aws-sdk/client-cloudwatch-logs";
-import { ResultTruncationError } from "../errors";
 import { createSilentLogger } from "../testing";
 import {
   isTerminalStatus,
   parseEvaluationLogEvent,
   readEvaluationResults,
 } from "./batchEvaluationResults";
+import type { CloudWatchLogEvent } from "./observability/index";
 
-// fakeLogs returns a CloudWatchLogsClient that serves `events` as a single page,
-// then signals exhaustion by echoing the same nextForwardToken on the next call —
-// exactly how GetLogEvents ends pagination. Records the tokens it was called with.
-function fakeLogs(events: OutputLogEvent[]): CloudWatchLogsClient {
-  let served = false;
-  return {
-    send: async () => {
-      if (!served) {
-        served = true;
-        return { events, nextForwardToken: "t-end" };
-      }
-      return { events: [], nextForwardToken: "t-end" }; // token unchanged → done
-    },
-  } as unknown as CloudWatchLogsClient;
-}
-
-// fakePagedLogs serves each element of `pages` on successive calls, advancing the
-// forward token per page and repeating the last token once to end. Captures every
-// nextToken the caller sent, so a test can assert the loop paged correctly.
-function fakePagedLogs(pages: OutputLogEvent[][]): {
-  client: CloudWatchLogsClient;
-  tokens: (string | undefined)[];
-} {
-  const tokens: (string | undefined)[] = [];
-  let call = 0;
-  const client = {
-    send: async (command: { input: { nextToken?: string } }) => {
-      tokens.push(command.input.nextToken);
-      const i = call++;
-      if (i < pages.length) return { events: pages[i], nextForwardToken: `t-${i}` };
-      return { events: [], nextForwardToken: `t-${pages.length - 1}` }; // repeat last → done
-    },
-  } as unknown as CloudWatchLogsClient;
-  return { client, tokens };
+async function* events(
+  messages: ({ message?: string } | string)[],
+): AsyncGenerator<CloudWatchLogEvent> {
+  for (const item of messages) {
+    yield {
+      timestamp: new Date(0),
+      message: typeof item === "string" ? item : (item.message ?? ""),
+    };
+  }
 }
 
 // A realistic stream shaped after the real `gen_ai.evaluation.result` records
 // (see the recorded fixture below): the level is
 // attributes["aws.bedrock_agentcore.evaluation_level"] (Title-case), session.id
 // sits under attributes, and the trace id is the top-level camelCase `traceId`.
 // One SESSION-level and one TRACE-level record, plus a non-JSON control line.
-const EVENTS: OutputLogEvent[] = [
+const EVENTS = [
   {
     message: JSON.stringify({
       attributes: {
@@ -88,7 +62,7 @@ test("isTerminalStatus recognizes the terminal arm only", () => {
 });
 
 test("readEvaluationResults keeps level + scope so sessions and traces are distinguishable", async () => {
-  const results = await readEvaluationResults(fakeLogs(EVENTS), "lg", "ls", createSilentLogger());
+  const results = await readEvaluationResults(events(EVENTS), createSilentLogger());
 
   // The non-JSON control line is skipped; the two evaluation records parse.
   expect(results).toHaveLength(2);
@@ -109,81 +83,23 @@ test("readEvaluationResults keeps level + scope so sessions and traces are disti
   expect(results.map((r) => r.level)).toEqual(["Session", "Trace"]);
 });
 
-test("readEvaluationResults follows pagination until the forward token stops advancing", async () => {
-  const page = (name: string): OutputLogEvent => ({
-    message: JSON.stringify({
-      attributes: {
-        "gen_ai.evaluation.name": name,
-        "aws.bedrock_agentcore.evaluation_level": "Trace",
-        "session.id": "s1",
-      },
-    }),
-  });
-  const { client, tokens } = fakePagedLogs([
-    [page("Builtin.Correctness")],
-    [page("Builtin.Helpfulness")],
-    [page("Builtin.Faithfulness")],
-  ]);
-
-  const results = await readEvaluationResults(client, "lg", "ls", createSilentLogger());
-
-  // All three pages' records are collected.
-  expect(results.map((r) => r.evaluatorId)).toEqual([
-    "Builtin.Correctness",
-    "Builtin.Helpfulness",
-    "Builtin.Faithfulness",
-  ]);
-  // First call has no token; later calls carry the prior page's forward token; a
-  // final call detects the repeated token and stops.
-  expect(tokens).toEqual([undefined, "t-0", "t-1", "t-2"]);
-});
-
 // Real-log-shape validation lives in the fixture-backed command-flow test
 // (batch-evaluation.fixture.test.tsx), where RECORD=1 captures a live GetLogEvents
 // response and matchGolden pins the parsed output. This file stays a pure unit
 // test over inline synthetic events, matching the rest of src/core.
 
 test("readEvaluationResults skips lines without an evaluation name", async () => {
   const results = await readEvaluationResults(
-    fakeLogs([
+    events([
       { message: JSON.stringify({ attributes: { "some.other.metric": 1 } }) },
       { message: "" },
       { message: undefined },
     ]),
-    "lg",
-    "ls",
     createSilentLogger(),
   );
   expect(results).toEqual([]);
 });
 
-test("readEvaluationResults throws (not silently truncates) when it hits the page cap", async () => {
-  // Token advances on every call, so the loop never detects exhaustion and runs
-  // into MAX_RESULT_PAGES. It must throw so the caller surfaces truncation, rather
-  // than returning the accumulated partial list as if it were complete.
-  let call = 0;
-  const everAdvancing = {
-    send: async () => ({
-      events: [
-        {
-          message: JSON.stringify({
-            attributes: { "gen_ai.evaluation.name": "Builtin.Correctness" },
-          }),
-        },
-      ],
-      nextForwardToken: `t-${call++}`, // always changes → never exhausts
-    }),
-  } as unknown as CloudWatchLogsClient;
-
-  const err = await readEvaluationResults(everAdvancing, "lg", "ls", createSilentLogger()).then(
-    () => undefined,
-    (e) => e as ResultTruncationError,
-  );
-  expect(err).toBeInstanceOf(ResultTruncationError);
-  expect(err?.message).toMatch(/incomplete/);
-  expect(err?.source).toBe("internal"); // our page cap, not a user or service fault
-});
-
 test("parseEvaluationLogEvent warns on and skips an unparseable line", () => {
   const warnings: string[] = [];
   const logger = createSilentLogger();

src/core/batchEvaluationResults.tsx

@@ -1,15 +1,11 @@
-import { GetLogEventsCommand, type CloudWatchLogsClient } from "@aws-sdk/client-cloudwatch-logs";
-import { ResultTruncationError } from "../errors";
+import type { CloudWatchLogEvent } from "./observability/index";
 import type { BatchEvaluationResultEntry } from "../handlers/eval/types";
 import type { Logger } from "../logging";
 
-// Per-session batch-evaluation result retrieval, mirroring
-// core/onlineEvalExecutionRole.tsx's pattern: a self-contained module that takes
-// an injected AWS client (here CloudWatchLogsClient) and owns one slice of Core's
-// behavior. A completed batch evaluation writes each score as an OTel-shaped log
-// record to a per-job CloudWatch stream; this module reads that stream and parses
-// the records. EvalClient calls readEvaluationResults with the client from
-// `this.clients.logs(...)` and the log group + stream from the job's outputConfig.
+// A completed batch evaluation writes each score as an OTel-shaped log record to
+// a per-job CloudWatch stream. CloudWatchClient owns retrieving and paginating
+// that exact stream; this module owns only the eval-specific record parsing and
+// output shape.
 
 // Terminal batch-evaluation statuses — after these, results are final and worth
 // retrieving. Mirrors the AgentCore BatchEvaluationStatus enum's terminal arm.
@@ -19,65 +15,19 @@ export function isTerminalStatus(status?: string): boolean {
   return !!status && TERMINAL_STATUSES.has(status);
 }
 
-// GetLogEvents returns at most 1 MB / 10,000 events per call, so a job with many
-// results spans multiple pages. This caps the page loop as a safety valve against
-// a non-advancing token (see below). At 10k events/page it allows ~1M results, but
-// the 1 MB limit binds first — large explanations can cap a page well under 10k, so
-// this is not a "far beyond any real job" ceiling. Hitting it means the results are
-// truncated, which we surface as an error (see below) rather than silently
-// returning a partial list as if complete.
-const MAX_RESULT_PAGES = 100;
-
-// readEvaluationResults reads and parses the per-session/-trace/-tool scores from
-// a completed batch evaluation's CloudWatch result stream, following pagination to
-// completion. Throws if the stream exceeds MAX_RESULT_PAGES (results would be
-// truncated) — see the throw site. The caller supplies the log group and stream
-// name from the job's GetBatchEvaluation outputConfig (the service-selected values
-// — we do not derive the stream name, since its format is not part of the SDK
-// contract).
+// readEvaluationResults parses the per-session/-trace/-tool scores from a
+// completed batch evaluation's normalized CloudWatch events.
 export async function readEvaluationResults(
-  logs: CloudWatchLogsClient,
-  logGroupName: string,
-  logStreamName: string,
+  events: AsyncIterable<CloudWatchLogEvent>,
   logger: Logger,
 ): Promise<BatchEvaluationResultEntry[]> {
   const results: BatchEvaluationResultEntry[] = [];
-
-  // Page forward from the head. GetLogEvents echoes the input token back as
-  // nextForwardToken once the stream is exhausted, so the loop ends when the
-  // token stops advancing. startFromHead is only honored on the first call (no
-  // token); subsequent calls are positioned by the token.
-  let token: string | undefined;
-  for (let page = 0; page < MAX_RESULT_PAGES; page++) {
-    const response = await logs.send(
-      new GetLogEventsCommand({
-        logGroupName,
-        logStreamName,
-        startFromHead: true,
-        nextToken: token,
-      }),
-    );
-
-    for (const event of response.events ?? []) {
-      if (!event.message) continue;
-      const entry = parseEvaluationLogEvent(event.message, logger);
-      if (entry) results.push(entry);
-    }
-
-    const next = response.nextForwardToken;
-    if (!next || next === token) return results; // exhausted: token stopped advancing
-    token = next;
+  for await (const event of events) {
+    if (!event.message) continue;
+    const entry = parseEvaluationLogEvent(event.message, logger);
+    if (entry) results.push(entry);
   }
-
-  // Cap reached with the token still advancing: the stream has more pages than we
-  // read, so `results` is truncated. Throw rather than return the partial list —
-  // getBatchEvaluation catches this into `resultsError`, which the CLI surfaces as
-  // a stderr warning (stdout metadata stays clean), the same customer-visible path
-  // as any other CloudWatch read failure. A silent partial list would read as
-  // complete.
-  throw new ResultTruncationError(
-    `batch-evaluation results exceed ${MAX_RESULT_PAGES} CloudWatch pages; retrieved ${results.length} results are incomplete`,
-  );
+  return results;
 }
 
 // parseEvaluationLogEvent turns one CloudWatch result-log message into a result

src/core/eval.tsx

@@ -116,6 +116,7 @@ import {
   sanitizeQueryValue,
   type InsightsRowLimit,
 } from "./observability";
+import { CloudWatchClient } from "./observability/index";
 import type {
   BatchEvaluationDetail,
   CodeBasedUpdate,
@@ -175,6 +176,7 @@ import {
   scopePolicyName,
 } from "./onlineEvalExecutionRole";
 import { accountIdFromArn, deleteAbTestRole, provisionAbTestRole } from "./abTestExecutionRole";
+import { harnessRuntimeFromResponse } from "./harness";
 
 const DEFAULT_INGESTION_WAIT_MS = 180_000;
 const DATASET_EXAMPLES_BATCH_LIMIT = 1000;
@@ -212,6 +214,7 @@ const EVAL_INSIGHTS_ROW_LIMIT: InsightsRowLimit = {
 };
 
 const DEFAULT_BATCH_INSIGHTS_PAGE_SIZE = 50;
+const BATCH_EVALUATION_RESULT_MAX_PAGES = 100;
 
 // noopLogger is the default for the optional logger arg so callers that don't
 // need batch-evaluation result-log diagnostics (e.g. dataset-only tests) can
@@ -235,6 +238,7 @@ export class EvalClient implements CoreEvalClient {
     private readonly logger: Logger = noopLogger,
     private readonly newSessionId: () => string = randomUUID,
     private readonly now: () => number = () => Date.now(),
+    private readonly cloudWatch: CloudWatchClient = new CloudWatchClient(clients),
   ) {}
 
   async createEvaluator(
@@ -434,9 +438,16 @@ export class EvalClient implements CoreEvalClient {
 
     try {
       detail.results = await readEvaluationResults(
-        this.clients.logs({ region: options.region }),
-        cw.logGroupName,
-        cw.logStreamName,
+        this.cloudWatch.readLogStream(
+          {
+            logGroupName: cw.logGroupName,
+            logStreamName: cw.logStreamName,
+          },
+          {
+            maxPages: BATCH_EVALUATION_RESULT_MAX_PAGES,
+          },
+          options,
+        ),
         this.logger,
       );
       return { detail };
@@ -1946,17 +1957,15 @@ async function resolveAgentToNameAndId(
   }
 
   const harness = await control.send(new GetHarnessCommand({ harnessId: agent }));
-  const environment = harness.harness?.environment;
-  const runtimeEnv =
-    environment && "agentCoreRuntimeEnvironment" in environment
-      ? environment.agentCoreRuntimeEnvironment
-      : undefined;
-  if (!runtimeEnv?.agentRuntimeId || !runtimeEnv?.agentRuntimeName) {
+  try {
+    return harnessRuntimeFromResponse(agent, harness);
+  } catch (error) {
+    if (!(error instanceof InputValidationError)) throw error;
     throw new InputValidationError(`"${agent}" does not exist as a runtime or a harness`, {
+      cause: error,
       meta: { agent },
     });
   }
-  return { runtimeId: runtimeEnv.agentRuntimeId, runtimeName: runtimeEnv.agentRuntimeName };
 }
 
 // agentDataSource builds the CloudWatch data source for an agent id, resolving it

src/core/harness.test.ts

@@ -0,0 +1,59 @@
+import { describe, expect, test } from "bun:test";
+import {
+  GetHarnessCommand,
+  type GetHarnessResponse,
+} from "@aws-sdk/client-bedrock-agentcore-control";
+import { InputValidationError } from "../errors";
+import type { AwsClients, ClientConfig } from "./types";
+import { HarnessClient, harnessRuntimeFromResponse } from "./harness";
+
+const RESPONSE = {
+  harness: {
+    environment: {
+      agentCoreRuntimeEnvironment: {
+        agentRuntimeId: "harness_MyHarness-AbC123",
+        agentRuntimeName: "harness_MyHarness",
+      },
+    },
+  },
+} as GetHarnessResponse;
+
+describe("harnessRuntimeFromResponse", () => {
+  test("rejects a harness without a resolvable Runtime environment", () => {
+    expect(() =>
+      harnessRuntimeFromResponse("MyHarness-abc123", {
+        harness: { environment: { $unknown: ["futureProvider", {}] } },
+      } as GetHarnessResponse),
+    ).toThrow(InputValidationError);
+  });
+});
+
+describe("HarnessClient.resolveRuntime", () => {
+  test("gets the harness with the configured client and forwards cancellation", async () => {
+    const configs: ClientConfig[] = [];
+    const controller = new AbortController();
+    const clients = {
+      control: (config: ClientConfig) => {
+        configs.push(config);
+        return {
+          send: async (command: unknown, options?: { abortSignal?: AbortSignal }) => {
+            expect(command).toBeInstanceOf(GetHarnessCommand);
+            expect((command as GetHarnessCommand).input).toEqual({
+              harnessId: "MyHarness-abc123",
+            });
+            expect(options?.abortSignal).toBe(controller.signal);
+            return RESPONSE;
+          },
+        };
+      },
+    } as unknown as AwsClients;
+    const client = new HarnessClient(clients);
+
+    await client.resolveRuntime(
+      "MyHarness-abc123",
+      { region: "us-west-2", endpointUrl: "https://control.test" },
+      controller.signal,
+    );
+    expect(configs).toEqual([{ region: "us-west-2", endpoint: "https://control.test" }]);
+  });
+});