fix(telemetry): review fixes — flush/shutdown ordering, session retirement, whole-exporter passthrough

Three review findings, each pinned by a red test first:

- The OTel backend retains the latest turn-boundary flush promise and
  shutdown() awaits it before provider.shutdown(): the SDK's
  concurrent-flush guard makes the shutdown-internal flush return early
  while one is in flight, silently dropping everything enqueued after
  the flush snapshot (including the coordinator's dispose-time shutdown
  markers).
- The coordinator retires sessions from the adopted set on
  session/disposed (mirroring the persistence coordinator): a long-lived
  backend no longer retains closed sessions and their event logs, and
  final unload no longer stamps shutdown markers for dead sessions.
- The exporter config passes through whole to OTLPLogExporter and its
  type/JSDoc now advertise the full OTLPExporterNodeConfigBase shape:
  rebuilding url/headers only silently dropped documented SDK options
  (timeoutMillis, compression, keepAlive, ...).

packages/telemetry/session-telemetry-otel/README.md

@@ -15,7 +15,7 @@ The OpenTelemetry backend for [the telemetry seam](../session-telemetry/) — th
     processor: {}            # optional; passed verbatim to BatchLogRecordProcessor
 ```
 
-`exporter.url` is the one field this package validates itself — required, no default, must parse as `http(s)` — so a missing endpoint fails at plugin load. Everything else is the SDK's option shape, owned and documented by the SDK; batching, retry, queue bounds, and loss policy under sustained failure are its documented behavior, tuned through the `processor` passthrough. Removing this block from `cordis.yml` is the opt-out: no residual state, no `enabled` flag.
+`exporter.url` is the one field this package validates itself — required, no default, must parse as `http(s)` — so a missing endpoint fails at plugin load. Everything else is the SDK's option shape, owned and documented by the SDK, and both blocks pass through whole: every `OTLPExporterNodeConfigBase` field (`headers`, `timeoutMillis`, `compression`, `keepAlive`, …) reaches the exporter, and batching, retry, queue bounds, and loss policy under sustained failure are the SDK's documented behavior, tuned through the `processor` passthrough. Removing this block from `cordis.yml` is the opt-out: no residual state, no `enabled` flag.
 
 ## What leaves the machine
 

packages/telemetry/session-telemetry-otel/package.json

@@ -30,6 +30,7 @@
     "@opentelemetry/api": "^1.9.1",
     "@opentelemetry/api-logs": "^0.220.0",
     "@opentelemetry/exporter-logs-otlp-http": "^0.220.0",
+    "@opentelemetry/otlp-exporter-base": "^0.220.0",
     "@opentelemetry/resources": "^2.9.0",
     "@opentelemetry/sdk-logs": "^0.220.0",
     "schemastery": "^3.18.0"

packages/telemetry/session-telemetry-otel/src/index.ts

@@ -23,6 +23,7 @@ import {
   type BatchLogRecordProcessorOptions,
 } from '@opentelemetry/sdk-logs'
 import { OTLPLogExporter } from '@opentelemetry/exporter-logs-otlp-http'
+import type { OTLPExporterNodeConfigBase } from '@opentelemetry/otlp-exporter-base'
 import { SeverityNumber, type AnyValue, type Logger } from '@opentelemetry/api-logs'
 import { resourceFromAttributes } from '@opentelemetry/resources'
 
@@ -37,12 +38,15 @@ const { version } = createRequire(import.meta.url)('../package.json') as { versi
  * must fail at plugin load, not at first export.
  */
 export interface Config {
-  /** Passed verbatim to the SDK's OTLP/HTTP log exporter. */
-  exporter?: {
+  /**
+   * Passed verbatim to the SDK's OTLP/HTTP log exporter — the complete
+   * `OTLPExporterNodeConfigBase` shape (`headers`, `timeoutMillis`,
+   * `compression`, `keepAlive`, …), owned and documented by the SDK. `url`
+   * is the one field this package requires and validates itself.
+   */
+  exporter?: OTLPExporterNodeConfigBase & {
     /** Full logs endpoint (e.g. `https://collector.example.com/v1/logs`). Required; validated at plugin load. */
     url?: string
