fix(dev): harden OTEL collector, container reachability, and dev lifecycle

Address reviewer findings on the collector and dev wiring:

- Validate top-level OTLP shape and return 400 instead of mislabeling a bad
  payload as a persistence error.
- Guard the shared HTTP server against a client that disconnects mid-response
  so it can no longer crash project dev; add an optional bind host.
- Bind the collector to 0.0.0.0 for container runtimes so a container can
  reach it over the host bridge on Linux.
- Run the container template under opentelemetry-instrument so it emits traces.
- Keep the collector alive through the child's shutdown grace so final spans
  are not lost.
- Force the OTEL settings that would otherwise let shell or .env.local values
  disable or break local collection.
- Make the uv sitecustomize discovery abortable and read its path from a
  marker rather than the last merged output line.

Author: tejaskash

src/assets/templates/hello-world-python-container/Dockerfile

@@ -34,4 +34,6 @@ USER bedrock_agentcore
 # 9000: A2A Mode
 EXPOSE 8080 8000 9000
 
-CMD ["python", "-m", "main"]
+# opentelemetry-instrument (from aws-opentelemetry-distro) starts a real
+# TracerProvider; plain `python -m main` would export nothing.
+CMD ["opentelemetry-instrument", "python", "-m", "main"]

src/core/dev/codezip.test.ts

@@ -54,21 +54,25 @@ async function projectRoot(withNodeModules = false): Promise<string> {
   return root;
 }
 
