fix(telemetry): keep OpenRouter Broadcast traces attached to their Maple session

apps/ai/src/chat/turn-runner.ts

@@ -54,6 +54,50 @@ const telemetry = MapleCloudflareSDK.make(
 	}),
 )
 
+/**
+ * How often running turns ship the spans they have finished so far.
+ *
+ * A turn is background work on the Durable Object: no request holds the isolate open for it, and
+ * an investigation has no subscriber at all. A single flush at the end of the turn was therefore
+ * one fetch that had to survive the object's whole remaining life — and on 2026-09-17 it did not
+ * for 22 of 101 investigation passes: OpenRouter's Broadcast mirror arrived nested under span ids
+ * the warehouse never saw, and the Agent Sessions list showed the pass as an OpenRouter-only
+ * session with no agent. Flushing on an interval bounds a lost flush to the tail of the turn.
+ *
+ * One timer per isolate, not per turn: `telemetry` and its buffers are shared by every
+ * `ChatSession` in the isolate, so a timer per turn would drain the same buffer N times a window
+ * and re-export the cumulative metric snapshot each time. The SDK resolves its endpoint from the
+ * first `env` it sees, so which turn's `env` the timer captured does not matter.
+ *
+ * The trade: a POST that fails after the collector persisted it is retried by the next tick, so
+ * a span can now land twice. Session usage nets by response id; raw span counts do not.
+ */
+const FLUSH_INTERVAL_MS = 10_000
+/**
+ * How long the turn waits for `runtime.dispose()`. Its finalizers — the Postgres connection, the
+ * model client — are the one part of the turn that can hang, and past this the second flush and
+ * the session's `endTurn` matter more than a clean close.
+ */
+const DISPOSE_TIMEOUT = "5 seconds"
+let liveTurns = 0
+let flushTimer: ReturnType<typeof setInterval> | undefined
+
+const retainFlushTimer = (env: Record<string, unknown>): void => {
+	liveTurns += 1
+	if (flushTimer !== undefined) return
+	// Host timer on purpose: it must outlive every turn's Effect runtime, and `flush` is a Promise
+	// API that never rejects (`guardFlush`).
+	// oxlint-disable-next-line effecttsgo/global-timers
+	flushTimer = setInterval(() => void telemetry.flush(env), FLUSH_INTERVAL_MS)
+}
+
+const releaseFlushTimer = (): void => {
+	liveTurns -= 1
+	if (liveTurns > 0 || flushTimer === undefined) return
+	clearInterval(flushTimer)
+	flushTimer = undefined
+}
+
 export interface RunChatSessionTurnInput {
 	/** The Durable Object itself. Appends are direct calls, not stub RPC. */
 	readonly session: ChatSession
@@ -423,6 +467,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis
 		}),
 	)
 
+	retainFlushTimer(input.env)
 	try {
 		await runtime.runPromise(program)
 	} catch {
@@ -437,7 +482,16 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis
 			})
 		}
 	} finally {
-		await runtime.dispose().catch(() => undefined)
-		await telemetry.flush(input.env).catch(() => undefined)
+		releaseFlushTimer()
+		// The turn's own spans have ended by now; ship them before `dispose`, whose finalizers (the
+		// Postgres connection, the model client) are the one part of the turn that can still hang.
+		// `force`: these two are the last flushes this turn makes, so a tick's failed POST must not
+		// leave them skipped inside the cooldown. `flush` never rejects and serializes overlapping
+		// calls — a tick in flight finishes (or times out) first, then this one drains.
+		await telemetry.flush(input.env, { force: true })
+		await Effect.runPromise(
+			Effect.promise(() => runtime.dispose()).pipe(Effect.timeout(DISPOSE_TIMEOUT), Effect.ignoreCause),
+		)
+		await telemetry.flush(input.env, { force: true })
 	}
 }

apps/slack-agent/agent/lib/agent-model.test.ts

