fix(telemetry): fail closed outside full mode

packages/telemetry/session-telemetry-otel/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/telemetry/session-telemetry-otel/README.md
-README.md: 01d803236329afbe65e2d92960928441aaff301c
-README.zh.md: 8adf4a3c11b95dc302f8afd0dc79e99433e50f22
+README.md: af177dc86bc30a7b17e34e3c8c3592326b9026f2
+README.zh.md: 9a3ad628bb7d480a1c4cd8346669ad8ebbd6258f

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

@@ -26,6 +26,8 @@ The OpenTelemetry backend for [the telemetry seam](../session-telemetry/) — th
 
 Programmatic TypeScript configuration uses the exported `TelemetryMode` enum (`TelemetryMode.FULL`, `TelemetryMode.FEEDBACK_ONLY`, or `TelemetryMode.DISABLED`); raw string literals are not assignable. Serialized Cordis configuration continues to use the string values shown above.
 
+Upload authorization is positive and fail-closed. An unknown direct-construction mode fails before transport configuration is read. Only `FULL` accepts direct `ctx.telemetry.emit()` calls. `FEEDBACK_ONLY` gives its on-demand coordinator a private backend capability and treats only the exact `feedback/record` object already stored at `session.events[event.seq]` as consent; an independently emitted bus value is ignored. `DISABLED` never constructs the SDK pipeline, even when exporter options are present.
+
 `exporter.url` is required in `FULL` and `FEEDBACK_ONLY`, has no default, and must parse as `http(s)`; it is optional and unused in `DISABLED`. Uploading modes also reject a non-positive-integer `processor.maxExportBatchSize`, which the SDK accepts but then hangs on at shutdown. 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, export cadence (`scheduledDelayMillis`), retry, queue bounds, and loss policy under sustained failure are the SDK's documented behavior, tuned through the `processor` passthrough. The backend deliberately implements no `flush()`: the batch processor is the only flusher in the process, which is what makes `shutdown()`'s drain complete.
 
 ## What leaves the machine

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

@@ -26,6 +26,8 @@
 
 程序化 TypeScript 配置使用导出的 `TelemetryMode` 枚举（`TelemetryMode.FULL`、`TelemetryMode.FEEDBACK_ONLY` 或 `TelemetryMode.DISABLED`）；原始字符串字面量不可赋值。序列化后的 Cordis 配置继续使用上表所示的字符串值。
 
+上传授权采用显式许可，且为 fail-closed。通过直接构造传入未知模式时，会在读取传输配置前失败。只有 `FULL` 接受对 `ctx.telemetry.emit()` 的直接调用。`FEEDBACK_ONLY` 向其按需协调器提供私有后端能力，并且仅在 `feedback/record` 对象已经存储于 `session.events[event.seq]` 且对象身份完全相同时，才将其视为同意；独立发出的总线值会被忽略。即使存在导出器选项，`DISABLED` 也绝不会构造 SDK 流水线。
+
 `exporter.url` 在 `FULL` 与 `FEEDBACK_ONLY` 中必填，无默认值，且必须能解析为 `http(s)`；在 `DISABLED` 中可省略且不使用。上传模式也会拒绝不是正整数的 `processor.maxExportBatchSize`，SDK 虽会接受该值，但随后会在关闭时挂起。其余全部是 SDK 自己的选项形态，由 SDK 拥有并在 SDK 文档中说明，两个配置块都整体透传（passthrough）：`OTLPExporterNodeConfigBase` 的每个字段（`headers`、`timeoutMillis`、`compression`、`keepAlive` 等）都会到达导出器；批处理、导出节奏（`scheduledDelayMillis`）、重试、队列上限，以及持续失败下的丢失策略，都是 SDK 的文档化行为，经 `processor` 透传调优。该后端刻意不实现 `flush()`：批处理器是进程内唯一执行 flush 的组件，`shutdown()` 的排空正因如此才是完整的。
 
 ## 哪些数据会离开本机

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

@@ -20,7 +20,7 @@ import type {} from '@deepseek-ai/dsh-command-feedback'
 import {
   Telemetry,
   TelemetryCoordinator,
-  type TelemetryCapture,
+  type TelemetryBackend,
   type TelemetryRecord,
   type TelemetrySeverity,
 } from '@deepseek-ai/dsh-session-telemetry'
@@ -54,6 +54,26 @@ export const TELEMETRY_MODES = [
 ] as const
 
 const DISABLED_FEEDBACK_WARNING = 'session telemetry is DISABLED; nothing will be shared and this feedback remains local'
