fix(cli): bound telemetry shutdown on signals

Refs #1356

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: ad4a97868c28dc3873c839490aa506271459e249
-README.zh.md: f1ad73ddf66aacc30a9024a9290df2c44686efe9
+README.md: 3abd97187cafee132823c02a0b0d103a86bda7db
+README.zh.md: 223e6a663933da81032a1fbbb4211555c4bcc159

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

@@ -10,6 +10,7 @@ The OpenTelemetry backend for [the telemetry seam](../session-telemetry/) — th
 - id: telemetry-otel
   name: '@deepseek-ai/dsh-session-telemetry-otel'
   config:
+    shutdownTimeoutMillis: 3000 # optional; defaults to 3000
     exporter:                # passed verbatim to the SDK's OTLP/HTTP log exporter
       url: https://collector.example.com/v1/logs
       headers:
@@ -17,7 +18,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 (as does 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. Removing this block from `cordis.yml` is the opt-out: no residual state, no `enabled` flag.
+`exporter.url` is required, has no default, and must parse as `http(s)`; `shutdownTimeoutMillis` is a positive finite DSH-owned outer deadline and defaults to 3000 ms; a non-positive-integer `processor.maxExportBatchSize` also fails at plugin load because the SDK accepts it but then hangs on shutdown. Both SDK 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 SDK behavior tuned through `processor`. The backend implements no `flush()`: the batch processor owns ordinary flushing. During shutdown, however, OTel awaits `exporter.forceFlush()` before the processor's `exportTimeoutMillis`-bounded completion promise; if that transport promise never settles, this package abandons the wait at `shutdownTimeoutMillis`, logs the contained shutdown failure through the coordinator, and lets application teardown continue. The deadline cannot cancel the SDK transport, so records still pending then may be lost at process exit. 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/README.zh.md

@@ -10,6 +10,7 @@
 - id: telemetry-otel
   name: '@deepseek-ai/dsh-session-telemetry-otel'
   config:
+    shutdownTimeoutMillis: 3000 # optional; defaults to 3000
     exporter:                # passed verbatim to the SDK's OTLP/HTTP log exporter
       url: https://collector.example.com/v1/logs
       headers:
@@ -17,7 +18,7 @@
     processor: {}            # optional; passed verbatim to BatchLogRecordProcessor
 ```
 
-`exporter.url` 是本包（package）唯一自行校验的字段：必填、无默认值、必须能解析为 `http(s)`，因此缺失端点会在插件加载时失败（`processor.maxExportBatchSize` 不是正整数时同样如此：SDK 会接受该值，随后却在关闭时因它挂起）。其余全部是 SDK 自己的选项形态，由 SDK 拥有并在 SDK 文档中说明，两个配置块都整体透传（passthrough）：`OTLPExporterNodeConfigBase` 的每个字段（`headers`、`timeoutMillis`、`compression`、`keepAlive` 等）都会到达导出器；批处理、导出节奏（`scheduledDelayMillis`）、重试、队列上限，以及持续失败下的丢失策略，都是 SDK 的文档化行为，经 `processor` 透传调优。该后端刻意不实现 `flush()`：批处理器是进程内唯一执行 flush 的组件，`shutdown()` 的排空正因如此才是完整的。从 `cordis.yml` 中删除该配置块即为退出方式：无残留状态，也没有 `enabled` 开关。
+`exporter.url` 是必填项、没有默认值，并且必须能解析为 `http(s)`；`shutdownTimeoutMillis` 是由 DSH 管理的有限正数外层截止时间，默认值为 3000 ms；`processor.maxExportBatchSize` 不是正整数时也会在插件加载时失败，因为 SDK 会接受该值，随后却在关闭时挂起。两个 SDK 配置块都整体透传（passthrough）：`OTLPExporterNodeConfigBase` 的每个字段（`headers`、`timeoutMillis`、`compression`、`keepAlive` 等）都会到达导出器；批处理、导出节奏（`scheduledDelayMillis`）、重试、队列上限，以及持续失败下的丢失策略，都是通过 `processor` 调节的 SDK 行为。该后端不实现 `flush()`：常规 flush 由批处理器负责。但在关闭期间，OTel 会先等待 `exporter.forceFlush()`，再进入受 `exportTimeoutMillis` 限制的处理器完成 promise；如果该传输 promise 始终不结算，本包（package）会在 `shutdownTimeoutMillis` 到期时放弃等待，沿协调器现有的失败隔离路径记录关闭失败，并让应用继续拆卸。该截止时间无法取消 SDK 传输，因此届时仍待处理的记录可能在进程退出时丢失。从 `cordis.yml` 中删除该配置块即为退出方式：无残留状态，也没有 `enabled` 开关。
 
 ## 哪些数据会离开本机
 

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

@@ -6,8 +6,9 @@
  * record handed over by the seam onto `logger.emit()`. Per the seam's
  * boundary axiom, everything downstream of that call (batching, retry,
  * queueing, loss policy) is the SDK's documented behavior, configured
- * verbatim through the `exporter`/`processor` passthroughs; this package
- * adds no knobs of its own on top of them.
+ * verbatim through the `exporter`/`processor` passthroughs. The one
+ * backend-owned policy is an outer shutdown deadline: the SDK's export
+ * timeout does not bound its preceding `forceFlush()` wait.
  *
  * @module @deepseek-ai/dsh-session-telemetry-otel
  */
@@ -33,10 +34,9 @@ import { resourceFromAttributes } from '@opentelemetry/resources'
 const { version } = createRequire(import.meta.url)('../package.json') as { version: string }
 
 /**
- * Plugin configuration: two verbatim SDK option shapes plus nothing else.
- * `exporter.url` is the one field this package validates itself — required,
- * no default, must parse as an `http(s)` URL — because a missing endpoint
- * must fail at plugin load, not at first export.
+ * Plugin configuration: two verbatim SDK option shapes plus one DSH-owned
+ * shutdown bound. The package validates its endpoint and shutdown deadline
+ * because both must fail at plugin load rather than at first export or exit.
  */
 export interface Config {
   /**
@@ -54,21 +54,31 @@ export interface Config {
    * which this plugin fills); the SDK owns and documents these knobs.
    */
   processor?: Omit<BatchLogRecordProcessorOptions, 'exporter'>
+  /** Maximum time spent awaiting the SDK provider's complete shutdown path. */
+  shutdownTimeoutMillis?: number
 }
 
 /**
  * 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. Both slots are opaque
- * passthroughs: the SDK owns their shapes and validates its own options;
+ * starts. Shape-level only — load-bearing value checks live in the constructor
+ * so their errors name the fields. Both SDK 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.any(),
   processor: z.any(),
+  shutdownTimeoutMillis: z.number(),
 })
 
+/** Default outer allowance for the SDK's complete shutdown sequence. */
+export const DEFAULT_SHUTDOWN_TIMEOUT_MILLIS = 3_000
+
+// Node clamps larger timer delays to one millisecond. This is a runtime
+// protocol limit, not a deployment default.
+const MAX_TIMER_DELAY_MILLIS = 2_147_483_647
+
 /** Severity mapping from the seam's three-level vocabulary to OTel severity numbers. */
 const SEVERITY: Record<TelemetrySeverity, { severityNumber: SeverityNumber; severityText: string }> = {
   info: { severityNumber: SeverityNumber.INFO, severityText: 'INFO' },
@@ -90,6 +100,7 @@ export class TelemetryOtel extends Telemetry {
   private readonly provider: LoggerProvider
   private readonly ledger: Logger
   private readonly ops: Logger
+  private readonly shutdownTimeoutMillis: number
 
   constructor(ctx: Context, config: Config) {
     super(ctx)
@@ -115,6 +126,11 @@ export class TelemetryOtel extends Telemetry {
     if (batchSize !== undefined && (!Number.isInteger(batchSize) || batchSize < 1)) {
       throw new Error(`session-telemetry-otel: processor.maxExportBatchSize must be a positive integer, got ${String(batchSize)}`)
     }
+    const shutdownTimeoutMillis = config.shutdownTimeoutMillis ?? DEFAULT_SHUTDOWN_TIMEOUT_MILLIS
+    if (!Number.isFinite(shutdownTimeoutMillis) || shutdownTimeoutMillis <= 0 || shutdownTimeoutMillis > MAX_TIMER_DELAY_MILLIS) {
+      throw new Error(`session-telemetry-otel: shutdownTimeoutMillis must be a positive finite number no greater than ${MAX_TIMER_DELAY_MILLIS}, got ${String(shutdownTimeoutMillis)}`)
+    }
+    this.shutdownTimeoutMillis = shutdownTimeoutMillis
     this.provider = new LoggerProvider({
       resource: resourceFromAttributes({
         'service.name': APP_IDENTITY.product,
@@ -170,16 +186,26 @@ export class TelemetryOtel extends Telemetry {
   // the revival Agent Note.
 
   /**
-   * Delegate disposal to the SDK's shutdown contract: drain the queue and
-   * quiesce. With no concurrent `forceFlush()` in the process (see above),
-   * shutdown's internal drain is complete — everything emitted before this
-   * call, including the coordinator's dispose-time `shutdown` markers, is
-   * exported before the exporter closes. Awaited (and error-contained) by
-   * the coordinator's disposer.
-   * @returns resolves when the SDK pipeline has quiesced.
+   * Ask the SDK to drain and quiesce, but reject after the backend-owned
+   * deadline. OTel's processor export timeout wraps `exportCompleted` only;
+   * shutdown awaits `exporter.forceFlush()` first, which can remain pending
+   * when the transport never obtains a socket. The provider promise remains
+   * observed after the deadline so a later rejection cannot become unhandled.
+   * @returns resolves when the SDK pipeline quiesces, or rejects at the configured deadline.
    */
-  shutdown(): Promise<void> {
-    return this.provider.shutdown()
+  async shutdown(): Promise<void> {
+    const providerShutdown = this.provider.shutdown()
+    let timer: ReturnType<typeof setTimeout> | undefined
+    const deadline = new Promise<never>((_resolve, reject) => {
+      timer = setTimeout(() => {
+        reject(new Error(`session-telemetry-otel: provider shutdown exceeded ${this.shutdownTimeoutMillis}ms`))
+      }, this.shutdownTimeoutMillis)
+    })
+    try {
+      await Promise.race([providerShutdown, deadline])
+    } finally {
+      if (timer !== undefined) clearTimeout(timer)
+    }
   }
 }
 

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

@@ -180,6 +180,38 @@ describe('TelemetryOtel wire', () => {
     expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } })
   })
 
+  it('bounds the SDK forceFlush wait when an in-flight transport never settles', async () => {
+    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 = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(TelemetryOtel, {
+      exporter: { url, timeoutMillis: 60_000 },
+      processor: { scheduledDelayMillis: 10, exportTimeoutMillis: 60_000 },
+      shutdownTimeoutMillis: 50,
+    })
+    const session = ctx.sessions.create(SessionId('bounded-shutdown'), { meta: {} })
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    await arrived.promise
+
+    const started = performance.now()
+    await fiber.dispose()
+    expect(performance.now() - started).toBeLessThan(1_000)
+    expect(captures).toHaveLength(0)
+
+    // The outer deadline cannot cancel the SDK transport. Let it finish so
+    // the real provider promise remains clean after the test has proved the
+    // Cordis disposer no longer waits for it.
+    gate.resolve(true)
+    await expect.poll(() => captures.length).toBeGreaterThanOrEqual(2)
+  })
+
   it('passes exporter options beyond url and headers through to the SDK exporter', async () => {
     const { url, captures } = await mockCollector()
     const ctx = new Context()
@@ -227,6 +259,8 @@ describe('TelemetryOtel config fails loud', () => {
     // splices empty batches forever — dispose would hang, so reject at load.
     [{ exporter: { url: 'http://c/v1/logs' }, processor: { maxExportBatchSize: 0 } }, /maxExportBatchSize/],
     [{ exporter: { url: 'http://c/v1/logs' }, processor: { maxExportBatchSize: 0.5 } }, /maxExportBatchSize/],
+    [{ exporter: { url: 'http://c/v1/logs' }, shutdownTimeoutMillis: 0 }, /shutdownTimeoutMillis/],
+    [{ exporter: { url: 'http://c/v1/logs' }, shutdownTimeoutMillis: Number.POSITIVE_INFINITY }, /shutdownTimeoutMillis/],
   ])('rejects %j at plugin load', async (config, message) => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)