@@ -2,6 +2,7 @@ import { afterEach, describe, expect, test } from "bun:test"
 import { generateText, type LanguageModel } from "ai"
 import { agentModel } from "./agent-model.js"
 import { installFetchStub, type FetchStub } from "./fetch-stub.js"
+import { withActiveSpan } from "./test-span.js"
 
 type StepStarted = NonNullable<(typeof agentModel.events)["step.started"]>
 
@@ -48,6 +49,18 @@ describe("agentModel", () => {
 		expect(body.usage).toEqual({ include: true })
 	})
 
+	test("a call made under a span nests OpenRouter's mirror under it", async () => {
+		const { ids, body } = await withActiveSpan("ai.streamText.doStream", async (ids) => ({
+			ids,
+			body: await capturedBody(agentModel.fallback),
+		}))
+		expect(body.trace).toEqual({
+			trace_name: "slack",
+			trace_id: ids.traceId,
+			parent_span_id: ids.spanId,
+		})
+	})
+
 	test("the fallback, used before any session exists, carries none", async () => {
 		const body = await capturedBody(agentModel.fallback)
 		expect(body).not.toHaveProperty("session_id")

apps/slack-agent/agent/lib/agent-model.ts

@@ -1,5 +1,6 @@
 import { createOpenRouter } from "@openrouter/ai-sdk-provider"
 import { defineDynamic } from "eve"
+import { openRouterFetch } from "./openrouter-trace.js"
 
 /**
  * OpenRouter over its REST API.
@@ -8,12 +9,14 @@ import { defineDynamic } from "eve"
  * traffic to Maple's app page on openrouter.ai. Same URL and title as `apps/api` on purpose: the
  * referer is the app's identity, so a different one here would mint a second app entry and split
  * the rankings. Surfaces are told apart by `trace.trace_name` instead — static, because this
- * process only ever is the Slack agent.
+ * process only ever is the Slack agent. `fetch` adds the calling span's ids per request, which is
+ * what nests OpenRouter's Broadcast mirror of the call under this agent's own trace.
  */
 const openrouter = createOpenRouter({
 	apiKey: process.env.OPENROUTER_API_KEY ?? "",
 	appUrl: "https://maple.dev",
 	appName: "Maple",
+	fetch: openRouterFetch,
 	extraBody: { trace: { trace_name: "slack" } },
 })
 

apps/slack-agent/agent/lib/follow-up-relevance.ts

@@ -1,5 +1,6 @@
 import { generateText } from "ai"
 import { createOpenRouter } from "@openrouter/ai-sdk-provider"
+import { openRouterFetch } from "./openrouter-trace.js"
 import type { SlackThreadMessage } from "eve/channels/slack"
 import {
 	formatContextBlock,
@@ -104,6 +105,7 @@ function buildDefaultDeps(): FollowUpRelevanceDeps {
 		apiKey: process.env.OPENROUTER_API_KEY ?? "",
 		appUrl: "https://maple.dev",
 		appName: "Maple",
+		fetch: openRouterFetch,
 		extraBody: { trace: { trace_name: "slack" } },
 	})
 	const model = openrouter(gateModelId())

apps/slack-agent/agent/lib/openrouter-trace.test.ts

@@ -0,0 +1,68 @@
+import { afterEach, describe, expect, test } from "bun:test"
+import { installFetchStub, type FetchStub } from "./fetch-stub.js"
+import { openRouterFetch } from "./openrouter-trace.js"
+import { withActiveSpan } from "./test-span.js"
+
+let stub: FetchStub | undefined
+
+afterEach(() => {
+	stub?.restore()
+	stub = undefined
+})
+
+const URL = "https://openrouter.ai/api/v1/chat/completions"
+const post = (body: string) => openRouterFetch(URL, { method: "POST", body })
+
+const sent = (index = 0): Record<string, unknown> => {
+	const body = stub?.calls[index]?.body
+	if (typeof body !== "string") throw new Error("no request reached the transport")
+	// SAFETY: the test posted a JSON object below and reads its keys one at a time.
+	return JSON.parse(body) as Record<string, unknown>
+}
+
+describe("openRouterFetch", () => {
+	test("stamps the active span's ids into the request's trace object", async () => {
+		stub = installFetchStub(() => new Response("{}"))
+		const ids = await withActiveSpan("ai.streamText.doStream", async (ids) => {
+			await post(JSON.stringify({ model: "m", trace: { trace_name: "slack" } }))
+			return ids
+		})
+		expect(sent().trace).toEqual({
+			trace_name: "slack",
+			trace_id: ids.traceId,
+			parent_span_id: ids.spanId,
+		})
+		expect(sent().model).toBe("m")
+	})
+
+	test("each call carries its own span, not the first one's", async () => {
+		stub = installFetchStub(() => new Response("{}"))
+		const first = await withActiveSpan("call 1", async (ids) => {
+			await post(JSON.stringify({ model: "m" }))
+			return ids
+		})
+		const second = await withActiveSpan("call 2", async (ids) => {
+			await post(JSON.stringify({ model: "m" }))
+			return ids
+		})
+		expect(first.spanId).not.toBe(second.spanId)
+		expect(sent(0).trace).toEqual({ trace_id: first.traceId, parent_span_id: first.spanId })
+		expect(sent(1).trace).toEqual({ trace_id: second.traceId, parent_span_id: second.spanId })
+	})
+
+	test("leaves the request alone when no span is active", async () => {
+		stub = installFetchStub(() => new Response("{}"))
+		await post(JSON.stringify({ model: "m", trace: { trace_name: "slack" } }))
+		expect(sent().trace).toEqual({ trace_name: "slack" })
+	})
+
+	test("passes a body that is not a JSON object through untouched", async () => {
+		stub = installFetchStub(() => new Response("{}"))
+		await withActiveSpan("ai.streamText.doStream", async () => {
+			await post("not json")
+			await post("[1]")
+		})
+		expect(stub.calls[0]?.body).toBe("not json")
+		expect(stub.calls[1]?.body).toBe("[1]")
+	})
+})

apps/slack-agent/agent/lib/openrouter-trace.ts

@@ -0,0 +1,50 @@
+import { isSpanContextValid, trace, TraceFlags } from "@opentelemetry/api"
+
+/**
+ * The `fetch` the OpenRouter provider sends its requests through: it stamps the calling span's
+ * W3C ids onto the request as `trace.trace_id` / `trace.parent_span_id`.
+ *
+ * OpenRouter's Broadcast export uses them verbatim, so the `LLM Generation` trace it sends Maple
+ * for every call (provider attempts, fallbacks, router latency) nests under the AI SDK span that
+ * made the call. Without them each call arrived as a root trace of its own — the session still
+ * held them, through `session_id`, but as one detached trace per model call. The provider's
+ * `extraBody` cannot carry the ids: it is fixed when the model is built, and the ids are per call.
+ *
+ * Only a sampled span is stamped: a sampled-out one has valid-looking ids that Maple will never
+ * receive, and a mirror nested under those would be an orphan rather than a root trace.
+ *
+ * `globalThis.fetch` is read per call, not captured, so the tests' stub sees the request.
+ */
+// SAFETY: Bun's `typeof fetch` also declares `preconnect`, a warm-up hint the provider never calls;
+// the call signature is the whole contract here.
+export const openRouterFetch = ((input: RequestInfo | URL, init?: RequestInit) => {
+	const spanContext = trace.getActiveSpan()?.spanContext()
+	if (
+		spanContext === undefined ||
+		!isSpanContextValid(spanContext) ||
+		(spanContext.traceFlags & TraceFlags.SAMPLED) === 0 ||
+		typeof init?.body !== "string"
+	) {
+		return globalThis.fetch(input, init)
+	}
+	return globalThis.fetch(input, { ...init, body: withTraceIds(init.body, spanContext) })
+}) as typeof fetch
+
+const withTraceIds = (body: string, span: { traceId: string; spanId: string }): string => {
+	try {
+		// SAFETY: the provider serialised a JSON object; anything else is passed through untouched.
+		const parsed = JSON.parse(body) as unknown
+		if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) return body
+		const request = parsed as Record<string, unknown>
+		const existing =
+			typeof request.trace === "object" && request.trace !== null && !Array.isArray(request.trace)
+				? request.trace
+				: undefined
+		return JSON.stringify({
+			...request,
+			trace: { ...existing, trace_id: span.traceId, parent_span_id: span.spanId },
+		})
+	} catch {
+		return body
+	}
+}

apps/slack-agent/agent/lib/test-span.ts

@@ -0,0 +1,27 @@
+/**
+ * Test-only: run a function under an active, sampled OTel span — the way the Node SDK's spans are
+ * active in production. Not imported by any production module; lives beside `fetch-stub.ts` for
+ * the same reason it does (a shared seam that no `*.test.ts` should have to import from another).
+ */
+import { context, trace } from "@opentelemetry/api"
+import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"
+import { BasicTracerProvider } from "@opentelemetry/sdk-trace-base"
+
+// Without a context manager `startActiveSpan` never makes the span visible to `getActiveSpan()`.
+// A second registration in one process is refused; `withActiveSpan` checks the outcome instead.
+context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable())
+
+const tracer = new BasicTracerProvider().getTracer("test")
+
+export const withActiveSpan = <A>(
+	name: string,
+	fn: (ids: { traceId: string; spanId: string }) => Promise<A>,
+): Promise<A> =>
+	tracer.startActiveSpan(name, async (span) => {
+		try {
+			if (trace.getActiveSpan() !== span) throw new Error("no active-span context manager")
+			return await fn(span.spanContext())
+		} finally {
+			span.end()
+		}
+	})

