feat: add observability abstractions and wire to runtime

src/core/batchEvaluationResults.test.ts

@@ -1,47 +1,38 @@
 import { test, expect } from "bun:test";
-import type { CloudWatchLogsClient, OutputLogEvent } from "@aws-sdk/client-cloudwatch-logs";
-import { ResultTruncationError } from "../errors";
+import type { OutputLogEvent } from "@aws-sdk/client-cloudwatch-logs";
 import { createSilentLogger } from "../testing";
+import type { LogStreamReadQuery, LogStreamSource, SourceReader } from "./observability";
 import {
   isTerminalStatus,
   parseEvaluationLogEvent,
   readEvaluationResults,
 } from "./batchEvaluationResults";
 
-// 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;
+const SOURCE: LogStreamSource = {
+  provider: "cloudwatch",
+  logGroupName: "lg",
+  logStreamName: "ls",
+};
+const OPTIONS = { region: "us-west-2" };
+
+function fakeReader(
+  events: OutputLogEvent[],
+  onRead?: (source: LogStreamSource, query: LogStreamReadQuery) => void,
+): Pick<SourceReader, "readLogStream"> {
   return {
-    send: async () => {
-      if (!served) {
-        served = true;
-        return { events, nextForwardToken: "t-end" };
+    async *readLogStream(source, query) {
+      onRead?.(source, query);
+      for (const event of events) {
+        yield {
+          timestamp: event.timestamp ?? 0,
+          message: event.message ?? "",
+          ...(event.ingestionTime !== undefined ? { ingestionTime: event.ingestionTime } : {}),
+          logStreamName: source.logStreamName,
+          raw: event,
+        };
       }
-      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 };
+  };
 }
 
 // A realistic stream shaped after the real `gen_ai.evaluation.result` records
@@ -88,7 +79,13 @@ 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 reads: { source: LogStreamSource; query: LogStreamReadQuery }[] = [];
+  const results = await readEvaluationResults(
+    fakeReader(EVENTS, (source, query) => reads.push({ source, query })),
+    SOURCE,
+    OPTIONS,
+    createSilentLogger(),
+  );
 
   // The non-JSON control line is skipped; the two evaluation records parse.
   expect(results).toHaveLength(2);
@@ -107,35 +104,7 @@ test("readEvaluationResults keeps level + scope so sessions and traces are disti
     traceId: "4bf92f3577b34da6a3ce929d0e0e4736",
   });
   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"]);
+  expect(reads).toEqual([{ source: SOURCE, query: { maxPages: 100 } }]);
 });
 
 // Real-log-shape validation lives in the fixture-backed command-flow test