-    /** Extra request headers (auth etc.); owned and sent by the SDK exporter. */
-    headers?: Record<string, string>
   }
   /**
    * Passed verbatim to `BatchLogRecordProcessor` (minus the exporter slot,
@@ -54,15 +58,13 @@ export interface Config {
 /**
  * Schemastery validator for {@link Config}; cordis runs it before the plugin
  * starts. Shape-level only — the load-bearing `exporter.url` check lives in
- * the constructor so its error message names the field.
+ * the constructor so its error message names the field. Both slots are opaque
+ * passthroughs: the SDK owns their shapes and validates its own options;
+ * re-declaring them field-by-field here would violate the boundary axiom
+ * (and silently drop every field not re-declared).
  */
 export const Config: z<Config> = z.object({
-  exporter: z.object({
-    url: z.string(),
-    headers: z.dict(z.string()),
-  }),
-  // Opaque passthrough: the SDK owns this shape and validates its own
-  // options; re-declaring them here would violate the boundary axiom.
+  exporter: z.any(),
   processor: z.any(),
 })
 
@@ -112,14 +114,13 @@ export class TelemetryOtel extends Telemetry {
       processors: [
         new BatchLogRecordProcessor({
           ...config.processor,
-          exporter: new OTLPLogExporter({
-            url,
-            // App identity travels in the Resource (service.name/version);
-            // the transport-level user-agent is the SDK's own, per the axiom.
-            // Schemastery fills `headers` with {} before cordis constructs the
-            // plugin, so the optional type exists for hand-authors only.
-            headers: config.exporter?.headers as Record<string, string>,
-          }),
+          // The complete validated exporter object, verbatim: every SDK
+          // option (`timeoutMillis`, `compression`, `keepAlive`, …) reaches
+          // the exporter — rebuilding selected fields here would silently
+          // ignore the rest. App identity travels in the Resource
+          // (service.name/version); the transport-level user-agent is the
+          // SDK's own, per the axiom.
+          exporter: new OTLPLogExporter(config.exporter),
         }),
       ],
     })
@@ -146,22 +147,34 @@ export class TelemetryOtel extends Telemetry {
     })
   }
 
+  /** The latest turn-boundary flush, retained so {@link shutdown} can order behind it. */
+  private inflightFlush: Promise<void> = Promise.resolve()
+
   /** Forward the turn-boundary hint to the SDK's flush, fire-and-forget. */
   override flush(): void {
     // Best-effort hint: the SDK resolves forceFlush even when exports fail
     // (failures go to its own diagnostics), and the coordinator stops calling
-    // this once the fiber is disposed — a rejection would be SDK drift.
+    // this once the fiber is disposed — a rejection would be SDK drift. The
+    // settled promise is retained (not awaited): the SDK's concurrent-flush
+    // guard makes a flush that overlaps another return WITHOUT draining, so
+    // shutdown must wait this one out before trusting its own flush.
     /* v8 ignore next -- unreachable guard: forceFlush does not reject while the provider is alive */
-    void this.provider.forceFlush().catch(() => {})
+    this.inflightFlush = this.provider.forceFlush().catch(() => {})
   }
 
   /**
    * Delegate disposal to the SDK's shutdown contract: flush the queue and
-   * quiesce. Awaited (and error-contained) by the coordinator's disposer.
+   * quiesce. Orders behind the last turn-boundary flush first — shutdown's
+   * internal flush is a no-op while one is in flight (the SDK's
+   * concurrent-flush guard), which would silently drop everything enqueued
+   * after that flush snapshot, including the coordinator's dispose-time
+   * `shutdown` markers. Awaited (and error-contained) by the coordinator's
+   * disposer.
    * @returns resolves when the SDK pipeline has quiesced.
    */
-  shutdown(): Promise<void> {
-    return this.provider.shutdown()
+  async shutdown(): Promise<void> {
+    await this.inflightFlush
+    await this.provider.shutdown()
   }
 }
 

packages/telemetry/session-telemetry-otel/tests/otel.spec.ts

@@ -8,6 +8,7 @@
 import { afterEach, describe, expect, it } from 'vitest'
 import { createServer, type Server } from 'node:http'
 import { once } from 'node:events'
+import { gunzipSync } from 'node:zlib'
 import { Context } from 'cordis'
 import Loader from '@cordisjs/plugin-loader'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
@@ -43,17 +44,26 @@ afterEach(async () => {
   }
 })
 