apps/slack-agent/package.json

@@ -36,6 +36,7 @@
 		"zod": "4.4.3"
 	},
 	"devDependencies": {
+		"@opentelemetry/context-async-hooks": "^2.1.0",
 		"@types/bun": "^1.3.14",
 		"@types/node": "24.x",
 		"typescript": "7.0.2"

packages/effect-sdk/README.md

@@ -73,7 +73,7 @@ export default {
 
 `telemetry.layer` MUST live in the same runtime as your routes — provide it to the layer composition you hand to `HttpRouter.toWebHandler`, not a separate per-request runtime, or your spans won't pick up the Tracer reference.
 
-When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, the failed batch is restored ahead of newer telemetry and that signal sleeps 60s before retrying. Trace and log cooldowns are independent.
+When `MAPLE_INGEST_KEY` is unset, the SDK runs in no-op mode: buffers are drained so they don't grow across the isolate's lifetime, but no requests are made. After a flush failure, the failed batch is put back ahead of newer telemetry and that signal sleeps 60s before retrying; buffers hold 10k items per signal and evict the oldest past that, so a long outage loses the earliest telemetry first. Trace and log cooldowns are independent. `flush(env, { force: true })` posts even inside that cooldown: use it for the last flush of a unit of work that nothing will flush after (a Durable Object's background turn, say), and run it under `ctx.waitUntil()` like any other flush so the isolate stays up until it completes.
 
 ### Cloudflare-specific options
 

packages/effect-sdk/src/cloudflare/index.test.ts

@@ -299,6 +299,14 @@ describe("MapleCloudflareSDK.make", () => {
 		await telemetry.flush(env) // second flush — within cooldown, should be a no-op
 		expect(calls.length).toBe(failedCount)
 		expect(consoleErrorSpy).toHaveBeenCalled()
+
+		// A forced flush is the last one a piece of background work makes: it posts through the
+		// cooldown (and, failing again here, re-arms it).
+		await telemetry.flush(env, { force: true })
+		expect(calls.length).toBe(failedCount + 1)
+		expect(calls.at(-1)?.url).toMatch(/\/v1\/traces$/)
+		await telemetry.flush(env)
+		expect(calls.length).toBe(failedCount + 1)
 	})
 
 	// Effect defers `span.end` and `withSpan` finalizers onto the scheduler's