+const NON_CANONICAL_FEEDBACK_WARNING = 'session telemetry ignored a feedback event absent from the canonical session log'
+const DROP_RECORD: TelemetryBackend['emit'] = () => {}
+
+/** Resolve the default and reject unknown runtime values before transport setup. */
+function resolveMode(mode: TelemetryMode | undefined): TelemetryMode {
+  const resolved = mode ?? TelemetryMode.FULL
+  switch (resolved) {
+    case TelemetryMode.FULL:
+    case TelemetryMode.FEEDBACK_ONLY:
+    case TelemetryMode.DISABLED:
+      return resolved
+    default:
+      return assertNever(resolved)
+  }
+}
+
+/** Fail closed when direct construction bypasses the runtime config schema. */
+function assertNever(value: never): never {
+  throw new Error(`session-telemetry-otel: unsupported mode ${JSON.stringify(value)}`)
+}
 
 /**
  * Plugin configuration: one sharing policy plus two verbatim SDK option
@@ -111,17 +131,15 @@ export class TelemetryOtel extends Telemetry {
   static inject = ['sessions']
   static Config = Config
 
+  private readonly directEmit: TelemetryBackend['emit']
   private readonly provider: LoggerProvider | undefined
-  private readonly ledger: Logger | undefined
-  private readonly ops: Logger | undefined
 
   constructor(ctx: Context, config: Config) {
+    const mode = resolveMode(config.mode)
     super(ctx)
-    const mode = config.mode ?? TelemetryMode.FULL
     if (mode === TelemetryMode.DISABLED) {
+      this.directEmit = DROP_RECORD
       this.provider = undefined
-      this.ledger = undefined
-      this.ops = undefined
       ctx.on('session/event', (_session, event) => {
         if (event.type === 'feedback/record') ctx.logger.warn(DISABLED_FEEDBACK_WARNING)
       })
@@ -168,37 +186,50 @@ export class TelemetryOtel extends Telemetry {
         }),
       ],
     })
-    this.ledger = this.provider.getLogger('@deepseek-ai/dsh-session-telemetry-otel', version)
-    this.ops = this.provider.getLogger('@deepseek-ai/dsh-session-telemetry-otel/ops', version)
-    const capture: TelemetryCapture = mode === TelemetryMode.FULL ? 'live' : 'on-demand'
-    const coordinator = new TelemetryCoordinator(ctx, this, capture)
-    if (mode === TelemetryMode.FEEDBACK_ONLY) {
-      // Session.append commits before publishing `session/event`, so the
-      // canonical log already includes this feedback record when replay begins.
-      ctx.on('session/event', (session, event) => {
-        if (event.type === 'feedback/record') coordinator.captureSession(session, event.seq)
+    const ledger = this.provider.getLogger('@deepseek-ai/dsh-session-telemetry-otel', version)
+    const ops = this.provider.getLogger('@deepseek-ai/dsh-session-telemetry-otel/ops', version)
+    const enqueue: TelemetryBackend['emit'] = (record) => {
+      const logger: Logger = record.channel === 'ops' ? ops : ledger
+      logger.emit({
+        timestamp: record.time,
+        observedTimestamp: record.time,
+        ...SEVERITY[record.severity],
+        // JSON-serializable by the seam's contract (validated at Session.append),
+        // which is exactly the AnyValue subset.
+        body: record.body as AnyValue,
+        attributes: record.attributes,
       })
     }
+    const backend: TelemetryBackend = {
+      emit: enqueue,
+      shutdown: () => this.shutdown(),
+    }
+    if (mode === TelemetryMode.FULL) {
+      this.directEmit = enqueue
+      new TelemetryCoordinator(ctx, backend, 'live')
+      return
+    }
+    this.directEmit = DROP_RECORD
+    const coordinator = new TelemetryCoordinator(ctx, backend, 'on-demand')
+    ctx.on('session/event', (session, event) => {
+      if (event.type !== 'feedback/record') return
+      // Consent is the committed record, not an independently emitted bus value.
+      if (session.events[event.seq] !== event) {
+        ctx.logger.warn(NON_CANONICAL_FEEDBACK_WARNING)
+        return
+      }
+      coordinator.captureSession(session, event.seq)
+    })
   }
 
   /**
-   * Map one seam record onto the SDK logger for its channel — a synchronous
-   * enqueue into the batch processor's queue. Direct calls are no-ops in
-   * `DISABLED`, where no coordinator or SDK pipeline exists.
-   * @param record - the logical record handed over by the coordinator.
+   * Hand a direct service record to the SDK only in `FULL`. Direct calls are
+   * no-ops in `FEEDBACK_ONLY` and `DISABLED`; feedback replay uses a private
+   * backend capability created only for the canonical feedback listener.
+   * @param record - the logical record offered directly to the service.
    */
   emit(record: TelemetryRecord): void {
-    const logger = record.channel === 'ops' ? this.ops : this.ledger
-    if (logger === undefined) return
-    logger.emit({
-      timestamp: record.time,
-      observedTimestamp: record.time,
-      ...SEVERITY[record.severity],
-      // JSON-serializable by the seam's contract (validated at Session.append),
-      // which is exactly the AnyValue subset.
-      body: record.body as AnyValue,
-      attributes: record.attributes,
-    })
+    this.directEmit(record)
   }
 
   // The seam's optional flush() hint is deliberately NOT implemented. The

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

