feat(telemetry): OpenTelemetry export across all runtimes via Effect's OTLP exporters

packages/alchemy/src/AWS/EC2/hosted.ts

@@ -189,6 +189,7 @@ import { BunServices } from "@effect/platform-bun";
 import { BunHttpServer } from "alchemy/Http";
 import { Stack } from "alchemy/Stack";
 import { reifyBoundConfigProvider } from "alchemy/Runtime";
+import { provideProcessTelemetry } from "alchemy/Telemetry";
 import * as Config from "effect/Config";
 import * as ConfigProvider from "effect/ConfigProvider";
 import * as Credentials from "@distilled.cloud/aws/Credentials";
@@ -210,8 +211,15 @@ const platform = Layer.mergeAll(
 // and run it with a Bun HTTP server bound to PORT, so a returned { fetch }
 // handler is actually served and host.run loops stay alive.
 const program = handler.pipe(
-  Effect.flatMap((instance) => instance.RuntimeContext.exports),
-  Effect.flatMap((exports) => exports.program),
+  // Process-lifetime telemetry: built once into the root scope; exporters
+  // batch on their intervals and flush when the scope closes on graceful
+  // shutdown.
+  Effect.flatMap((instance) =>
+    instance.RuntimeContext.exports.pipe(
+      Effect.flatMap((exports) => exports.program),
+      provideProcessTelemetry(instance.RuntimeContext),
+    ),
+  ),
   Effect.provide(
     Layer.effect(
       Stack,

packages/alchemy/src/AWS/ECS/Task.ts

@@ -586,6 +586,7 @@ import { BunServices } from "@effect/platform-bun";
 import { BunHttpServer } from "alchemy/Http";
 import { Stack } from "alchemy/Stack";
 import { reifyBoundConfigProvider } from "alchemy/Runtime";
+import { provideProcessTelemetry } from "alchemy/Telemetry";
 import * as Config from "effect/Config";
 import * as ConfigProvider from "effect/ConfigProvider";
 import * as Credentials from "@distilled.cloud/aws/Credentials";
@@ -607,8 +608,15 @@ const platform = Layer.mergeAll(
 // and run it with a Bun HTTP server bound to PORT, so a returned { fetch }
 // handler is actually served and host.run loops stay alive.
 const program = handler.pipe(
-  Effect.flatMap((task) => task.RuntimeContext.exports),
-  Effect.flatMap((exports) => exports.program),
+  // Process-lifetime telemetry: built once into the root scope; exporters
+  // batch on their intervals and flush when the scope closes on graceful
+  // shutdown.
+  Effect.flatMap((task) =>
+    task.RuntimeContext.exports.pipe(
+      Effect.flatMap((exports) => exports.program),
+      provideProcessTelemetry(task.RuntimeContext),
+    ),
+  ),
   Effect.provide(
     Layer.effect(
       Stack,

packages/alchemy/src/AWS/Lambda/Function.ts

@@ -41,6 +41,7 @@ import { Resource, type ResourceBinding } from "../../Resource.ts";
 import { packEnvValue, unpackEnvValue } from "../../RuntimeContext.ts";
 import { Self } from "../../Self.ts";
 import * as Serverless from "../../Serverless/index.ts";
+import { buildEventTelemetry } from "../../Telemetry.ts";
 import { Stack } from "../../Stack.ts";
 import { Stage } from "../../Stage.ts";
 import {
@@ -613,53 +614,74 @@ export const Function: Platform<
       exports: Effect.sync(() => ({
         // construct an Effect that produces the Function's entrypoint
         // Effect<(event, context) => Promise<any>>
-        handler: Effect.map(
-          Effect.all(listeners, {
+        handler: Effect.gen(function* () {
+          const handlers = yield* Effect.all(listeners, {
             concurrency: "unbounded",
-          }),
-          (handlers) =>
-            async (event: any, context: lambda.Context): Promise<any> => {
-              for (const handler of handlers) {
-                const eff = handler(event);
-                if (Effect.isEffect(eff)) {
-                  // Each invocation gets a fresh request scope, matching the
-                  // Worker / Durable Object / Workflow bridges. The scope is
-                  // settled inline before returning: a buffered Lambda
-                  // response is not released to the caller until the Invoke
-                  // phase completes, so deferring cleanup (e.g. via an
-                  // INVOKE-subscribed extension window) shows up as response
-                  // latency anyway — keep request finalizers fast. A failing
-                  // finalizer is logged and ignored so it can't mask the
-                  // invocation's outcome.
-                  const scope = Scope.makeUnsafe();
-                  const exit = await eff.pipe(
-                    Effect.provide(
-                      Layer.mergeAll(
-                        Layer.succeed(HandlerContext, context),
-                        Layer.succeed(Scope.Scope, scope),
-                      ),
-                    ),
-                    Effect.tap(Effect.logDebug),
-                    Effect.runPromiseExit,
+          });
+          // Sandbox-lifetime services, captured so each invocation can
+          // build its telemetry exporters and run the handler effect
+          // against the same context the init phase saw (mirrors
+          // WorkerBridge). The build's memo map is stripped so a Layer the
+          // user `Effect.provide`s inside a handler builds per invocation.
+          const services = Context.omit(Layer.CurrentMemoMap)(
+            yield* Effect.context<never>(),
+          );
+          return async (event: any, context: lambda.Context): Promise<any> => {
+            for (const handler of handlers) {
+              const eff = handler(event);
+              if (Effect.isEffect(eff)) {
+                // Each invocation gets a fresh request scope, matching the
+                // Worker / Durable Object / Workflow bridges. The scope is
+                // settled inline before returning: a buffered Lambda
+                // response is not released to the caller until the Invoke
+                // phase completes, so deferring cleanup (e.g. via an
+                // INVOKE-subscribed extension window) shows up as response
+                // latency anyway — keep request finalizers fast. A failing
+                // finalizer is logged and ignored so it can't mask the
+                // invocation's outcome.
+                const scope = Scope.makeUnsafe();
+                // Telemetry exporters build into the invocation scope so
+                // buffered spans/logs/metrics flush when it settles below.
+                const exit = await buildEventTelemetry(
+                  services,
+                  scope,
+                  (ctx as Serverless.FunctionContext).telemetry,
+                ).pipe(
+                  Effect.flatMap((telemetry) =>
+                    Effect.provideContext(eff, telemetry),
+                  ),
+                  Effect.provide(
+                    Layer.mergeAll(
+                      Layer.succeed(HandlerContext, context),
+                      Layer.succeed(Scope.Scope, scope),
+                    ).pipe(Layer.provideMerge(Layer.succeedContext(services))),
+                  ),
+                  Effect.tap(Effect.logDebug),
+                  Effect.runPromiseExit,
+                );
+                if (!isScopeEjected(scope)) {
+                  // The HttpMiddleware tracer ends the request's root span
+                  // in a dispatcher task scheduled after the handler effect
+                  // resolves; yield one macrotask so it reaches the
+                  // telemetry exporter's buffer before the flush finalizer.
+                  await new Promise((resolve) => setTimeout(resolve, 0));
+                  await Scope.close(scope, exit).pipe(
+                    Effect.ignoreCause({
+                      log: "Warn",
+                      message: "Lambda invocation scope close failed",
+                    }),
+                    Effect.runPromise,
                   );
-                  if (!isScopeEjected(scope)) {
-                    await Scope.close(scope, exit).pipe(
-                      Effect.ignoreCause({
-                        log: "Warn",
-                        message: "Lambda invocation scope close failed",
-                      }),
-                      Effect.runPromise,
-                    );
-                  }
-                  if (Exit.isSuccess(exit)) {
-                    return exit.value;
-                  }
-                  throw Cause.squash(exit.cause);
                 }
+                if (Exit.isSuccess(exit)) {
+                  return exit.value;
+                }
+                throw Cause.squash(exit.cause);
               }
-              throw new Error("No event handler found");
-            },
-        ),
+            }
+            throw new Error("No event handler found");
+          };
+        }),
       })),
     };
     return ctx;

packages/alchemy/src/AWS/Lambda/HttpServer.ts

@@ -7,6 +7,7 @@ import type {
 import * as Effect from "effect/Effect";
 import * as Option from "effect/Option";
 import type { Scope } from "effect/Scope";
+import * as HttpMiddleware from "effect/unstable/http/HttpMiddleware";
 import * as HttpServerRequest from "effect/unstable/http/HttpServerRequest";
 import * as HttpServerResponse from "effect/unstable/http/HttpServerResponse";
 import * as Http from "../../Http.ts";
@@ -32,7 +33,12 @@ export const isApiGatewayProxyEvent = (
 };
 
 export const makeFunctionHttpHandler = <Req>(handler: Http.HttpEffect<Req>) => {
-  const safeHandler = Http.safeHttpEffect(handler);
+  // `HttpMiddleware.tracer` creates the `http.server` root span per request
+  // (continuing an incoming `traceparent`), matching the Worker bridge's
+  // fetch path. With the default no-op tracer this is free; with a telemetry
+  // exporter installed the span is exported when the invocation scope
+  // flushes.
+  const safeHandler = HttpMiddleware.tracer(Http.safeHttpEffect(handler));
   return (
     event: any,
   ):

packages/alchemy/src/AWS/Lambda/MicrovmBundle.ts

@@ -136,6 +136,7 @@ const HttpServer = NodeHttpServer;`
 }
 import { Stack } from "alchemy/Stack";
 import { makeEntrypointLayer } from "alchemy/Runtime";
+import { provideProcessTelemetry } from "alchemy/Telemetry";
 import * as Effect from "effect/Effect";
 import * as FetchHttpClient from "effect/unstable/http/FetchHttpClient";
 import * as Layer from "effect/Layer";
@@ -162,8 +163,15 @@ const stack = Layer.succeed(Stack, {
 });
 
 const serverEffect = tag.pipe(
-  Effect.flatMap(func => func.RuntimeContext.exports),
-  Effect.flatMap(exports => exports.default),
+  // Process-lifetime telemetry: built once into the root scope; exporters
+  // batch on their intervals and flush when the scope closes on graceful
+  // shutdown.
+  Effect.flatMap((func) =>
+    func.RuntimeContext.exports.pipe(
+      Effect.flatMap((exports) => exports.default),
+      provideProcessTelemetry(func.RuntimeContext),
+    ),
+  ),
   Effect.provide(
     layer.pipe(
       Layer.provideMerge(stack),

packages/alchemy/src/Axiom/Telemetry.ts

@@ -0,0 +1,119 @@
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Redacted from "effect/Redacted";
+import type { Input } from "../Input.ts";
+import * as Output from "../Output.ts";
+import { layerOtlp, type OtlpSignalOptions } from "../Telemetry.ts";
+import type { ApiToken } from "./ApiToken.ts";
+import type { Dataset } from "./Dataset.ts";
+
+/**
+ * A resource passed to the layer: either the module-scope declaration (an
+ * Effect that resolves to the instance, same as what capability bindings
+ * accept) or an already-yielded instance.
+ */
+export type ResourceInput<T> = T | Effect.Effect<T, never, any>;
+
+/**
+ * Options for {@link Telemetry}: an ingest {@link ApiToken} plus one
+ * {@link Dataset} per OTel signal to export.
+ */
+export interface AxiomTelemetryProps {
+  /**
+   * The ingest credential. Must have `ingest: ["create"]` capability on
+   * every dataset passed below.
+   */
+  token: ResourceInput<ApiToken>;
+  /** Dataset (kind `otel:traces:v1`) to export traces into. */
+  traces?: ResourceInput<Dataset> | undefined;
+  /** Dataset (kind `otel:logs:v1`) to export logs into. */
+  logs?: ResourceInput<Dataset> | undefined;
+  /** Dataset (kind `otel:metrics:v1`) to export metrics into. */
+  metrics?: ResourceInput<Dataset> | undefined;
+  /**
+   * The exported `service.name`.
+   * @default the deployed Function/Worker's physical name
+   */
+  serviceName?: Input<string> | undefined;
+}
+
+/**
+ * Resource declarations are Effects — yield them to get the instance with
+ * attribute Output accessors, same as `Binding.Service`'s callable does for
+ * capability bindings.
+ */
+const instance = <T>(resource: ResourceInput<T>): Effect.Effect<T> =>
+  Effect.isEffect(resource)
+    ? (resource as Effect.Effect<T>)
+    : Effect.succeed(resource);
+
+const signal = (
+  token: ApiToken,
+  dataset: Dataset | undefined,
+  urlAttr: "otelTracesEndpoint" | "otelLogsEndpoint" | "otelMetricsEndpoint",
+): OtlpSignalOptions | undefined =>
+  dataset === undefined
+    ? undefined
+    : {
+        url: dataset[urlAttr],
+        headers: {
+          Authorization: Output.map(
+            token.token,
+            (bearer) =>
+              Redacted.make(
+                `Bearer ${Redacted.value(bearer)}`,
+              ) as Redacted.Redacted<string>,
+          ),
+          "X-Axiom-Dataset": dataset.name,
+        },
+      };
+
+/**
+ * Export a Function/Worker's telemetry to Axiom.
+ *
+ * A binding layer over {@link layerOtlp | Alchemy.Telemetry.layerOtlp}:
+ * building it binds each dataset's OTLP endpoint and the ingest token's
+ * `Authorization` header (as a secret) onto the host, and at runtime the
+ * built-in exporter ships each signal to its dataset, flushed per event.
+ *
+ * Compose it into the Function/Worker's single `Effect.provide`:
+ *
+ * ```ts
+ * import * as Axiom from "alchemy/Axiom";
+ * import { Bucket } from "./bucket.ts";
+ * import { Ingest, Logs, Traces } from "./observability.ts";
+ *
+ * export default Cloudflare.Worker(
+ *   "Worker",
+ *   { main: import.meta.url },
+ *   Effect.gen(function* () {
+ *     // ...
+ *   }).pipe(
+ *     Effect.provide(
+ *       Layer.mergeAll(
+ *         Cloudflare.R2.ReadWriteBucketBinding,
+ *         Axiom.Telemetry({ token: Ingest, traces: Traces, logs: Logs }),
+ *       ),
+ *     ),
+ *   ),
+ * );
+ * ```
+ */
+export const Telemetry = (props: AxiomTelemetryProps): Layer.Layer<never> =>
+  Layer.unwrap(
+    Effect.gen(function* () {
+      // Declarations are yielded to instances (registering them on the
+      // Stack if the enclosing host is the first to reference them), so
+      // attribute accessors below produce real Outputs.
+      const token = yield* instance(props.token);
+      const traces = props.traces && (yield* instance(props.traces));
+      const logs = props.logs && (yield* instance(props.logs));
+      const metrics = props.metrics && (yield* instance(props.metrics));
+      return layerOtlp({
+        serviceName: props.serviceName,
+        traces: signal(token, traces, "otelTracesEndpoint"),
+        logs: signal(token, logs, "otelLogsEndpoint"),
+        metrics: signal(token, metrics, "otelMetricsEndpoint"),
+      });
+    }),
+  ) as Layer.Layer<never>;

packages/alchemy/src/Axiom/index.ts

@@ -7,6 +7,7 @@ export * from "./Dashboard.ts";
 export * from "./Dataset.ts";
 export * from "./Monitor.ts";
 export * from "./Notifier.ts";
+export * from "./Telemetry.ts";
 export * from "./Providers.ts";
 export * from "./View.ts";
 export * from "./VirtualField.ts";

packages/alchemy/src/Cloudflare/Containers/ContainerBundle.ts

@@ -200,6 +200,7 @@ const HttpServer = NodeHttpServer;
 }
 import { Stack } from "alchemy/Stack";
 import { makeEntrypointLayer, reifyBoundConfigProvider } from "alchemy/Runtime";
+import { provideProcessTelemetry } from "alchemy/Telemetry";
 import { CloudflareEnvironment } from "alchemy/Cloudflare";
 import * as ConfigProvider from "effect/ConfigProvider";
 import * as Effect from "effect/Effect";
@@ -229,8 +230,15 @@ const stack = Layer.succeed(Stack, {
 });
 
 const serverEffect = tag.pipe(
-  Effect.flatMap(func => func.RuntimeContext.exports),
-  Effect.flatMap(exports => exports.default),
+  // Process-lifetime telemetry: built once into the root scope; exporters
+  // batch on their intervals and flush when the scope closes on graceful
+  // shutdown.
+  Effect.flatMap((func) =>
+    func.RuntimeContext.exports.pipe(
+      Effect.flatMap((exports) => exports.default),
+      provideProcessTelemetry(func.RuntimeContext),
+    ),
+  ),
   Effect.provide(
     layer.pipe(
       Layer.provideMerge(stack),