@@ -145,45 +114,18 @@ test("readEvaluationResults follows pagination until the forward token stops adv
 
 test("readEvaluationResults skips lines without an evaluation name", async () => {
   const results = await readEvaluationResults(
-    fakeLogs([
+    fakeReader([
       { message: JSON.stringify({ attributes: { "some.other.metric": 1 } }) },
       { message: "" },
       { message: undefined },
     ]),
-    "lg",
-    "ls",
+    SOURCE,
+    OPTIONS,
     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 { BatchEvaluationResultEntry } from "../handlers/eval/types";
 import type { Logger } from "../logging";
+import type { CoreOptions } from "./types";
+import type { LogStreamSource, SourceReader } from "./observability";
 
-// 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. The shared source reader owns provider access and
+// pagination; this module owns the Eval-specific result shape.
 
 // Terminal batch-evaluation statuses — after these, results are final and worth
 // retrieving. Mirrors the AgentCore BatchEvaluationStatus enum's terminal arm.
@@ -19,65 +15,33 @@ 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.
+// At 10k events/page this allows ~1M results, but CloudWatch's 1 MB response limit
+// can bind first. Hitting the cap throws from the shared source reader rather than
+// returning a partial result set as if it were 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).
+// a completed batch evaluation's CloudWatch result stream. The caller supplies
+// the exact group and stream returned by GetBatchEvaluation; no resource resolver
+// is involved.
 export async function readEvaluationResults(
-  logs: CloudWatchLogsClient,
-  logGroupName: string,
-  logStreamName: string,
+  sourceReader: Pick<SourceReader, "readLogStream">,
+  source: LogStreamSource,
+  options: CoreOptions,
   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 record of sourceReader.readLogStream(
+    source,
+    { maxPages: MAX_RESULT_PAGES },
+    options,
+  )) {
+    if (!record.message) continue;
+    const entry = parseEvaluationLogEvent(record.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/core.test.ts

@@ -209,6 +209,7 @@ test("exposes feature sub-clients", () => {
   expect(core.harness).toBeDefined();
   expect(core.memory).toBeDefined();
   expect(core.gateway).toBeDefined();
+  expect(core.observability).toBeDefined();
 });
 
 test("getEvent sends a GetEventCommand on the data client", async () => {

src/core/eval.tsx

@@ -109,12 +109,14 @@ import {
   ResourceNotFoundError,
 } from "../errors";
 import {
+  CloudWatchSourceReader,
   DEFAULT_ENDPOINT_QUALIFIER,
   INSIGHTS_MAX_ROWS,
   runInsightsQuery,
   runtimeLogGroup,
   sanitizeQueryValue,
   type InsightsRowLimit,
+  type SourceReader,
 } from "./observability";
 import type {
   BatchEvaluationDetail,
@@ -233,6 +235,7 @@ export class EvalClient implements CoreEvalClient {
     private readonly logger: Logger = noopLogger,
     private readonly newSessionId: () => string = randomUUID,
     private readonly now: () => number = () => Date.now(),
+    private readonly sourceReader: SourceReader = new CloudWatchSourceReader(clients),
   ) {}
 
   async createEvaluator(
@@ -432,9 +435,13 @@ export class EvalClient implements CoreEvalClient {
 
     try {
       detail.results = await readEvaluationResults(
-        this.clients.logs({ region: options.region }),
-        cw.logGroupName,
-        cw.logStreamName,
+        this.sourceReader,
+        {
+          provider: "cloudwatch",
+          logGroupName: cw.logGroupName,
+          logStreamName: cw.logStreamName,
+        },
+        options,
         this.logger,
       );
       return { detail };

src/core/index.tsx

@@ -7,9 +7,12 @@ import { GatewayClient } from "./gateway";
 import { HarnessClient } from "./harness";
 import { IdentityClient } from "./identity";
 import { MemoryClient } from "./memory";
-import { ObservabilityClient } from "./observability";
+import {
+  CloudWatchSourceReader,
+  ObservabilityClient,
+  RuntimeSourceResolver,
+} from "./observability";
 import { RuntimeClient } from "./runtime";
-import { FsReadWriteJson } from "../io";
 import type {
   AwsClients,
   ClientConfig,
@@ -84,6 +87,7 @@ export class CoreClient implements AwsClients {
     this.createLogsClient = config.createLogsClient;
     this.logger = config.logger;
     const fetch = config.fetch ?? globalThis.fetch;
+    const sourceReader = new CloudWatchSourceReader(this);
     this.runtime = new RuntimeClient(this, fetch, this.logger.child({ module: "runtime" }));
     this.gateway = new GatewayClient(this, fetch, this.logger.child({ module: "gateway" }));
     // EvalClient shares the injected fetch: dataset content is served from a
@@ -95,16 +99,12 @@ export class CoreClient implements AwsClients {
       this.logger.child({ module: "eval" }),
       config.newSessionId,
       config.now,
+      sourceReader,
+    );
+    this.observability = new ObservabilityClient(
+      { runtime: new RuntimeSourceResolver() },
+      sourceReader,
     );
-
-    // Observability resolves a project's deployed runtime from its stack
-    // outputs, so it reads aws-targets.json through the same JSON layer the
-    // project manager uses.
-    this.observability = new ObservabilityClient(this, {
-      readJson: new FsReadWriteJson({
-        logger: this.logger.child({ module: "observability" }),
-      }),
-    });
 
     this.projectManager = new FsProjectManager({
       logger: this.logger.child({ module: "projectManager" }),
@@ -150,7 +150,7 @@ export class CoreClient implements AwsClients {
   }
 
   // logs returns the CloudWatch Logs client for `config`, creating and caching it
-  // on first use (used to read batch-evaluation result log streams).
+  // on first use for customer-facing observability and evaluation result streams.
   logs(config: ClientConfig): CloudWatchLogsClient {
     const key = cacheKey(config);
     let client = this.logsClients.get(key);