-async function mockCollector(): Promise<{ url: string; captures: Capture[] }> {
+async function mockCollector(
+  beforeRespond?: (requestIndex: number) => Promise<void> | void,
+): Promise<{ url: string; captures: Capture[] }> {
   const captures: Capture[] = []
+  let requestIndex = 0
   const server = createServer((request, response) => {
     const chunks: Buffer[] = []
     request.on('data', chunk => chunks.push(chunk as Buffer))
     request.on('end', () => {
-      captures.push({
-        headers: request.headers,
-        body: JSON.parse(Buffer.concat(chunks).toString()) as OtlpLogsRequest,
-      })
-      response.writeHead(200, { 'content-type': 'application/json' }).end('{}')
+      const index = requestIndex++
+      void (async () => {
+        await beforeRespond?.(index)
+        const raw = Buffer.concat(chunks)
+        const body = request.headers['content-encoding'] === 'gzip' ? gunzipSync(raw) : raw
+        captures.push({
+          headers: request.headers,
+          body: JSON.parse(body.toString()) as OtlpLogsRequest,
+        })
+        response.writeHead(200, { 'content-type': 'application/json' }).end('{}')
+      })()
     })
   })
   servers.push(server)
@@ -113,6 +123,59 @@ describe('TelemetryOtel wire', () => {
     expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } })
   })
 
