fix(telemetry): join overlapping flush hints; contain adoption replay per event

Second review round, both pinned red-first:

- Overlapping turn-boundary flush hints now JOIN the outstanding flush
  promise (Promise.all) instead of displacing it: the SDK's
  concurrent-flush guard resolves an overlapping forceFlush()
  immediately, so retaining only the latest promise let shutdown()
  proceed while the first export was still in flight — the same silent
  drop the single-flush fix closed.
- Adoption replay contains failures per event, matching the firehose:
  one rejected record is withheld fail-closed while the rest of the
  historical log still hands off. Wrapping the whole loop let a single
  failure silently skip the remainder on an already-adopted session.

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

@@ -147,7 +147,7 @@ export class TelemetryOtel extends Telemetry {
     })
   }
 
-  /** The latest turn-boundary flush, retained so {@link shutdown} can order behind it. */
+  /** Every not-yet-settled turn-boundary flush, retained so {@link shutdown} can order behind ALL of them. */
   private inflightFlush: Promise<void> = Promise.resolve()
 
   /** Forward the turn-boundary hint to the SDK's flush, fire-and-forget. */
@@ -157,15 +157,17 @@ export class TelemetryOtel extends Telemetry {
     // 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.
+    // an overlapping hint resolves instantly and must JOIN the outstanding
+    // one, not displace it — shutdown orders behind the whole set.
     /* v8 ignore next -- unreachable guard: forceFlush does not reject while the provider is alive */
-    this.inflightFlush = this.provider.forceFlush().catch(() => {})
+    const flush = this.provider.forceFlush().catch(() => {})
+    this.inflightFlush = Promise.all([this.inflightFlush, flush]).then(() => undefined)
   }
 
   /**
    * Delegate disposal to the SDK's shutdown contract: flush the queue and
-   * 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
+   * quiesce. Orders behind every outstanding 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

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

@@ -155,6 +155,39 @@ describe('TelemetryOtel wire', () => {
     expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } })
   })
 
+  it('orders shutdown behind the OLDEST in-flight flush when hints overlap', async () => {
+    // The SDK's concurrent-flush guard resolves an overlapping forceFlush()
+    // immediately; if the backend RETAINS only the latest flush promise, two
+    // back-to-back turn flushes leave shutdown awaiting the instantly-resolved
+    // second one while the first still exports — reopening the same silent
+    // drop the single-flush race test pins.
+    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('race2'), { meta: {} })
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    ctx.telemetry.flush!()
+    await arrived.promise
+    // Second hint while the first export is held open: resolves immediately
+    // under the SDK's guard and must not displace the outstanding one.
+    ctx.telemetry.flush!()
+
+    const disposal = fiber.dispose()
+    await new Promise(resolve => setTimeout(resolve, 50))
+    gate.resolve(true)
+    await disposal
+
+    const ops = allRecords(captures).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()

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

@@ -113,15 +113,19 @@ export class TelemetryCoordinator {
    * @param session - the live session to adopt; a second adoption is a no-op.
    */
   private adopt(session: Session): void {
-    this.contain(() => {
-      if (this.adopted.has(session)) return
-      this.adopted.add(session)
-      const cursor = handoffCursor.get(session) ?? -1
-      for (const event of session.events) {
+    if (this.adopted.has(session)) return
+    this.adopted.add(session)
+    const cursor = handoffCursor.get(session) ?? -1
+    // Containment is PER EVENT, matching the firehose: one rejected record
+    // is withheld fail-closed while the rest of the historical replay
+    // proceeds — wrapping the whole loop would let a single failure silently
+    // skip the remainder of the log on an already-adopted session.
+    for (const event of session.events) {
+      this.contain(() => {
         if (event.seq <= cursor) this.track(session, event)
         else this.capture(session, event)
-      }
-    })
+      })
+    }
   }
 
   /** Feed the chunk projection without handing off — the ≤cursor half of re-adoption. */

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

@@ -26,11 +26,15 @@ class FakeBackend implements TelemetryBackend {
   records: TelemetryRecord[] = []
   calls: string[] = []
   emitError: Error | undefined
+  rejectSeq: number | undefined
   shutdownError: Error | undefined
   shutdownResolved = false
 
   emit(record: TelemetryRecord): void {
     if (this.emitError) throw this.emitError
+    if (this.rejectSeq !== undefined && record.attributes['event.seq'] === this.rejectSeq) {
+      throw new Error(`backend rejected seq ${this.rejectSeq}`)
+    }
     this.records.push(record)
     this.calls.push(`emit:${String(record.attributes['event.seq'] ?? record.attributes['telemetry.op'])}`)
   }
@@ -213,6 +217,28 @@ describe('TelemetryCoordinator adoption', () => {
     expect(second.ledger().map(r => r.attributes['event.type'])).toEqual(['turn/end'])
   })
 
+  it('replays past a record the backend rejects: one event withheld, the rest adopted', async () => {
+    const backend = new FakeBackend()
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
+    const session = liveSession(ctx, 'partial')
+    appendTurn(session)
+    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    // The backend rejects exactly the middle historical event: fail-closed
+    // must withhold THAT record only — an adoption replay that dies on the
+    // first contained failure would silently skip the rest of the log while
+    // the session stays marked adopted.
+    backend.rejectSeq = 1
+    await ctx.plugin({
+      name: 'fake-telemetry',
+      inject: ['sessions'],
+      apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
+    })
+    expect(backend.ledger().map(r => r.attributes['event.seq'])).toEqual([0, 2])
+    expect(warn).toHaveBeenCalled()
+  })
+
   it('re-hands the full log when no cursor survived (fresh session object)', async () => {
     const backend = new FakeBackend()
     const ctx = new Context()