-function harness(output: ProcessEvent[] = [], probe: { dir?: string; fail?: boolean } = {}) {
+function harness(
+  output: ProcessEvent[] = [],
+  site: { dir?: string; fail?: boolean; noise?: string[] } = {},
+) {
   const calls: ProcessCall[] = [];
-  const probeCalls: string[][] = [];
+  const discoverCalls: string[][] = [];
   const fakeStreamProcess: ProcessStreamer = async function* (command, options) {
     calls.push({ command, options });
     yield* output;
   };
   const fakeRunProcess: ProcessRunner = async (command, options) => {
-    probeCalls.push(command);
-    if (probe.fail) throw new Error("probe failed");
-    options.onOutput?.(`${probe.dir ?? ""}\n`);
+    discoverCalls.push(command);
+    if (site.fail) throw new Error("discovery failed");
+    for (const line of site.noise ?? []) options.onOutput?.(`${line}\n`);
+    if (site.dir !== undefined) options.onOutput?.(`AGENTCORE_OTEL_SITECUSTOMIZE=${site.dir}\n`);
   };
   return {
     calls,
-    probeCalls,
+    discoverCalls,
     runner: new CodeZipDevRunner({ streamProcess: fakeStreamProcess, runProcess: fakeRunProcess }),
   };
 }
@@ -220,11 +224,24 @@ describe("CodeZipDevRunner OTEL instrumentation", () => {
   test("prepends the sitecustomize directory to PYTHONPATH when instrumentation is installed", async () => {
     const root = await projectRoot();
     const directory = await sitecustomizeDir();
-    const { calls, probeCalls, runner } = harness([], { dir: directory });
+    const { calls, discoverCalls, runner } = harness([], { dir: directory });
+
+    await collect(runner.run(otelInput(root)));
+
+    expect(discoverCalls[0]?.slice(0, 4)).toEqual(["uv", "run", "python", "-c"]);
+    expect(calls[0]?.options.env?.PYTHONPATH).toBe(directory);
+  });
+
+  test("reads the marked path even when uv writes progress to the merged output", async () => {
+    const root = await projectRoot();
+    const directory = await sitecustomizeDir();
+    const { calls, runner } = harness([], {
+      dir: directory,
+      noise: ["Resolved 12 packages", "Installed 12 packages"],
+    });
 
     await collect(runner.run(otelInput(root)));
 
-    expect(probeCalls[0]?.slice(0, 4)).toEqual(["uv", "run", "python", "-c"]);
     expect(calls[0]?.options.env?.PYTHONPATH).toBe(directory);
   });
 
@@ -238,26 +255,26 @@ describe("CodeZipDevRunner OTEL instrumentation", () => {
     expect(calls[0]?.options.env?.PYTHONPATH).toBe(`${directory}${delimiter}/existing`);
   });
 
-  test("does not probe without an OTEL endpoint or for Node entrypoints", async () => {
+  test("does not run discovery without an OTEL endpoint or for Node entrypoints", async () => {
     const root = await projectRoot(true);
-    const { probeCalls, runner } = harness();
+    const { discoverCalls, runner } = harness();
 
     await collect(runner.run(input(root, runtime())));
     await collect(runner.run({ ...otelInput(root), runtime: runtime({ entrypoint: "index.js" }) }));
 
-    expect(probeCalls).toEqual([]);
+    expect(discoverCalls).toEqual([]);
   });
 
   test.each([
-    ["probe failure", { fail: true }],
+    ["discovery failure", { fail: true }],
     ["missing sitecustomize.py", { dir: "/nonexistent" }],
-  ] as const)("warns and starts untraced on %s", async (_case, probe) => {
+  ] as const)("warns and starts untraced on %s", async (_case, site) => {
     const root = await projectRoot();
-    const { calls, probeCalls, runner } = harness([], probe);
+    const { calls, discoverCalls, runner } = harness([], site);
 
     const events = await collect(runner.run(otelInput(root)));
 
-    expect(probeCalls).toHaveLength(1);
+    expect(discoverCalls).toHaveLength(1);
     expect(calls).toHaveLength(1);
     expect(calls[0]?.options.env?.PYTHONPATH).toBeUndefined();
     expect(events).toContainEqual({

src/core/dev/codezip.ts

@@ -16,6 +16,8 @@ type CodeZipDevRunnerConfig = {
   runProcess?: ProcessRunner;
 };
 
+const SITECUSTOMIZE_MARKER = "AGENTCORE_OTEL_SITECUSTOMIZE=";
+
 export class CodeZipDevRunner implements DevRunner {
   private readonly streamProcess: ProcessStreamer;
   private readonly runProcess: ProcessRunner;
@@ -51,7 +53,7 @@ export class CodeZipDevRunner implements DevRunner {
     yield { type: "status", message: "Starting development server" };
     const serverProcess = commandForRuntime(entrypoint!, directory, input);
     if (entrypoint!.endsWith(".py") && input.env?.OTEL_EXPORTER_OTLP_ENDPOINT) {
-      const sitecustomizeDir = await this.findOtelSitecustomizeDir(directory);
+      const sitecustomizeDir = await this.findOtelSitecustomizeDir(directory, input.signal);
       if (sitecustomizeDir) {
         const existing = serverProcess.options.env?.PYTHONPATH;
         serverProcess.options.env = {
@@ -75,19 +77,29 @@ export class CodeZipDevRunner implements DevRunner {
    * an `opentelemetry-instrument` wrapper would only instrument uvicorn's reloader
    * parent, leaving the re-spawned worker processes untraced.
    */
-  private async findOtelSitecustomizeDir(directory: string): Promise<string | undefined> {
+  private async findOtelSitecustomizeDir(
+    directory: string,
+    signal: AbortSignal,
+  ): Promise<string | undefined> {
     const output: string[] = [];
-    const probe =
-      "import opentelemetry.instrumentation.auto_instrumentation as m; import os; print(os.path.dirname(m.__file__))";
+    // uv writes sync progress to stderr, which merges into onOutput, so the path
+    // is printed behind a marker and read from that line rather than the last one.
+    const script = `import opentelemetry.instrumentation.auto_instrumentation as m, os; print("${SITECUSTOMIZE_MARKER}" + os.path.dirname(m.__file__))`;
     try {
-      await this.runProcess(["uv", "run", "python", "-c", probe], {
+      await this.runProcess(["uv", "run", "python", "-c", script], {
         cwd: directory,
         onOutput: (chunk) => output.push(chunk),
+        signal,
       });
     } catch {
       return undefined;
     }
-    const sitecustomizeDir = output.join("").trim().split("\n").at(-1)?.trim();
+    const marked = output
+      .join("")
+      .split("\n")
+      .map((line) => line.trim())
+      .find((line) => line.startsWith(SITECUSTOMIZE_MARKER));
+    const sitecustomizeDir = marked?.slice(SITECUSTOMIZE_MARKER.length);
     if (!sitecustomizeDir || !existsSync(join(sitecustomizeDir, "sitecustomize.py")))
       return undefined;
     return sitecustomizeDir;

src/core/dev/otel/collector.test.ts

@@ -138,6 +138,29 @@ describe("startOtelCollector", () => {
     expect(await collector.store.list()).toEqual([]);
   });
 
+  test.each(["null", "[]", "42", '{"resourceSpans":5}'])(
+    "rejects structurally invalid JSON %s with 400 instead of a persistence error",
+    async (body) => {
+      const errors: unknown[] = [];
+      const strict = await startOtelCollector({
+        tracesDirectory: directory,
+        onError: (error) => errors.push(error),
+      });
+      try {
+        const response = await fetch(`http://127.0.0.1:${strict.port}/v1/traces`, {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body,
+        });
+        expect(response.status).toBe(400);
+        expect(errors).toEqual([]);
+        expect(await strict.store.list()).toEqual([]);
+      } finally {
+        await strict.close();
+      }
+    },
+  );
+
   test("acks with 200 and reports onError when persistence fails", async () => {
     // A traces dir nested under a regular file makes mkdir (and thus append) fail.
     const blocker = join(directory, "blocker");
@@ -184,6 +207,17 @@ describe("startOtelCollector", () => {
     });
   });
 
+  test("envVars force the settings that would otherwise break local collection", () => {
+    expect(collector.envVars).toMatchObject({
+      OTEL_SDK_DISABLED: "false",
+      OTEL_TRACES_EXPORTER: "otlp",
+      OTEL_LOGS_EXPORTER: "otlp",
+      OTEL_EXPORTER_OTLP_COMPRESSION: "none",
+      OTEL_EXPORTER_OTLP_TRACES_COMPRESSION: "none",
+      OTEL_EXPORTER_OTLP_LOGS_COMPRESSION: "none",
+    });
+  });
+
   test("abort signal closes the receiver", async () => {
     const controller = new AbortController();
     const aborted = await startOtelCollector({

src/core/dev/otel/collector.ts

@@ -33,7 +33,7 @@ export const ExportLogsServiceRequest = logs.v1.ExportLogsServiceRequest;
 type OtlpDecoder = Pick<OtlpMessageType, "decode">;
 
 export interface OtelCollector {
-  /** The loopback port the OTLP/HTTP receiver listens on. */
+  /** The port the OTLP/HTTP receiver listens on. */
   port: number;
   /** Reads the traces this collector persists. */
   store: TraceStore;
@@ -46,22 +46,25 @@ export interface OtelCollector {
 export interface StartOtelCollectorOptions {
   /** Directory to persist OTLP JSON Lines files into. */
   tracesDirectory: string;
+  /** Address to bind. Defaults to 127.0.0.1; use 0.0.0.0 to reach it from a container. */
+  host?: string;
   /** Closes the collector when aborted. */
   signal?: AbortSignal;
   /** Called when a batch can't be persisted; the export is still acked to stop retries. */
   onError?: (error: unknown) => void;
 }
 
 /**
- * Starts an in-process OTLP/HTTP receiver for dev mode on an OS-assigned
- * loopback port. Accepts `POST /v1/traces` and `POST /v1/logs` in protobuf or
- * JSON encoding and appends the raw payloads to a TraceStore.
+ * Starts an in-process OTLP/HTTP receiver for dev mode on an OS-assigned port.
+ * Accepts `POST /v1/traces` and `POST /v1/logs` in protobuf or JSON encoding and
+ * appends the raw payloads to a TraceStore.
  */
 export async function startOtelCollector(
   options: StartOtelCollectorOptions,
 ): Promise<OtelCollector> {
   const store = new TraceStore(options.tracesDirectory);
   const server = await startHttpServer((request) => route(request, store, options.onError), {
+    host: options.host,
     signal: options.signal,
   });
 
@@ -74,10 +77,10 @@ async function route(
   onError?: (error: unknown) => void,
 ): Promise<HttpResponse> {
   if (request.method === "POST" && request.url === "/v1/traces") {
-    return ingest(request, store, ExportTraceServiceRequest, onError);
+    return ingest(request, store, ExportTraceServiceRequest, "resourceSpans", onError);
   }
   if (request.method === "POST" && request.url === "/v1/logs") {
-    return ingest(request, store, ExportLogsServiceRequest, onError);
+    return ingest(request, store, ExportLogsServiceRequest, "resourceLogs", onError);
   }
   if (request.method === "GET" && request.url === "/") {
     return json(200, { status: "ok" });
@@ -89,16 +92,20 @@ async function ingest(
   request: HttpRequest,
   store: TraceStore,
   decoder: OtlpDecoder,
+  field: "resourceSpans" | "resourceLogs",
   onError?: (error: unknown) => void,
 ): Promise<HttpResponse> {
-  let payload: OtlpPayload;
+  let decoded: unknown;
   try {
-    payload = decodePayload(request.body, String(request.headers["content-type"] ?? ""), decoder);
+    decoded = decodePayload(request.body, String(request.headers["content-type"] ?? ""), decoder);
   } catch {
     return json(400, { error: "Invalid OTLP payload" });
   }
+  if (!isOtlpPayload(decoded, field)) {
+    return json(400, { error: "Invalid OTLP payload" });
+  }
   try {
-    await store.append(payload);
+    await store.append(decoded);
   } catch (error) {
     // A persistence failure (disk full, permissions) is the collector's problem,
     // not the agent's: ack the export anyway so the SDK exporter stops retrying the
@@ -108,24 +115,44 @@ async function ingest(
   return json(200, {});
 }
 
-/**
- * Decode an OTLP payload. The JSON round-trip on the protobuf path converts the
- * message to plain objects (protobufjs renders Long as string and bytes as base64).
- */
-function decodePayload(body: Buffer, contentType: string, decoder: OtlpDecoder): OtlpPayload {
+/** Decode an OTLP export body by its content type into a plain, unvalidated object. */
+function decodePayload(body: Buffer, contentType: string, decoder: OtlpDecoder): unknown {
   if (contentType.includes("application/json")) {
-    return JSON.parse(body.toString()) as OtlpPayload;
+    return JSON.parse(body.toString());
   }
-  return JSON.parse(JSON.stringify(decoder.decode(new Uint8Array(body)))) as OtlpPayload;
+  return decodeProtobufToPlainObject(body, decoder);
+}
+
+/**
+ * Decode a protobuf export and flatten it to plain objects. The JSON round-trip
+ * is what does the flattening: protobufjs renders Long as string and bytes as
+ * base64, which is exactly the wire shape the rest of the code reads.
+ */
+function decodeProtobufToPlainObject(body: Buffer, decoder: OtlpDecoder): unknown {
+  return JSON.parse(JSON.stringify(decoder.decode(new Uint8Array(body))));
+}
+
+/**
+ * A payload is only valid when it is a plain object whose export field, if
+ * present, is an array. This rejects non-objects and shapes like
+ * `{ resourceSpans: 5 }` at the 400 boundary instead of letting them fail later
+ * inside the store as a mislabeled persistence error.
+ */
+function isOtlpPayload(
+  value: unknown,
+  field: "resourceSpans" | "resourceLogs",
+): value is OtlpPayload {
+  if (typeof value !== "object" || value === null || Array.isArray(value)) return false;
+  const records = (value as Record<string, unknown>)[field];
+  return records === undefined || Array.isArray(records);
 }
 
 /**
- * Environment for a spawned agent so its OTEL SDK exports to the collector at
- * `port`. Signal-specific variables are set alongside the generic ones because
- * they take precedence in the SDK — a stray OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
- * from the shell or .env.local must not silently redirect traces elsewhere.
- * Per the OTEL spec, signal-specific endpoints are full URLs (the signal path
- * is only appended to the generic endpoint).
+ * Env that points a spawned agent's OTEL SDK at the collector on `port`. While
+ * tracing is on the CLI owns these settings, so nothing from the shell or
+ * .env.local can turn collection off or break it: compression is off (the
+ * collector reads bodies undecompressed) and the SDK and exporters stay on.
+ * Signal-specific endpoints are full URLs and win over the generic one.
  */
 export function otelEnvVars(port: number): Record<string, string> {
   const endpoint = `http://127.0.0.1:${port}`;
@@ -136,6 +163,12 @@ export function otelEnvVars(port: number): Record<string, string> {
     OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf",
     OTEL_EXPORTER_OTLP_TRACES_PROTOCOL: "http/protobuf",
     OTEL_EXPORTER_OTLP_LOGS_PROTOCOL: "http/protobuf",
+    OTEL_EXPORTER_OTLP_COMPRESSION: "none",
+    OTEL_EXPORTER_OTLP_TRACES_COMPRESSION: "none",
+    OTEL_EXPORTER_OTLP_LOGS_COMPRESSION: "none",
+    OTEL_SDK_DISABLED: "false",
+    OTEL_TRACES_EXPORTER: "otlp",
+    OTEL_LOGS_EXPORTER: "otlp",
     OTEL_METRICS_EXPORTER: "none",
     AGENT_OBSERVABILITY_ENABLED: "true",
     OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT: "true",

src/handlers/project/dev/index.test.ts

@@ -188,7 +188,7 @@ describe("project dev trace collection", () => {
     expect(subject.collector.starts).toEqual([
       {
         tracesDirectory: join("/workspace/project", "agentcore", ".cli", "traces", "otlp"),
-        signal: expect.any(AbortSignal),
+        host: "127.0.0.1",
         onError: expect.any(Function),
       },
     ]);
@@ -200,6 +200,15 @@ describe("project dev trace collection", () => {
     expect(subject.collector.state.closed).toBe(1);
   });
 
+  test("binds the collector to all interfaces so a container can reach it", async () => {
+    const subject = harness({
+      project: project(runtime("support", "Container")),
+    });
+    await subject.run();
+
+    expect(subject.collector.starts[0]?.host).toBe("0.0.0.0");
+  });
+
   test("reports a trace-persistence failure once, not per failed export", async () => {
     const subject = harness();
     await subject.run();

src/handlers/project/dev/index.ts

@@ -123,7 +123,9 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) =>
           let tracePersistErrorReported = false;
           collector = await config.startTraceCollector({
             tracesDirectory,
-            signal: controller.signal,
+            // A container reaches the collector over the host bridge, which a
+            // 127.0.0.1 bind refuses, so the container path binds all interfaces.
+            host: runtime.build === "Container" ? "0.0.0.0" : "127.0.0.1",
             // Persistence can fail after startup (disk, permissions). Warn once —
             // exports are still acked, so without this the loss would be silent.
             onError: (error) => {
@@ -167,6 +169,8 @@ export const createDevProjectHandler = (config: DevProjectHandlerConfig) =>
         throw error;
       } finally {
         for (const signal of signals) process.removeListener(signal, interrupt);
+        // Close only after the runner returns, which is after the child's own
+        // shutdown grace, so the agent's final spans still reach the collector.
         await collector?.close();
       }
     },