+  it('delivers records enqueued while a turn-boundary flush is in flight (flush/shutdown race)', async () => {
+    // Hold the collector's response to the flush-triggered export open until
+    // after disposal has begun: the SDK's concurrent-flush guard makes the
+    // shutdown-internal flush return early while another flush is running, so
+    // without ordering in the backend the coordinator's dispose-time shutdown
+    // marker (enqueued after the flush snapshot) would be dropped silently.
+    const gate = Promise.withResolvers<boolean>()
+    const arrived = Promise.withResolvers<boolean>()
+    const { url, captures } = await mockCollector(async (index) => {
+      if (index === 0) {
+        arrived.resolve(true)
+        await gate.promise
+      }
+    })
+    const { ctx, fiber } = await boot(url)
+    const session = ctx.sessions.create(SessionId('race'), { meta: {} })
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    ctx.telemetry.flush!()
+    await arrived.promise
+
+    const disposal = fiber.dispose()
+    // Let disposal reach the backend's shutdown while the export is held open.
+    await new Promise(resolve => setTimeout(resolve, 50))
+    gate.resolve(true)
+    await disposal
+
+    const records = allRecords(captures)
+    const ops = records.filter(r => r.scope === '@deepseek-ai/dsh-session-telemetry-otel/ops')
+    expect(ops).toHaveLength(1)
+    expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } })
+  })
+
+  it('passes exporter options beyond url and headers through to the SDK exporter', async () => {
+    const { url, captures } = await mockCollector()
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    // `compression` is a documented SDK exporter option; the advertised
+    // verbatim passthrough must hand it (and every other field) to the
+    // exporter rather than silently rebuilding url/headers only.
+    const fiber = await ctx.plugin(TelemetryOtel, {
+      exporter: { url, compression: 'gzip' },
+    } as Config)
+    const session = ctx.sessions.create(SessionId('gzip'), { meta: {} })
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    await fiber.dispose()
+
+    expect(captures.length).toBeGreaterThan(0)
+    expect(captures[0]!.headers['content-encoding']).toBe('gzip')
+    const types = allRecords(captures).flatMap(({ record }) =>
+      record.attributes?.flatMap(a => a.key === 'event.type' ? [a.value.stringValue] : []) ?? [])
+    expect(types).toContain('turn/start')
+  })
+
   it('maps the warn severity and forwards the flush hint to the SDK', async () => {
     const { url, captures } = await mockCollector()
     const { ctx, fiber } = await boot(url)

packages/telemetry/session-telemetry/README.md

@@ -8,7 +8,7 @@ The telemetry seam: the CAPTURE side of session-event reporting, behind a backen
 
 ## Capture points
 
-The coordinator registers, all through the composing fiber's effects: `session/created` (adopt: record the header, read the log back through the projection — seed events from fork/resume never re-emit on the firehose), `session/event` (project, deep-copy, redact, hand off; zero I/O), `session/flush` (forward the optional `flush()` hint and return void — the loop's awaited parallel must never wait on telemetry), `agent/error` (the one live-bus relay; turn-enclosure structurally bars those errors from the log), a dispose effect (per adopted session emit its `shutdown` operational record, then await the backend's `shutdown()`; failures warn instead of throwing), and an adoption sweep of `ctx.sessions.list()` (a hot reload does not replay `session/created`).
+The coordinator registers, all through the composing fiber's effects: `session/created` (adopt: record the header, read the log back through the projection — seed events from fork/resume never re-emit on the firehose), `session/event` (project, deep-copy, redact, hand off; zero I/O), `session/flush` (forward the optional `flush()` hint and return void — the loop's awaited parallel must never wait on telemetry), `session/disposed` (retire: release the adopted entry so a long-lived backend neither retains closed sessions nor stamps dispose-time markers for them), `agent/error` (the one live-bus relay; turn-enclosure structurally bars those errors from the log), a dispose effect (per still-adopted session emit its `shutdown` operational record, then await the backend's `shutdown()`; failures warn instead of throwing), and an adoption sweep of `ctx.sessions.list()` (a hot reload does not replay `session/created`).
 
 ## The redact waterfall
 

packages/telemetry/session-telemetry/src/coordinator.ts

@@ -35,13 +35,20 @@ const handoffCursor = new WeakMap<Session, number>()
  * Registers the persistence-coordinator listener set plus the `agent/error`
  * relay, all through `ctx.effect()`/`ctx.on()` on the composing fiber, and
  * sweeps already-live sessions (a hot reload does not replay
- * `session/created`). Disposal emits each adopted session's `shutdown`
+ * `session/created`). A `session/disposed` retires the session from the
+ * adopted set — a long-lived backend must not retain closed sessions (and
+ * their frozen event logs) or stamp dispose-time markers for sessions that
+ * already ended. Disposal emits each still-adopted session's `shutdown`
  * operational record and then awaits the backend's `shutdown()`; a failure
  * there warns instead of throwing — best-effort reporting must not fail
  * application teardown.
  */
 export class TelemetryCoordinator {
-  /** Sessions adopted by THIS fiber, for dispose-time `shutdown` records and double-adoption protection. */
+  /**
+   * Sessions adopted by THIS fiber and still live, for dispose-time
+   * `shutdown` records and double-adoption protection; `session/disposed`
+   * retires entries.
+   */
   private readonly adopted = new Set<Session>()
   /** Per session, the `turn:step` keys whose first chunk already shipped; rebuilt from the log on re-adoption. */
   private readonly chunkSeen = new WeakMap<Session, Set<string>>()
@@ -57,6 +64,11 @@ export class TelemetryCoordinator {
     ctx.on('session/created', (session) => {
       this.adopt(session)
     })
+    // Retirement is observe-only: the projection/cursor WeakMaps die with the
+    // Session object; only the strong adopted set needs the explicit release.
+    ctx.on('session/disposed', (session) => {
+      this.adopted.delete(session)
+    })
     ctx.on('session/event', (session, event) => {
       this.contain(() => {
         this.capture(session, event)

packages/telemetry/session-telemetry/src/index.ts

@@ -108,9 +108,13 @@ export interface TelemetryBackend {
   flush?(): void
   /**
    * Forward the fiber's disposal to the SDK: flush whatever is queued and
-   * reach quiescence, per the SDK's own shutdown contract. Awaited by the
-   * coordinator's dispose; a rejection is logged as a warning and never
-   * fails application teardown.
+   * reach quiescence, per the SDK's own shutdown contract. Everything
+   * emitted before this call must still be delivered — including records
+   * enqueued while a {@link flush} hint is in flight, so a backend whose SDK
+   * guards against concurrent flushes orders behind the outstanding one (the
+   * coordinator emits its dispose-time `shutdown` markers immediately before
+   * calling this). Awaited by the coordinator's dispose; a rejection is
+   * logged as a warning and never fails application teardown.
    * @returns resolves when the backend's pipeline has quiesced.
    */
   shutdown(): Promise<void>

packages/telemetry/session-telemetry/tests/telemetry.spec.ts

@@ -262,6 +262,23 @@ describe('TelemetryCoordinator lifecycle and containment', () => {
     expect(ops.every(r => !('event.seq' in r.attributes) && !('event.type' in r.attributes))).toBe(true)
   })
 
+  it('retires a disposed session: no retention, no stale shutdown marker at unload', async () => {
+    const { ctx, backend, fiber } = await setup()
+    liveSession(ctx, 'survivor')
+    // A session owned by its own fiber: disposing the fiber detaches it from
+    // the store and emits `session/disposed` — the authoritative retirement
+    // signal a long-lived telemetry backend must honor, or every closed
+    // session (and its full event log) stays strongly held for the backend's
+    // lifetime and final unload emits shutdown markers for dead sessions.
+    const owner = await ctx.plugin(Object.assign((inner: Context) => {
+      inner.sessions.create(SessionId('ephemeral'), { meta: {} })
+    }, { inject: ['sessions'] }))
+    await owner.dispose()
+    await fiber.dispose()
+    const ops = backend.records.filter(r => r.channel === 'ops')
+    expect(ops.map(r => r.attributes['session.id'])).toEqual(['survivor'])
+  })
+
   it('warns instead of throwing when backend shutdown fails', async () => {
     const backend = new FakeBackend()
     backend.shutdownError = new Error('exporter unreachable')