feat: add observability abstractions and wire to runtime

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/index.tsx

@@ -7,6 +7,11 @@ import { GatewayClient } from "./gateway";
 import { HarnessClient } from "./harness";
 import { IdentityClient } from "./identity";
 import { MemoryClient } from "./memory";
+import {
+  CloudWatchSourceReader,
+  ObservabilityClient,
+  RuntimeSourceResolver,
+} from "./observability";
 import { RuntimeClient } from "./runtime";
 import type {
   AwsClients,
@@ -67,6 +72,7 @@ export class CoreClient implements AwsClients {
   readonly runtime: RuntimeClient;
   readonly gateway: GatewayClient;
   readonly eval: EvalClient;
+  readonly observability: ObservabilityClient;
 
   readonly projectManager: ProjectManager;
 
@@ -88,6 +94,10 @@ export class CoreClient implements AwsClients {
       this.logger.child({ module: "eval" }),
       config.newSessionId,
     );
+    this.observability = new ObservabilityClient(
+      { runtime: new RuntimeSourceResolver() },
+      new CloudWatchSourceReader(this),
+    );
 
     this.projectManager = new FsProjectManager({
       logger: this.logger.child({ module: "projectManager" }),
@@ -132,7 +142,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);

src/core/observability/client.test.ts

@@ -0,0 +1,144 @@
+import { describe, expect, test } from "bun:test";
+import type { CoreOptions } from "../types";
+import { ObservabilityClient, type LogRecord } from "./client";
+import type { LogSource, ObservabilitySourceResolverRegistry } from "./resolver";
+import type { LogSearchQuery, LogTailQuery, RawLogRecord, SourceReader } from "./sourceReader";
+
+const SOURCE: LogSource = {
+  provider: "cloudwatch",
+  logGroupName: "/aws/runtime-1",
+};
+const OPTIONS = { region: "us-east-1" };
+
+async function collect(records: AsyncIterable<LogRecord>) {
+  const result: LogRecord[] = [];
+  for await (const record of records) result.push(record);
+  return result;
+}
+
+function createClient(rawRecords: RawLogRecord[]) {
+  const calls: { method: string; args: unknown[] }[] = [];
+  const resolvers: ObservabilitySourceResolverRegistry = {
+    runtime: {
+      resolve: async (...args) => {
+        calls.push({ method: "resolve", args });
+        return {
+          resource: {
+            kind: "runtime",
+            id: args[0].id,
+            qualifier: args[0].qualifier ?? "DEFAULT",
+          },
+          logs: [SOURCE],
+        };
+      },
+    },
+  };
+  const reader: SourceReader = {
+    async *searchLogs(
+      source: LogSource,
+      query: LogSearchQuery,
+      options: CoreOptions,
+      signal?: AbortSignal,
+    ) {
+      calls.push({
+        method: "searchLogs",
+        args: [source, query, options, signal],
+      });
+      yield* rawRecords;
+    },
+    async *tailLogs(
+      source: LogSource,
+      query: LogTailQuery,
+      options: CoreOptions,
+      signal: AbortSignal,
+    ) {
+      calls.push({
+        method: "tailLogs",
+        args: [source, query, options, signal],
+      });
+      yield* rawRecords;
+    },
+  };
+  return { client: new ObservabilityClient(resolvers, reader), calls };
+}
+
+describe("ObservabilityClient", () => {
+  test("resolves, reads, and normalizes common log metadata", async () => {
+    const raw = {
+      timestamp: 1_709_391_000_000,
+      ingestionTime: 1_709_391_000_100,
+      logStreamName: "runtime-stream",
+      message: JSON.stringify({
+        traceId: "trace-1",
+        spanId: "span-1",
+        parentSpanId: "parent-1",
+        severityText: "INFO",
+        attributes: { "session.id": "session-1" },
+      }),
+      raw: { eventId: "event-1" },
+    };
+    const { client, calls } = createClient([raw]);
+    const signal = new AbortController().signal;
+    const query = { startTimeMs: 1, endTimeMs: 2 };
+
+    const records = await collect(
+      client.searchLogs(
+        { kind: "runtime", id: "runtime-1", qualifier: "blue" },
+        query,
+        OPTIONS,
+        signal,
+      ),
+    );
+
+    expect(calls.map((call) => call.method)).toEqual(["resolve", "searchLogs"]);
+    expect(records).toEqual([
+      {
+        timestamp: new Date(1_709_391_000_000),
+        ingestionTime: new Date(1_709_391_000_100),
+        message: raw.message,
+        correlation: {
+          traceId: "trace-1",
+          spanId: "span-1",
+          parentSpanId: "parent-1",
+          sessionId: "session-1",
+        },
+        severity: "INFO",
+        source: {
+          provider: "cloudwatch",
+          resource: {
+            kind: "runtime",
+            id: "runtime-1",
+            qualifier: "blue",
+          },
+          logGroupName: SOURCE.logGroupName,
+          logStreamName: "runtime-stream",
+        },
+        raw: { eventId: "event-1" },
+      },
+    ]);
+  });
+
+  test("uses the same orchestration path for Live Tail records", async () => {
+    const { client, calls } = createClient([
+      {
+        timestamp: 1,
+        message: "plain text",
+        raw: { message: "plain text" },
+      },
+    ]);
+    const signal = new AbortController().signal;
+
+    const records = await collect(
+      client.tailLogs(
+        { kind: "runtime", id: "runtime-1" },
+        { filterPattern: "ERROR" },
+        OPTIONS,
+        signal,
+      ),
+    );
+
+    expect(calls.map((call) => call.method)).toEqual(["resolve", "tailLogs"]);
+    expect(records[0]).not.toHaveProperty("correlation");
+    expect(records[0]).not.toHaveProperty("severity");
+  });
+});

src/core/observability/client.ts

@@ -0,0 +1,148 @@
+import type { CoreOptions } from "../types";
+import type {
+  LogSource,
+  ObservableResourceRef,
+  ObservabilitySourceResolver,
+  ObservabilitySourceResolverRegistry,
+  ResolvedObservabilityTarget,
+  ResolvedResourceIdentity,
+} from "./resolver";
+import type { LogSearchQuery, LogTailQuery, RawLogRecord, SourceReader } from "./sourceReader";
+
+export interface LogRecord {
+  timestamp: Date;
+  message: string;
+  correlation?: {
+    traceId?: string;
+    spanId?: string;
+    parentSpanId?: string;
+    sessionId?: string;
+  };
+  severity?: string;
+  ingestionTime?: Date;
+  source: {
+    provider: "cloudwatch";
+    resource: ResolvedResourceIdentity;
+    logGroupName: string;
+    logStreamName?: string;
+  };
+  raw?: unknown;
+}
+
+export interface CoreObservabilityClient {
+  searchLogs(
+    resource: ObservableResourceRef,
+    query: LogSearchQuery,
+    options: CoreOptions,
+    signal?: AbortSignal,
+  ): AsyncIterable<LogRecord>;
+
+  tailLogs(
+    resource: ObservableResourceRef,
+    query: LogTailQuery,
+    options: CoreOptions,
+    signal: AbortSignal,
+  ): AsyncIterable<LogRecord>;
+}
+
+/**
+ * Shared entry point for logs. It orchestrates resolution and provider reads,
+ * then normalizes provider events into the stable record contract.
+ */
+export class ObservabilityClient implements CoreObservabilityClient {
+  constructor(
+    private readonly resolvers: ObservabilitySourceResolverRegistry,
+    private readonly sourceReader: SourceReader,
+  ) {}
+
+  async *searchLogs(
+    resource: ObservableResourceRef,
+    query: LogSearchQuery,
+    options: CoreOptions,
+    signal?: AbortSignal,
+  ): AsyncGenerator<LogRecord, void> {
+    const target = await this.resolve(resource, options, signal);
+    for (const source of target.logs) {
+      for await (const raw of this.sourceReader.searchLogs(source, query, options, signal)) {
+        yield toLogRecord(target.resource, source, raw);
+      }
+    }
+  }
+
+  async *tailLogs(
+    resource: ObservableResourceRef,
+    query: LogTailQuery,
+    options: CoreOptions,
+    signal: AbortSignal,
+  ): AsyncGenerator<LogRecord, void> {
+    const target = await this.resolve(resource, options, signal);
+    for (const source of target.logs) {
+      for await (const raw of this.sourceReader.tailLogs(source, query, options, signal)) {
+        yield toLogRecord(target.resource, source, raw);
+      }
+    }
+  }
+
+  private resolve(
+    resource: ObservableResourceRef,
+    options: CoreOptions,
+    signal?: AbortSignal,
+  ): Promise<ResolvedObservabilityTarget> {
+    const resolver = this.resolvers[resource.kind] as ObservabilitySourceResolver<typeof resource>;
+    return resolver.resolve(resource, options, signal);
+  }
+}
+
+function toLogRecord(
+  resource: ResolvedResourceIdentity,
+  source: LogSource,
+  record: RawLogRecord,
+): LogRecord {
+  const metadata = extractCommonMetadata(record.message);
+  return {
+    timestamp: new Date(record.timestamp),
+    message: record.message,
+    ...metadata,
+    ...(record.ingestionTime !== undefined
+      ? { ingestionTime: new Date(record.ingestionTime) }
+      : {}),
+    source: {
+      provider: source.provider,
+      resource,
+      logGroupName: source.logGroupName,
+      ...(record.logStreamName ? { logStreamName: record.logStreamName } : {}),
+    },
+    raw: record.raw,
+  };
+}
+
+function extractCommonMetadata(message: string): Pick<LogRecord, "correlation" | "severity"> {
+  let parsed: Record<string, unknown>;
+  try {
+    parsed = JSON.parse(message) as Record<string, unknown>;
+  } catch {
+    return {};
+  }
+  if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) return {};
+
+  const attributes =
+    parsed.attributes && typeof parsed.attributes === "object" && !Array.isArray(parsed.attributes)
+      ? (parsed.attributes as Record<string, unknown>)
+      : {};
+  const stringValue = (value: unknown): string | undefined =>
+    typeof value === "string" ? value : undefined;
+  const correlation = {
+    traceId: stringValue(parsed.traceId),
+    spanId: stringValue(parsed.spanId),
+    parentSpanId: stringValue(parsed.parentSpanId),
+    sessionId: stringValue(parsed.sessionId) ?? stringValue(attributes["session.id"]),
+  };
+  const hasCorrelation = Object.values(correlation).some((value) => value !== undefined);
+  const severity =
+    stringValue(parsed.severityText) ?? stringValue(parsed.severity) ?? stringValue(parsed.level);
+
+  return {
+    ...(hasCorrelation ? { correlation } : {}),
+    ...(severity ? { severity } : {}),
+  };
+}

src/core/observability/index.ts

@@ -0,0 +1,19 @@
+export { ObservabilityClient, type CoreObservabilityClient, type LogRecord } from "./client";
+export {
+  DEFAULT_RUNTIME_QUALIFIER,
+  RuntimeSourceResolver,
+  runtimeLogGroup,
+  type LogSource,
+  type ObservableResourceRef,
+  type ObservabilitySourceResolver,
+  type ObservabilitySourceResolverRegistry,
+  type ResolvedObservabilityTarget,
+  type ResolvedResourceIdentity,
+} from "./resolver";
+export {
+  CloudWatchSourceReader,
+  type LogSearchQuery,
+  type LogTailQuery,
+  type RawLogRecord,
+  type SourceReader,
+} from "./sourceReader";