diff --git a/apps/ai/src/chat/apply-proposal.ts b/apps/ai/src/chat/apply-proposal.ts index 1654dac915..bd0ecde965 100644 --- a/apps/ai/src/chat/apply-proposal.ts +++ b/apps/ai/src/chat/apply-proposal.ts @@ -11,7 +11,6 @@ * connector half of the approval policy; a connector approver's chat account rides along as * forensic context on the audit entry. */ -import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { connectorApprovalTenant, decidedBy, @@ -25,19 +24,17 @@ import type { TenantContext } from "@maple/backend/services/auth/tenant-context" import { OrgMembershipService } from "@maple/backend/services/auth/OrgMembershipService" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { runtimeTelemetry } from "@maple/infra/worker-telemetry" import { Cause, Effect, Layer, ManagedRuntime, Option, Schema } from "effect" import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "../mcp/expected-failures" import { MUTATING_TOOL_NAMES } from "../mcp/tools/mutating" import { toTenantContext, withConnectorActor } from "./turn-actor" /** Its own instance, at module scope, so an approval exports its spans from the isolate it ran in. */ -const telemetry = MapleCloudflareSDK.make( - workerTelemetryConfig({ - serviceName: "maple-chat", - anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, - }), -) +const telemetry = runtimeTelemetry({ + serviceName: "maple-chat", + anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, +}) export type ApplyChatProposalInput = ChatProposalApproval & { readonly env: Record diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 44a8032871..8d3f139605 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -18,14 +18,13 @@ * `env: workerEnv ?? {}` fallback silently degrading the model config when a caller forgot to * thread the worker env through. */ -import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "../mcp/expected-failures" import { ChatMessage, type ChatTurnOrigin, type ChatTurnTenantEncoded } from "@maple/domain/chat-session" import type { PrReviewFailureReason } from "@maple/domain/http" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" import { forkRequestScoped } from "@maple/backend/platform/fork-request-scoped" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { runtimeTelemetry } from "@maple/infra/worker-telemetry" import { Cause, Effect, Exit, Layer, ManagedRuntime, Match, Option, Redacted } from "effect" import { FetchHttpClient } from "effect/http" import type { WorkersAiBinding } from "../platform/WorkersAiHttpClient" @@ -91,13 +90,11 @@ const ENGINE_BOOKKEEPING_SPANS = [ // Deliberately not `maple-api`: background work sharing the request-facing // service's name skewed its percentiles (p99 32s, 2026-09-04). -const telemetry = MapleCloudflareSDK.make( - workerTelemetryConfig({ - serviceName: "maple-chat", - anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, - dropSpanNames: ENGINE_BOOKKEEPING_SPANS, - }), -) +const telemetry = runtimeTelemetry({ + serviceName: "maple-chat", + anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, + dropSpanNames: ENGINE_BOOKKEEPING_SPANS, +}) export interface RunChatSessionTurnInput { /** The Durable Object itself. Appends are direct calls, not stub RPC. */ diff --git a/apps/api/src/http/api-observability.ts b/apps/api/src/http/api-observability.ts index 8384d3f3b7..e34deb7ad7 100644 --- a/apps/api/src/http/api-observability.ts +++ b/apps/api/src/http/api-observability.ts @@ -35,10 +35,11 @@ export const ApiObservabilityLive = Layer.mergeAll( EMAIL_UNSUBSCRIBE_PATH.test(request.url) || /\.(png|ico|jpg|jpeg|gif|css|js|svg|webp|woff2?)(\?.*)?$/i.test(request.url), ), - // Every request header lands on the server span as `http.request.header.`. - // Effect's defaults cover the usual credential headers; the provider webhook - // signatures are ours to add (a GitHub webhook HMAC is replayable alongside its - // body, and there is no reason to retain it). + // Spans record only the allowlisted headers (`MapleHttpTracingLive`), none of + // these; the list still redacts them wherever a `Headers` value is rendered or + // logged. Effect's defaults cover the usual credential headers; the provider + // webhook signatures are ours to add (a GitHub webhook HMAC is replayable + // alongside its body, and there is no reason to retain it). Layer.succeed(Headers.CurrentRedactedNames, [ "authorization", "cookie", diff --git a/apps/chat-bot/src/relay/run.ts b/apps/chat-bot/src/relay/run.ts index f2ade05419..eedc58ae3f 100644 --- a/apps/chat-bot/src/relay/run.ts +++ b/apps/chat-bot/src/relay/run.ts @@ -9,10 +9,9 @@ */ import type { ChatConnector, ConnectorCredentials, InboundEvent } from "@maple/chat-platform" import { connectors } from "@maple/chat-platform/connectors" -import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import type { ChatConnectorId, OrgId } from "@maple/domain/primitives" import type { IntegrationsPersistenceError } from "@maple/domain/http" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { runtimeTelemetry } from "@maple/infra/worker-telemetry" import { ChatSessions, type ChatSessionsApi } from "@maple/backend/platform/bindings" import { type ChatSessionNamespace, chatSessionsLayerIfBound } from "@maple/backend/platform/chat-sessions" import { envPorts } from "@maple/backend/platform/env-ports" @@ -48,12 +47,10 @@ import { CHAT_ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/chat-platform/anticip * for, so the spans a turn produces are exported from here or not at all — and flushed when the * event is done, because nothing else closes a scope around it. */ -const telemetry = MapleCloudflareSDK.make( - workerTelemetryConfig({ - serviceName: "maple-chat-bot", - anticipatedErrorIdentifiers: CHAT_ANTICIPATED_ERROR_IDENTIFIERS, - }), -) +const telemetry = runtimeTelemetry({ + serviceName: "maple-chat-bot", + anticipatedErrorIdentifiers: CHAT_ANTICIPATED_ERROR_IDENTIFIERS, +}) const APP_BASE_URL_FALLBACK = "https://app.maple.dev" diff --git a/apps/electric-sync/src/worker-bridge.test.ts b/apps/electric-sync/src/worker-bridge.test.ts index e693221e6d..85ad410cb4 100644 --- a/apps/electric-sync/src/worker-bridge.test.ts +++ b/apps/electric-sync/src/worker-bridge.test.ts @@ -1,6 +1,6 @@ import { assert, describe, it } from "@effect/vitest" import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" -import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" +import { MapleHttpTracingLive, workerTelemetryConfig } from "@maple/infra/worker-telemetry" import * as Cloudflare from "alchemy/Cloudflare" import { Effect, Exit, Layer, Schema, Scope } from "effect" import { HttpRouter } from "effect/http" @@ -73,7 +73,11 @@ const appLayer = (options: { readonly ensureConfigured?: Effect.Effect = appLayer()) => +const event = ( + path: string, + app: ReturnType = appLayer(), + headers: Record = {}, +) => Effect.gen(function* () { const recorded: Array = [] const realFetch = globalThis.fetch @@ -87,17 +91,21 @@ const event = (path: string, app: ReturnType = appLayer()) => const isolate = yield* Scope.make() const handler = yield* HttpRouter.toHttpEffect(app).pipe(Scope.provide(isolate)) const request = yield* Scope.make() + // What `WorkerTelemetry` registers with alchemy: the SDK and the header allowlist. const services = yield* Layer.buildWithScope( - MapleCloudflareSDK.make( - workerTelemetryConfig({ serviceName: "electric-sync" }), - ).requestLayer.pipe(Layer.provide(Layer.succeed(MapleCloudflareSDK.WorkerEnvironment, env))), + Layer.merge( + MapleCloudflareSDK.make( + workerTelemetryConfig({ serviceName: "electric-sync" }), + ).requestLayer.pipe(Layer.provide(Layer.succeed(MapleCloudflareSDK.WorkerEnvironment, env))), + MapleHttpTracingLive, + ), request, ) const fetchEvent: Effect.Effect | undefined = Cloudflare.Workers.makeRequestHandler(handler)({ kind: "Cloudflare.Workers.WorkerEvent", type: "fetch", - input: new Request(`http://sync.maple.test${path}`), + input: new Request(`http://sync.maple.test${path}`, { headers }), }) assert.isDefined(fetchEvent) const response = yield* fetchEvent!.pipe(Effect.provide(services), Scope.provide(request)) @@ -147,4 +155,18 @@ describe("electric-sync through alchemy's Worker bridge", () => { assert.strictEqual(statusCodeOf(server[0]), 500) }), ) + + it.effect("records the allowlisted request headers on the server span and no credential", () => + Effect.gen(function* () { + const { server } = yield* event("/nope", appLayer(), { + "cf-ray": "8f1a2b3c4d5e6f70-IAD", + authorization: "Bearer secret", + cookie: "session=secret", + }) + const keys = (server[0]?.attributes ?? []).map((attribute) => attribute.key) + assert.include(keys, "http.request.header.cf-ray") + assert.notInclude(keys, "http.request.header.authorization") + assert.notInclude(keys, "http.request.header.cookie") + }), + ) }) diff --git a/packages/effect-sdk/src/shared/flushable-tracer.test.ts b/packages/effect-sdk/src/shared/flushable-tracer.test.ts index d575bd4110..3385b039d4 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.test.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.test.ts @@ -286,6 +286,41 @@ describe("makeSpanBuffer rendered 5xx responses", () => { }), ) + // `@effect/platform-node` builds these when writing the response fails. + const writeFailures = [ + ["a cause", { cause: new Error("socket hang up") }], + ["a description", { description: "Error writing raw response" }], + ["both", { description: "Error writing raw response", cause: new Error("socket hang up") }], + ] as const + + for (const [label, detail] of writeFailures) { + it.effect(`keeps a failed response write carrying ${label} as its own error`, () => + Effect.gen(function* () { + const buffer = makeSpanBuffer() + const error = new HttpServerError.HttpServerError({ + reason: new HttpServerError.ResponseError({ + request: HttpServerRequest.fromWeb(new Request("http://localhost/api/sync/shape")), + response: HttpServerResponse.empty({ status: 500 }), + ...detail, + }), + }) + yield* Effect.annotateCurrentSpan({ "http.response.status_code": 500 }).pipe( + Effect.andThen(Effect.fail(error)), + Effect.withSpan("GET", { kind: "server" }), + Effect.provide(buffer.tracerLayer), + Effect.ignore, + ) + const [span] = buffer.drain() + assert.isDefined(span) + assert.strictEqual(span!.status.code, 2 /* Error */) + const types = span!.events + .filter((event) => event.name === "exception") + .map((event) => event.attributes.find((attribute) => attribute.key === "exception.type")?.value) + assert.deepStrictEqual(types, [{ stringValue: "HttpServerError" }]) + }), + ) + } + it.effect("keeps an exception the handler recorded itself instead of relabelling it", () => Effect.gen(function* () { const buffer = makeSpanBuffer() diff --git a/packages/effect-sdk/src/shared/flushable-tracer.ts b/packages/effect-sdk/src/shared/flushable-tracer.ts index 371863d706..e3c377c5e3 100644 --- a/packages/effect-sdk/src/shared/flushable-tracer.ts +++ b/packages/effect-sdk/src/shared/flushable-tracer.ts @@ -259,7 +259,13 @@ const renderedServerError = (self: SpanImpl): number | undefined => { } // Effect 4.0.2+ fails a 5xx server span itself, with an `HttpServerError` whose reason is -// `ResponseError`. That names nothing the rendered 5xx did not, so it takes the same road. +// `ResponseError`. That names nothing the rendered 5xx did not, so it takes the same road. One +// carrying a cause or description is a failed response write, a real error of its own. +const isStatusOnlyResponseError = (error: unknown): boolean => + Predicate.isTagged(error, "ResponseError") && + (!Predicate.hasProperty(error, "cause") || error.cause === undefined) && + (!Predicate.hasProperty(error, "description") || error.description === undefined) + const isResponseFailure = (exit: Exit.Exit): boolean => { if (exit._tag === "Success") return true const [reason, ...rest] = exit.cause.reasons @@ -269,7 +275,7 @@ const isResponseFailure = (exit: Exit.Exit): boolean => { Cause.isFailReason(reason) && Predicate.isTagged(reason.error, "HttpServerError") && Predicate.hasProperty(reason.error, "reason") && - Predicate.isTagged(reason.error.reason, "ResponseError") + isStatusOnlyResponseError(reason.error.reason) ) } diff --git a/packages/infra/src/cloudflare/worker-telemetry.test.ts b/packages/infra/src/cloudflare/worker-telemetry.test.ts new file mode 100644 index 0000000000..6f6b4709de --- /dev/null +++ b/packages/infra/src/cloudflare/worker-telemetry.test.ts @@ -0,0 +1,54 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect, Layer, Tracer } from "effect" +import { FetchHttpClient, HttpClient, HttpClientRequest } from "effect/http" +import { MapleHttpTracingLive } from "./worker-telemetry.ts" + +describe("MapleHttpTracingLive", () => { + it.effect("records the allowlisted headers and never a credential", () => + Effect.gen(function* () { + const spans: Array = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, + }) + const client = yield* HttpClient.HttpClient + yield* client + .execute( + HttpClientRequest.post("https://ch.example.test/").pipe( + HttpClientRequest.setHeaders({ + "Content-Type": "text/plain", + "X-ClickHouse-Key": "secret", + Authorization: "Bearer secret", + }), + ), + ) + .pipe(Effect.withTracer(tracer)) + const attributes = spans[0]?.attributes + assert.strictEqual(attributes?.get("http.request.header.content-type"), "text/plain") + assert.strictEqual(attributes?.get("http.response.header.x-ratelimit-remaining"), "9") + assert.isFalse(attributes?.has("http.request.header.x-clickhouse-key")) + assert.isFalse(attributes?.has("http.request.header.authorization")) + assert.isFalse(attributes?.has("http.response.header.set-cookie")) + }).pipe( + Effect.provide( + Layer.mergeAll( + FetchHttpClient.layer.pipe( + Layer.provide( + Layer.succeed( + FetchHttpClient.Fetch, + async () => + new Response("ok", { + headers: { "x-ratelimit-remaining": "9", "set-cookie": "session=secret" }, + }), + ), + ), + ), + MapleHttpTracingLive, + ), + ), + ), + ) +}) diff --git a/packages/infra/src/cloudflare/worker-telemetry.ts b/packages/infra/src/cloudflare/worker-telemetry.ts index 36f6384939..bb3a1ad856 100644 --- a/packages/infra/src/cloudflare/worker-telemetry.ts +++ b/packages/infra/src/cloudflare/worker-telemetry.ts @@ -6,10 +6,70 @@ import { Telemetry, type TelemetrySdkOptions } from "@maple-dev/alchemy/telemetr import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { ANTICIPATED_ERROR_IDENTIFIERS } from "@maple/domain/anticipated-errors" import { WorkerEnvironment } from "alchemy/Cloudflare" +import * as AlchemyTelemetry from "alchemy/Telemetry" +import { HttpClient, HttpMiddleware } from "effect/http" import * as Layer from "effect/Layer" export const MAPLE_REPOSITORY_URL = "https://github.com/MapleTechLabs/maple" +/** + * The headers worth recording on HTTP spans, by phase. effect 4.0.4 records none by default; an + * allowlist rather than "all but redacted" because outgoing requests carry credentials no + * redaction list knows about (a customer's ClickHouse password goes out as `x-clickhouse-key`). + */ +const RECORDED_HEADERS = { + // Which host and client a request came through: OAuth redirect URIs are built from these. No + // client IP: country is enough, and an address on every span is personal data. + request: new Set([ + "host", + "x-forwarded-host", + "x-forwarded-proto", + "origin", + "accept", + "content-type", + "content-length", + "cf-ray", + "cf-ipcountry", + "mcp-protocol-version", + ]), + // Vendor request ids, rate-limit state and auth challenges: what a 429, a 401 or a support + // ticket needs. ClickHouse's query id is the key into the customer's `system.query_log`. + response: new Set([ + "content-type", + "content-length", + "cf-ray", + "cf-placement", + "cf-cache-status", + "www-authenticate", + "retry-after", + "request-id", + "x-request-id", + "x-github-request-id", + "x-slack-req-id", + "x-clickhouse-query-id", + "x-clickhouse-exception-code", + "x-clickhouse-summary", + "ratelimit", + "ratelimit-policy", + "ratelimit-limit", + "ratelimit-remaining", + "ratelimit-reset", + "x-ratelimit-limit", + "x-ratelimit-remaining", + "x-ratelimit-reset", + ]), +} as const + +// Effect lowercases header names before the filter sees them. +const isRecordedHeader = (name: string, phase: "request" | "response"): boolean => + RECORDED_HEADERS[phase].has(name) + +/** Header capture for every HTTP span a Maple Worker records, server and client. */ +export const MapleHttpTracingLive = Layer.mergeAll( + Layer.succeed(HttpMiddleware.TracerHeaderFilter, isRecordedHeader), + Layer.succeed(HttpClient.TracerHeaderFilter, isRecordedHeader), +) + export interface WorkerTelemetryOptions { readonly serviceName: string /** Added to the domain's `ANTICIPATED_ERROR_IDENTIFIERS` (e.g. the MCP set). */ @@ -31,7 +91,11 @@ export const workerTelemetryConfig = (options: WorkerTelemetryOptions): Telemetr }) export const WorkerTelemetry = (options: WorkerTelemetryOptions): Layer.Layer => - Telemetry(workerTelemetryConfig(options)) + Layer.mergeAll( + Telemetry(workerTelemetryConfig(options)), + // Read by the bridge's HTTP tracer, so it is built into every event with the exporters. + AlchemyTelemetry.layer(MapleHttpTracingLive), + ) /** * The SDK under another service name, for background work (queue, cron) whose spans must not @@ -40,7 +104,20 @@ export const WorkerTelemetry = (options: WorkerTelemetryOptions): Layer.Layer => - MapleCloudflareSDK.make(workerTelemetryConfig(options)).requestLayer.pipe( - // Same key as alchemy's tag; a type bridge only. - Layer.provide(Layer.effect(MapleCloudflareSDK.WorkerEnvironment, WorkerEnvironment)), + Layer.mergeAll( + MapleCloudflareSDK.make(workerTelemetryConfig(options)).requestLayer.pipe( + // Same key as alchemy's tag; a type bridge only. + Layer.provide(Layer.effect(MapleCloudflareSDK.WorkerEnvironment, WorkerEnvironment)), + ), + MapleHttpTracingLive, ) + +/** + * The SDK for a runtime the Worker bridge does not wrap (a Durable Object's own + * `ManagedRuntime`), its `layer` already carrying {@link MapleHttpTracingLive}. Call once at + * module scope: buffers are per isolate. + */ +export const runtimeTelemetry = (options: WorkerTelemetryOptions): MapleCloudflareSDK.Telemetry => { + const telemetry = MapleCloudflareSDK.make(workerTelemetryConfig(options)) + return { ...telemetry, layer: Layer.merge(telemetry.layer, MapleHttpTracingLive) } +}