@@ -105,6 +105,13 @@ describe('TelemetryOtel wire', () => {
     const session = ctx.sessions.create(SessionId('wire'), { meta: { cwd: '/tmp/w' } })
     session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
     session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, message: 'boom' } })
+    ctx.telemetry.emit({
+      channel: 'ledger',
+      time: Date.now(),
+      severity: 'info',
+      attributes: { 'session.id': 'wire', 'event.type': 'manual', 'event.seq': 99 },
+      body: { direct: true },
+    })
     await fiber.dispose()
 
     expect(captures.length).toBeGreaterThan(0)
@@ -128,6 +135,7 @@ describe('TelemetryOtel wire', () => {
     const end = ledger.find(r => r.record.attributes?.some(a => a.key === 'event.type' && a.value.stringValue === 'turn/end'))
     expect(end?.record.severityNumber).toBe(17)
     expect(end?.record.severityText).toBe('ERROR')
+    expect(eventTypes(captures)).toContain('manual')
 
     expect(ops).toHaveLength(1)
     expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } })
@@ -214,6 +222,16 @@ describe('TelemetryOtel wire', () => {
       mode: TelemetryMode.FEEDBACK_ONLY,
       exporter: { url },
     })
+    ctx.on('telemetry/record', (_record, next) => {
+      ctx.telemetry.emit({
+        channel: 'ledger',
+        time: Date.now(),
+        severity: 'info',
+        attributes: { 'session.id': 'feedback-only', 'event.type': 'direct-bypass', 'event.seq': 99 },
+        body: { mustStayLocal: true },
+      })
+      return next()
+    })
     const session = ctx.sessions.create(SessionId('feedback-only'), { meta: {} })
     session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
     recordFeedback(session, 'first report')
@@ -231,25 +249,48 @@ describe('TelemetryOtel wire', () => {
     expect(allRecords(captures).some(({ scope }) => scope.endsWith('/ops'))).toBe(false)
   })
 
-  it('sends no request when feedback-only mode ends without feedback', async () => {
+  it('ignores direct emits and non-canonical feedback in feedback-only mode', async () => {
     const { url, captures } = await mockCollector()
     const ctx = new Context()
     await ctx.plugin(SessionStore)
+    const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
     const fiber = await ctx.plugin(TelemetryOtel, {
       mode: TelemetryMode.FEEDBACK_ONLY,
       exporter: { url },
     })
     const session = ctx.sessions.create(SessionId('no-feedback'), { meta: {} })
     session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    ctx.telemetry.emit({
+      channel: 'ledger',
+      time: Date.now(),
+      severity: 'info',
+      attributes: { 'session.id': 'no-feedback', 'event.type': 'direct', 'event.seq': 99 },
+      body: { mustStayLocal: true },
+    })
+    ctx.emit('session/event', session, {
+      type: 'feedback/record',
+      seq: session.events.length,
+      time: Date.now(),
+      data: { text: 'not committed' },
+    })
     await fiber.dispose()
+
+    expect(warn).toHaveBeenCalledWith(
+      'session telemetry ignored a feedback event absent from the canonical session log',
+    )
     expect(captures).toEqual([])
   })
 
-  it('boots disabled without exporter config and warns when feedback stays local', async () => {
+  it('constructs no disabled transport even when exporter options are present', async () => {
+    const { url, captures } = await mockCollector()
     const ctx = new Context()
     await ctx.plugin(SessionStore)
     const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
-    const fiber = await ctx.plugin(TelemetryOtel, { mode: TelemetryMode.DISABLED })
+    const fiber = await ctx.plugin(TelemetryOtel, {
+      mode: TelemetryMode.DISABLED,
+      exporter: { url },
+      processor: { maxExportBatchSize: 0 },
+    })
     const session = ctx.sessions.create(SessionId('disabled'), { meta: {} })
     session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
     recordFeedback(session, 'local report')
@@ -268,6 +309,7 @@ describe('TelemetryOtel wire', () => {
     await fiber.dispose()
     recordFeedback(session, 'after disposal')
     expect(warn).toHaveBeenCalledTimes(1)
+    expect(captures).toEqual([])
   })
 
   it('defaults direct construction to full delivery', async () => {
@@ -306,6 +348,22 @@ describe('TelemetryOtel config fails loud', () => {
     await ctx.plugin(SessionStore)
     await expect(ctx.plugin(TelemetryOtel, config as Config)).rejects.toThrow(message)
   })
+
+  it('rejects an unknown direct mode before reading transport config', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    let exporterRead = false
+    const config = {
+      mode: 'INVALID',
+      get exporter() {
+        exporterRead = true
+        throw new Error('transport config was read')
+      },
+    } as unknown as Config
+
+    expect(() => new TelemetryOtel(ctx, config)).toThrow(/unsupported mode "INVALID"/)
+    expect(exporterRead).toBe(false)
+  })
 })
 
 describe('dsh-session-telemetry-otel real-load-path guard', () => {