Merge remote-tracking branch 'origin/master' into fix/pty-handoff-grace
This commit is contained in:
Chinesezjc
@@ -1,6 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# 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
|
||||
README.md: 911f18547120eb3dbbc9e42bbcd41e3b6d518cfe
|
||||
README.zh.md: d6e1f0bf9b38b40944f8e3cebea3f6d90dcaceb5
|
||||
# pnpm run verify-translation-pairing --write packages/README.md
|
||||
README.md: d16e395a42e491461c0862227205931894c27e39
|
||||
README.zh.md: 3fb4181ce7ae7b0d79a13ca4358b9df39d83ef1f
|
||||
|
||||
@@ -37,6 +37,7 @@ Packages live at `packages/<group>/<pkg>/`; groups are containers, while names r
|
||||
| [`session-persistence/`](session-persistence/README.md) | Persistence capability family: the seam + JSONL/SQLite backends | Product — stable surface |
|
||||
| [`session-query/`](session-query/README.md) | Session retrieval family: logical corpus, bounded reads, lineage, event relationships, semantic filtering, and SQLite full-text search | Product — stable surface |
|
||||
| [`session-title/`](session-title/README.md) | Log-backed session titles: fallback service, shared LLM policy, and opt-in providers | Product — stable surface |
|
||||
| [`telemetry/`](telemetry/README.md) | Session reporting: capture/redact seam, OTel backend | Product — stable surface |
|
||||
| [`storage/`](storage/README.md) | Non-session storage hub + backends + domain form | Product — stable surface |
|
||||
| [`workspace/`](workspace/README.md) | Workspace entity | Product — stable surface |
|
||||
| [`sdk/`](sdk/README.md) | Project SDK tooling | Product — stable surface |
|
||||
|
||||
@@ -37,6 +37,7 @@
|
||||
| [`session-persistence/`](session-persistence/README.md) | 持久化能力系列：seam + JSONL/SQLite 后端 | 产品：稳定表面 |
|
||||
| [`session-query/`](session-query/README.md) | 会话检索系列：逻辑语料库、有界读取、血缘、事件关系、语义过滤和 SQLite 全文搜索 | 产品：稳定表面 |
|
||||
| [`session-title/`](session-title/README.md) | 日志支撑的会话标题：回退服务、共享 LLM 策略和选用提供方 | 产品：稳定表面 |
|
||||
| [`telemetry/`](telemetry/README.md) | 会话上报：捕获／脱敏 seam、OTel 后端 | 产品：稳定表面 |
|
||||
| [`storage/`](storage/README.md) | 非会话存储中枢 + 后端 + 领域形式 | 产品：稳定表面 |
|
||||
| [`workspace/`](workspace/README.md) | Workspace 实体 | 产品：稳定表面 |
|
||||
| [`sdk/`](sdk/README.md) | 项目 SDK 工具 | 产品：稳定表面 |
|
||||
|
||||
@@ -128,6 +128,11 @@
|
||||
gap: 10px;
|
||||
width: 100%;
|
||||
min-height: 42px;
|
||||
/* Rows are the scroll content, never the slack absorber: a shrinkable row
|
||||
collapses to min-height while its wrapped copy keeps the taller
|
||||
intrinsic height, and centered content then paints outside the row box —
|
||||
over the title and the next row. Overflow belongs to .options. */
|
||||
flex-shrink: 0;
|
||||
padding: 5px 8px;
|
||||
border: 1px solid transparent;
|
||||
border-radius: 12px;
|
||||
@@ -208,6 +213,9 @@
|
||||
}
|
||||
|
||||
.custom {
|
||||
/* Same reason as .option: the custom block is scroll content, and shrinking
|
||||
it pushes its trigger row (and the open textarea) past the footer. */
|
||||
flex-shrink: 0;
|
||||
border: 1px solid transparent;
|
||||
border-radius: 12px;
|
||||
}
|
||||
|
||||
@@ -822,6 +822,24 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
key: 'telemetry',
|
||||
summary: 'The backend contract in its loadable form: one implementation per context — the cordis `Service` registration under the `telemetry` key throws on a duplicate, cordis\' standard behavior.',
|
||||
methods: [
|
||||
{
|
||||
signature: 'abstract emit(record: TelemetryRecord): void',
|
||||
jsDoc: '/**\n * See {@link TelemetryBackend.emit} — the seam declaration is the contract\'s one home.\n * @param record - the logical record to report; owned by the backend after the call.\n */',
|
||||
},
|
||||
{
|
||||
signature: 'flush?(): void',
|
||||
jsDoc: '/** See {@link TelemetryBackend.flush}. */',
|
||||
},
|
||||
{
|
||||
signature: 'abstract shutdown(): Promise<void>',
|
||||
jsDoc: '/**\n * See {@link TelemetryBackend.shutdown}.\n * @returns resolves when the backend\'s pipeline has quiesced.\n */',
|
||||
},
|
||||
],
|
||||
},
|
||||
{
|
||||
key: 'tokenMeter',
|
||||
summary: 'Replay owner for one service-wide estimator and isolated per-session folds.',
|
||||
@@ -1261,6 +1279,13 @@ export const EVENT_API: readonly EventApiEntry[] = [
|
||||
jsDoc: '/**\n * Emitted when any prompt provider changes. This registry notification is\n * unfiltered because a global change affects every scope.\n * @mode emit\n */',
|
||||
summary: 'Emitted when any prompt provider changes.',
|
||||
},
|
||||
{
|
||||
name: 'telemetry/record',
|
||||
mode: 'waterfall',
|
||||
signature: '\'telemetry/record\'(record: TelemetryRecord, next: () => TelemetryRecord): TelemetryRecord',
|
||||
jsDoc: '/**\n * Transform one outbound record before it reaches the backend. This\n * waterfall is the seam\'s redaction extension point. It ships NO rules\n * of its own: the\n * innermost `next()` passes the record through unchanged, and with no\n * listener mounted records reach the backend as captured, so exported\n * data is exactly as clean as the rules a deployment mounts. Listeners\n * stack by transforming `next()`\'s return value; returning without\n * `next()` replaces everything beneath. Dispatched synchronously on the\n * capture hot path inside the coordinator\'s containment: a throwing\n * listener withholds that one record (fail-closed) and never reaches the\n * agent loop. Redaction applies to the exported copy only; the canonical\n * session log is never rewritten.\n * @param record - the candidate record, already the coordinator\'s own deep\n * copy; listeners return a (possibly new) record and must not mutate it.\n * @mode waterfall\n */',
|
||||
summary: 'Transform one outbound record before it reaches the backend.',
|
||||
},
|
||||
{
|
||||
name: 'tools/change',
|
||||
mode: 'emit',
|
||||
@@ -1987,7 +2012,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
},
|
||||
{
|
||||
name: 'Session',
|
||||
declaration: 'export class Session {\n get surface(): SessionSurface;\n readonly header: SessionHeader;\n get id(): SessionId;\n constructor(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader);\n get events(): readonly SessionEvent[];\n get seq(): number;\n append<T extends SessionEventType>(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent<T>;\n requestHeader(): EpochHeader | undefined;\n deriveMessages(): Message[];\n deriveEventMessage(event: SessionEvent): Message | null;\n}',
|
||||
declaration: 'export class Session {\n get surface(): SessionSurface;\n readonly header: SessionHeader;\n get id(): SessionId;\n readonly firstLiveSeq: number;\n constructor(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader);\n get events(): readonly SessionEvent[];\n get seq(): number;\n append<T extends SessionEventType>(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent<T>;\n requestHeader(): EpochHeader | undefined;\n deriveMessages(): Message[];\n deriveEventMessage(event: SessionEvent): Message | null;\n}',
|
||||
},
|
||||
{
|
||||
name: 'SessionAvailability',
|
||||
@@ -2377,6 +2402,14 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'TaskStatus',
|
||||
declaration: 'export type TaskStatus = \'running\' | \'stopping\' | \'completed\' | \'killed\' | \'failed\';',
|
||||
},
|
||||
{
|
||||
name: 'TelemetryRecord',
|
||||
declaration: 'export interface TelemetryRecord {\n channel: \'ledger\' | \'ops\';\n time: number;\n severity: TelemetrySeverity;\n attributes: Record<string, string | number>;\n body: unknown;\n}',
|
||||
},
|
||||
{
|
||||
name: 'TelemetrySeverity',
|
||||
declaration: 'export type TelemetrySeverity = \'info\' | \'warn\' | \'error\';',
|
||||
},
|
||||
{
|
||||
name: 'TerminalCallView',
|
||||
declaration: 'export interface TerminalCallView {\n card: \'terminal\';\n title: string;\n description?: string;\n cwd?: string;\n}',
|
||||
|
||||
@@ -305,6 +305,19 @@ export class Session {
|
||||
return this.header.id
|
||||
}
|
||||
|
||||
/**
|
||||
* The first seq appended IN THIS PROCESS: the length of the constructor
|
||||
* seed (0 without one). Events below it entered through construction —
|
||||
* replay, fork, or resume — and were never published on the `session/event`
|
||||
* firehose (constructor seeds do not emit), so consumers that replay the
|
||||
* log as a publication substitute (telemetry adoption) start here. Distinct
|
||||
* from `header.seedLength`, the DURABLE fork-lineage boundary: a resumed
|
||||
* session's constructor seed is its full stored log, while its header keeps
|
||||
* the original fork value — this field is the in-process construction fact
|
||||
* and is deliberately not persisted.
|
||||
*/
|
||||
readonly firstLiveSeq: number
|
||||
|
||||
constructor(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader) {
|
||||
if (seed) {
|
||||
// Validate the seed to the SAME invariants `append` enforces, so a
|
||||
@@ -337,6 +350,7 @@ export class Session {
|
||||
this.log.push(deepFreeze(snapshot))
|
||||
}
|
||||
}
|
||||
this.firstLiveSeq = this.log.length
|
||||
this.header = snapshotSessionHeader(id, header)
|
||||
}
|
||||
|
||||
|
||||
6
packages/telemetry/README.i18n.yaml
Normal file
6
packages/telemetry/README.i18n.yaml
Normal file
@@ -0,0 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# 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/README.md
|
||||
README.md: 944cb3f9bac6169feddf8b49bc481cfbe7c6fa9d
|
||||
README.zh.md: 7be8af93654565cd9fb4d8b4376e0960fd7eb72b
|
||||
10
packages/telemetry/README.md
Normal file
10
packages/telemetry/README.md
Normal file
@@ -0,0 +1,10 @@
|
||||
# telemetry/
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
Outbound session reporting: the telemetry seam plus its OpenTelemetry backend. The design — the boundary axiom (the harness's aspect ends at `emit()`; delivery is the reporting SDK's), the `telemetry/record` waterfall (deployment-mounted redaction rules; the seam ships none), the fixed chunk projection, the handoff cursor, and the operational-record channel — is pinned in [the revival Agent Note](../../.agents/notes/implemented/feature/2026-07-23-session-telemetry-otel-revival.md).
|
||||
|
||||
| Package | Role |
|
||||
|---|---|
|
||||
| [`@deepseek-ai/dsh-session-telemetry`](session-telemetry/) | The seam: capture points, projection, redaction, handoff cursor, ops signals, and the minimal backend contract (`emit`/`flush?`/`shutdown`). |
|
||||
| [`@deepseek-ai/dsh-session-telemetry-otel`](session-telemetry-otel/) | The backend a deployment loads: the OTel JS SDK's log pipeline (`LoggerProvider` + `BatchLogRecordProcessor` + OTLP/HTTP exporter), configured verbatim through passthroughs. |
|
||||
10
packages/telemetry/README.zh.md
Normal file
10
packages/telemetry/README.zh.md
Normal file
@@ -0,0 +1,10 @@
|
||||
# telemetry/
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
面向外部的会话上报：遥测（telemetry）seam 及其 OpenTelemetry 后端。整套设计归档于[复活 Agent Note（agent 决策记录）](../../.agents/notes/implemented/feature/2026-07-23-session-telemetry-otel-revival.md)：边界公理（harness 的职责止于 `emit()`，投递由上报 SDK 负责）、`telemetry/record` waterfall（瀑布式事件；脱敏规则由部署方挂载，seam 自身不带任何规则）、固定分片投影、handoff 游标，以及运维记录通道。
|
||||
|
||||
| 包 | 职责 |
|
||||
|---|---|
|
||||
| [`@deepseek-ai/dsh-session-telemetry`](session-telemetry/) | seam 本体：捕获点、投影、脱敏、handoff 游标、运维信号，以及最小后端契约（`emit`/`flush?`/`shutdown`）。 |
|
||||
| [`@deepseek-ai/dsh-session-telemetry-otel`](session-telemetry-otel/) | 部署方要加载的后端：OTel JS SDK 的日志流水线（`LoggerProvider` + `BatchLogRecordProcessor` + OTLP/HTTP 导出器），经透传（passthrough）原样配置。 |
|
||||
@@ -0,0 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# 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: 9b208e291e77bee50d9d4fd14808268dca75f2db
|
||||
README.zh.md: f36cfe74146b779c4ddb1101227cb45a06f0968c
|
||||
41
packages/telemetry/session-telemetry-otel/README.md
Normal file
41
packages/telemetry/session-telemetry-otel/README.md
Normal file
@@ -0,0 +1,41 @@
|
||||
# @deepseek-ai/dsh-session-telemetry-otel
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
The OpenTelemetry backend for [the telemetry seam](../session-telemetry/) — the only entry a deployment loads. It composes the OTel JS SDK as-is (`LoggerProvider` → `BatchLogRecordProcessor` → OTLP/HTTP log exporter) and maps each record the seam hands over onto `logger.emit()`, under two instrumentation scopes: ledger records on `@deepseek-ai/dsh-session-telemetry-otel`, operational records on `@deepseek-ai/dsh-session-telemetry-otel/ops`. Resource identity (`service.name`/`service.version`) comes from `dsh-llm`'s `APP_IDENTITY`, the same source the attribution headers use.
|
||||
|
||||
## Config
|
||||
|
||||
```yaml
|
||||
- id: telemetry-otel
|
||||
name: '@deepseek-ai/dsh-session-telemetry-otel'
|
||||
config:
|
||||
exporter: # passed verbatim to the SDK's OTLP/HTTP log exporter
|
||||
url: https://collector.example.com/v1/logs
|
||||
headers:
|
||||
authorization: !!js `Bearer ${process.env.OTLP_TOKEN}`
|
||||
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.
|
||||
|
||||
## What leaves the machine
|
||||
|
||||
Records carry the complete `event.data` as the seam's `telemetry/record` waterfall returns it — user and assistant message content, tool arguments and results (command output, file contents), the full system prompt and tool schemas (`request/header`), todo text, compaction summaries, hook `stderrSummary`, and the session `cwd` (a local path). The seam ships no redaction rules: with no `telemetry/record` listener mounted, that is the raw captured copy, so a deployment exporting beyond a trusted boundary mounts its own rules (see [the seam README](../session-telemetry/README.md#the-redact-waterfall)). Provider credentials never appear regardless: adapter API keys are constructor parameters, not session events, so they are structurally absent from the log and therefore from telemetry.
|
||||
|
||||
## Field mapping
|
||||
|
||||
Seam record → SDK log record: `time` → `timestamp`/`observedTimestamp`; `severity` → `severityNumber`/`severityText` (INFO 9 / WARN 13 / ERROR 17); `body` → the structured log body; `attributes` verbatim. Receivers dedupe on `(session.id, event.seq)`, alert on severity, and detect crashes by `shutdown`-record absence (a session with activity, no `shutdown` ops record, gone stale ended uncleanly). The marker means telemetry stopped observing the session cleanly — emitted at the session's own disposal, or at application teardown for sessions still running then; a marker followed by more of that session's events is a telemetry reload, not a session restart. Streams are not self-contained across lineage: a resumed session continues its own id's stream from where the previous process left off, and a forked session's stream starts at its inherited boundary — its prefix lives in the parent's stream, stitched via `session.parent_id` + `session.seed_length`. One consequence of continuing rather than replaying: a turn left open mid-stream and never closed marks the previous process dying inside it. The local log is repaired with synthetic closers at resume, but those repairs are never exported — the wire stream stays faithful to what the crashed process actually shipped, and a later clean `shutdown` marker attests only to the resumed process's own exit.
|
||||
|
||||
## Model Experience
|
||||
|
||||
None, as the backend only forwards the seam's redacted records into the OTel SDK pipeline; it never contributes to a model request.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
None; this package neither assembles nor sends a provider request.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Upstream experimental tree** — `@opentelemetry/sdk-logs` is still published from the upstream experimental tree; SDK API churn lands here and only here — the seam contract does not move.
|
||||
- **No live-collector coverage** — every test exports to a local mock collector; the keyless Loader-composition e2e (`tests/loader-composition.e2e.ts`) covers the wire shape on every run, and behavior against a real OTLP deployment (auth, TLS, throttling) is the SDK exporter's documented territory.
|
||||
41
packages/telemetry/session-telemetry-otel/README.zh.md
Normal file
41
packages/telemetry/session-telemetry-otel/README.zh.md
Normal file
@@ -0,0 +1,41 @@
|
||||
# @deepseek-ai/dsh-session-telemetry-otel
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
[遥测（telemetry）seam](../session-telemetry/) 的 OpenTelemetry 后端，也是部署方唯一要加载的条目。它原样组合 OTel JS SDK（`LoggerProvider` → `BatchLogRecordProcessor` → OTLP/HTTP 日志导出器），把 seam 交接过来的每条记录映射到 `logger.emit()`，并使用两个插桩作用域（instrumentation scope）：ledger 记录挂在 `@deepseek-ai/dsh-session-telemetry-otel` 下，运维记录挂在 `@deepseek-ai/dsh-session-telemetry-otel/ops` 下。资源身份（`service.name`/`service.version`）来自 `dsh-llm` 的 `APP_IDENTITY`，与归因标头同源。
|
||||
|
||||
## 配置
|
||||
|
||||
```yaml
|
||||
- id: telemetry-otel
|
||||
name: '@deepseek-ai/dsh-session-telemetry-otel'
|
||||
config:
|
||||
exporter: # passed verbatim to the SDK's OTLP/HTTP log exporter
|
||||
url: https://collector.example.com/v1/logs
|
||||
headers:
|
||||
authorization: !!js `Bearer ${process.env.OTLP_TOKEN}`
|
||||
processor: {} # optional; passed verbatim to BatchLogRecordProcessor
|
||||
```
|
||||
|
||||
`exporter.url` 是本包唯一自行校验的字段：必填、无默认值、必须能解析为 `http(s)`，因此缺失端点会在插件加载时失败（`processor.maxExportBatchSize` 不是正整数时同样如此：SDK 会接受该值，随后却在关闭时因它挂起）。其余全部是 SDK 自己的选项形态，由 SDK 拥有并在 SDK 文档中说明，两个配置块都整体透传（passthrough）：`OTLPExporterNodeConfigBase` 的每个字段（`headers`、`timeoutMillis`、`compression`、`keepAlive` 等）都会到达导出器；批处理、导出节奏（`scheduledDelayMillis`）、重试、队列上限，以及持续失败下的丢失策略，都是 SDK 的文档化行为，经 `processor` 透传调优。该后端刻意不实现 `flush()`：批处理器是进程内唯一执行 flush 的组件，`shutdown()` 的排空正因如此才是完整的。从 `cordis.yml` 中删除该配置块即为退出方式：无残留状态，也没有 `enabled` 开关。
|
||||
|
||||
## 哪些数据会离开本机
|
||||
|
||||
记录携带完整的 `event.data`，内容以 seam 的 `telemetry/record` waterfall（瀑布式事件）返回的结果为准：用户与 assistant 消息内容、工具参数与工具结果（命令输出、文件内容）、完整的系统提示词与工具 schema（`request/header`）、todo 文本、压缩（compaction）摘要、钩子的 `stderrSummary`，以及会话 `cwd`（一个本地路径）。seam 不带任何脱敏规则：未挂载 `telemetry/record` 监听器时，导出的就是捕获原样的副本，因此向可信边界之外导出的部署方要挂载自己的规则（见 [seam README](../session-telemetry/README.md#the-redact-waterfall)）。无论如何，提供方凭据都不会出现：适配器的 API key 是构造函数参数而非会话事件，因此它们在结构上就不存在于日志中，也就不存在于遥测中。
|
||||
|
||||
## 字段映射
|
||||
|
||||
seam 记录 → SDK 日志记录：`time` → `timestamp`/`observedTimestamp`；`severity` → `severityNumber`/`severityText`（INFO 9 / WARN 13 / ERROR 17）；`body` → 结构化日志 body；`attributes` 原样照搬。接收端基于 `(session.id, event.seq)` 去重、按严重级别告警，并通过 `shutdown` 记录的缺失检测崩溃（一个曾有活动、没有 `shutdown` 运维记录、且已然陈旧的会话，就是未干净结束的会话）。该标记的含义是遥测干净地停止了对该会话的观察：它在会话自身 dispose（资源释放）时发出，对于届时仍在运行的会话，则在应用拆卸时发出；标记之后又出现该会话的更多事件，说明发生的是遥测重载，而不是会话重启。跨谱系（lineage）的流并不自足：恢复的会话在其自身 id 的流上从上一个进程停止之处继续；fork 出的会话，其流从继承边界开始，前缀位于父会话的流中，由接收端基于 `session.parent_id` + `session.seed_length` 拼接。继续而非回放的一个后果：流中一个开启后再未关闭的轮次，标志着上一个进程死在了该轮次之内。恢复时本地日志会以合成的关闭事件修复，但这些修复绝不导出：导出的流忠实于崩溃进程实际发出的内容，其后干净的 `shutdown` 标记也只证明恢复后进程自身的退出。
|
||||
|
||||
## 模型体验
|
||||
|
||||
无。该后端只把 seam 脱敏后的记录转发进 OTel SDK 流水线；它绝不向模型请求贡献任何内容。
|
||||
|
||||
#### KV Cache 影响
|
||||
|
||||
无；本包既不组装也不发送提供方请求。
|
||||
|
||||
## 已知限制与延期工作
|
||||
|
||||
- **上游实验性源码树**：`@opentelemetry/sdk-logs` 仍从上游实验性（experimental）源码树发布；SDK API 的变动只会落在本包，也仅落在本包；seam 契约不动。
|
||||
- **无真实 collector 覆盖**：所有测试都导出到本地 mock collector；无密钥的 Loader 组合 e2e（`tests/loader-composition.e2e.ts`）在每次运行中都覆盖协议格式（wire format）形态，而面对真实 OTLP 部署的行为（认证、TLS、限流）属于 SDK 导出器文档的职责范围。
|
||||
53
packages/telemetry/session-telemetry-otel/package.json
Normal file
53
packages/telemetry/session-telemetry-otel/package.json
Normal file
@@ -0,0 +1,53 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-session-telemetry-otel",
|
||||
"description": "OpenTelemetry backend for the DeepSeek Harness telemetry seam: hands captured session records to the OTel JS SDK's log pipeline",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "lib/index.js",
|
||||
"types": "lib/types/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./lib/types/index.d.ts",
|
||||
"default": "./lib/index.js"
|
||||
},
|
||||
"./invariant": {
|
||||
"types": "./lib/types/invariant.d.ts",
|
||||
"default": "./lib/invariant.js"
|
||||
},
|
||||
"./src/*": "./src/*",
|
||||
"./package.json": "./package.json"
|
||||
},
|
||||
"files": [
|
||||
"lib/index.js",
|
||||
"lib/invariant.js",
|
||||
"lib/types/**/*.d.ts",
|
||||
"lib/types/**/*.d.ts.map",
|
||||
"src"
|
||||
],
|
||||
"license": "BSD-3-Clause",
|
||||
"dependencies": {
|
||||
"@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"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-invariants": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-telemetry": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@cordisjs/plugin-loader": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-telemetry": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
}
|
||||
}
|
||||
181
packages/telemetry/session-telemetry-otel/src/index.ts
Normal file
181
packages/telemetry/session-telemetry-otel/src/index.ts
Normal file
@@ -0,0 +1,181 @@
|
||||
/**
|
||||
* OpenTelemetry backend for the DeepSeek Harness telemetry seam.
|
||||
*
|
||||
* Composes the OTel JS SDK as-is — a `LoggerProvider` with a
|
||||
* `BatchLogRecordProcessor` and an OTLP/HTTP log exporter — and maps each
|
||||
* 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.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-session-telemetry-otel
|
||||
*/
|
||||
|
||||
import { createRequire } from 'node:module'
|
||||
import z from 'schemastery'
|
||||
import type { Context } from 'cordis'
|
||||
import { Telemetry, TelemetryCoordinator, type TelemetryRecord, type TelemetrySeverity } from '@deepseek-ai/dsh-session-telemetry'
|
||||
import { APP_IDENTITY } from '@deepseek-ai/dsh-llm'
|
||||
import {
|
||||
BatchLogRecordProcessor,
|
||||
LoggerProvider,
|
||||
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'
|
||||
|
||||
// The package's own manifest is the single source of the instrumentation-scope
|
||||
// version (same pattern as dsh-llm's attribution identity).
|
||||
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.
|
||||
*/
|
||||
export interface Config {
|
||||
/**
|
||||
* 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
|
||||
}
|
||||
/**
|
||||
* Passed verbatim to `BatchLogRecordProcessor` (minus the exporter slot,
|
||||
* which this plugin fills); the SDK owns and documents these knobs.
|
||||
*/
|
||||
processor?: Omit<BatchLogRecordProcessorOptions, 'exporter'>
|
||||
}
|
||||
|
||||
/**
|
||||
* 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;
|
||||
* 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(),
|
||||
})
|
||||
|
||||
/** 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' },
|
||||
warn: { severityNumber: SeverityNumber.WARN, severityText: 'WARN' },
|
||||
error: { severityNumber: SeverityNumber.ERROR, severityText: 'ERROR' },
|
||||
}
|
||||
|
||||
/**
|
||||
* The backend plugin — the only entry a deployment loads. Constructing it
|
||||
* wires the SDK pipeline, registers the `telemetry` service (duplicate load
|
||||
* throws, cordis' standard duplicate-service behavior), and composes the
|
||||
* seam's {@link TelemetryCoordinator}, which installs the capture side onto
|
||||
* this fiber.
|
||||
*/
|
||||
export class TelemetryOtel extends Telemetry {
|
||||
static inject = ['sessions']
|
||||
static Config = Config
|
||||
|
||||
private readonly provider: LoggerProvider
|
||||
private readonly ledger: Logger
|
||||
private readonly ops: Logger
|
||||
|
||||
constructor(ctx: Context, config: Config) {
|
||||
super(ctx)
|
||||
const url = config.exporter?.url
|
||||
if (url === undefined || url.length === 0) {
|
||||
throw new Error('session-telemetry-otel: exporter.url is required (the full OTLP logs endpoint)')
|
||||
}
|
||||
let parsed: URL
|
||||
try {
|
||||
parsed = new URL(url)
|
||||
} catch {
|
||||
// Re-thrown as a config error: the only way here is a malformed url string.
|
||||
throw new Error(`session-telemetry-otel: exporter.url is not a valid URL: ${JSON.stringify(url)}`)
|
||||
}
|
||||
if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') {
|
||||
throw new Error(`session-telemetry-otel: exporter.url must be http(s), got ${parsed.protocol}`)
|
||||
}
|
||||
// The one processor field checked beyond the SDK's own validation: the
|
||||
// SDK accepts a non-positive batch size, but its shutdown drain then
|
||||
// splices empty batches without consuming the queue — dispose would hang
|
||||
// forever with records queued. Misconfiguration fails at load instead.
|
||||
const batchSize = config.processor?.maxExportBatchSize
|
||||
if (batchSize !== undefined && (!Number.isInteger(batchSize) || batchSize < 1)) {
|
||||
throw new Error(`session-telemetry-otel: processor.maxExportBatchSize must be a positive integer, got ${String(batchSize)}`)
|
||||
}
|
||||
this.provider = new LoggerProvider({
|
||||
resource: resourceFromAttributes({
|
||||
'service.name': APP_IDENTITY.product,
|
||||
'service.version': APP_IDENTITY.version,
|
||||
}),
|
||||
processors: [
|
||||
new BatchLogRecordProcessor({
|
||||
...config.processor,
|
||||
// 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),
|
||||
}),
|
||||
],
|
||||
})
|
||||
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)
|
||||
new TelemetryCoordinator(ctx, this)
|
||||
}
|
||||
|
||||
/**
|
||||
* Map one seam record onto the SDK logger for its channel — a synchronous
|
||||
* enqueue into the batch processor's queue.
|
||||
* @param record - the logical record handed over by the coordinator.
|
||||
*/
|
||||
emit(record: TelemetryRecord): void {
|
||||
const logger = record.channel === 'ops' ? this.ops : this.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,
|
||||
})
|
||||
}
|
||||
|
||||
// The seam's optional flush() hint is deliberately NOT implemented. The
|
||||
// batch processor exports on its own cadence (`processor.scheduledDelayMillis`,
|
||||
// the SDK's documented knob), and this backend is the SDK pipeline's only
|
||||
// caller — forwarding the hint to `forceFlush()` was the sole source of
|
||||
// concurrent flushes, whose undocumented interactions with shutdown's
|
||||
// internal drain (concurrent-flush guard, provider-level flush timeout)
|
||||
// silently dropped tail records. Removal history and the revival trigger:
|
||||
// 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.
|
||||
*/
|
||||
shutdown(): Promise<void> {
|
||||
return this.provider.shutdown()
|
||||
}
|
||||
}
|
||||
|
||||
export default TelemetryOtel
|
||||
32
packages/telemetry/session-telemetry-otel/src/invariant.ts
Normal file
32
packages/telemetry/session-telemetry-otel/src/invariant.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
/**
|
||||
* Package-owned invariant companion for `@deepseek-ai/dsh-session-telemetry-otel`.
|
||||
* @module @deepseek-ai/dsh-session-telemetry-otel/invariant
|
||||
*/
|
||||
|
||||
/* jscpd:ignore-start */
|
||||
import type { Context } from 'cordis'
|
||||
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
|
||||
|
||||
const PACKAGE_NAME = '@deepseek-ai/dsh-session-telemetry-otel'
|
||||
|
||||
/** Cordis companion plugin name. */
|
||||
export const name = 'session-telemetry-otel-invariant'
|
||||
/** Service required before the companion can reserve package ownership. */
|
||||
export const inject = ['invariants']
|
||||
|
||||
/**
|
||||
* No runtime invariant: the backend forwards seam records into the OTel SDK's
|
||||
* in-process pipeline and appends nothing to any session; its only observable
|
||||
* effects (batching, export) happen inside the SDK past the seam's boundary
|
||||
* axiom, out of reach of an independent companion.
|
||||
*/
|
||||
const install: InvariantInstaller = () => {}
|
||||
|
||||
/**
|
||||
* Register this package's invariant companion.
|
||||
* @param ctx - Cordis context carrying the invariant service.
|
||||
* @returns the installed registration's disposer after setup succeeds.
|
||||
*/
|
||||
export const apply = (ctx: Context): Promise<() => void> =>
|
||||
Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
|
||||
/* jscpd:ignore-end */
|
||||
@@ -0,0 +1,99 @@
|
||||
/**
|
||||
* REAL-composition tier: boot the examples-owned telemetry Loader fixture as
|
||||
* a subprocess (per testing policy, through the same app/boot path a
|
||||
* deployment uses), run one mocked-model turn with a real bash round trip,
|
||||
* and assert against what the mock OTLP collector actually received on the
|
||||
* wire: ledger mirroring, the deployment-mounted redact rule applied to the
|
||||
* exported copy, ops markers, and the untouched canonical log.
|
||||
*/
|
||||
|
||||
import { readFile, readdir } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { LOADER_SMOKE_TEST_TIMEOUT_MS, runLoaderSmoke } from '@deepseek-ai/dsh-loader-smoke'
|
||||
|
||||
const driver = fileURLToPath(new URL(
|
||||
'../../../../examples/headless-agent/tests/fixtures/telemetry-otel-driver.ts',
|
||||
import.meta.url,
|
||||
))
|
||||
const configPath = fileURLToPath(new URL(
|
||||
'../../../../examples/headless-agent/tests/fixtures/telemetry-otel.cordis.yml',
|
||||
import.meta.url,
|
||||
))
|
||||
const repoTsconfig = fileURLToPath(new URL('../../../../tsconfig.json', import.meta.url))
|
||||
|
||||
const FIXTURE_SECRET = 'sk-e2efixture1234567890'
|
||||
const FIXTURE_PLACEHOLDER = '[E2E-REDACTED]'
|
||||
|
||||
interface OtlpLogRecord {
|
||||
attributes?: { key: string; value: Record<string, unknown> }[]
|
||||
body?: unknown
|
||||
}
|
||||
|
||||
interface OtlpCapture {
|
||||
resourceLogs: {
|
||||
scopeLogs: {
|
||||
scope: { name: string }
|
||||
logRecords: OtlpLogRecord[]
|
||||
}[]
|
||||
}[]
|
||||
}
|
||||
|
||||
async function jsonlFiles(dir: string): Promise<string[]> {
|
||||
const entries = await readdir(dir, { withFileTypes: true })
|
||||
const paths = await Promise.all(entries.map(async (entry) => {
|
||||
const path = join(dir, entry.name)
|
||||
if (entry.isDirectory()) return jsonlFiles(path)
|
||||
return entry.isFile() && entry.name.endsWith('.jsonl') ? [path] : []
|
||||
}))
|
||||
return paths.flat()
|
||||
}
|
||||
|
||||
describe('session-telemetry-otel through a real headless cordis.yml', () => {
|
||||
it('exports redacted ledger records to the collector while the canonical log keeps the secret', async () => {
|
||||
let captures: OtlpCapture[] = []
|
||||
let logContent = ''
|
||||
const { stderr } = await runLoaderSmoke({
|
||||
label: 'session-telemetry-otel loader smoke',
|
||||
tempDirPrefix: 'telemetry-otel-e2e-',
|
||||
binScript: driver,
|
||||
libBinScript: driver,
|
||||
configPath,
|
||||
tsconfigPath: repoTsconfig,
|
||||
inspect: async (cwd) => {
|
||||
captures = JSON.parse(await readFile(join(cwd, 'otlp-captures.json'), 'utf8')) as OtlpCapture[]
|
||||
const logs = await jsonlFiles(join(cwd, '.sessions'))
|
||||
expect(logs).toHaveLength(1)
|
||||
logContent = await readFile(logs[0] as string, 'utf8')
|
||||
},
|
||||
})
|
||||
expect(stderr).not.toContain('UNHANDLED')
|
||||
|
||||
const records = captures.flatMap(capture => capture.resourceLogs.flatMap(resource =>
|
||||
resource.scopeLogs.flatMap(scoped => scoped.logRecords.map(record => ({ scope: scoped.scope.name, record })))))
|
||||
expect(records.length).toBeGreaterThan(0)
|
||||
|
||||
const eventTypes = records.flatMap(({ record }) =>
|
||||
record.attributes?.flatMap(attribute =>
|
||||
attribute.key === 'event.type' && typeof attribute.value['stringValue'] === 'string'
|
||||
? [attribute.value['stringValue']]
|
||||
: []) ?? [])
|
||||
for (const expected of ['turn/start', 'user/message', 'tool/call', 'tool/result', 'assistant/message', 'turn/end']) {
|
||||
expect(eventTypes, expected).toContain(expected)
|
||||
}
|
||||
expect(records.some(({ scope }) => scope.endsWith('/ops'))).toBe(true)
|
||||
|
||||
// The deployment-mounted rule on the wire: the fixture credential never
|
||||
// leaves the process, its surrounding prose does, and the placeholder
|
||||
// marks the spot — the seam itself ships no rules.
|
||||
const wire = JSON.stringify(captures)
|
||||
expect(wire).not.toContain(FIXTURE_SECRET)
|
||||
expect(wire).toContain(FIXTURE_PLACEHOLDER)
|
||||
expect(wire).toContain('prove telemetry with key')
|
||||
|
||||
// The canonical session log is never rewritten.
|
||||
expect(logContent).toContain(FIXTURE_SECRET)
|
||||
expect(logContent).not.toContain(FIXTURE_PLACEHOLDER)
|
||||
}, LOADER_SMOKE_TEST_TIMEOUT_MS)
|
||||
})
|
||||
237
packages/telemetry/session-telemetry-otel/tests/otel.spec.ts
Normal file
237
packages/telemetry/session-telemetry-otel/tests/otel.spec.ts
Normal file
@@ -0,0 +1,237 @@
|
||||
/**
|
||||
* OTel backend unit tier: wire assertions against a scripted `node:http`
|
||||
* mock collector through the SDK's REAL pipeline (BatchLogRecordProcessor →
|
||||
* OTLP/HTTP JSON), config fail-loud cases, and the real-Loader-path guard
|
||||
* for the default-exported Service class.
|
||||
*/
|
||||
|
||||
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'
|
||||
import TelemetryOtel, { Config } from '../src/index.ts'
|
||||
|
||||
interface Capture {
|
||||
headers: import('node:http').IncomingHttpHeaders
|
||||
body: OtlpLogsRequest
|
||||
}
|
||||
|
||||
/** Just the slice of ExportLogsServiceRequest JSON these assertions touch. */
|
||||
interface OtlpLogsRequest {
|
||||
resourceLogs: {
|
||||
resource: { attributes: { key: string; value: { stringValue?: string } }[] }
|
||||
scopeLogs: {
|
||||
scope: { name: string }
|
||||
logRecords: {
|
||||
timeUnixNano: string
|
||||
severityNumber: number
|
||||
severityText: string
|
||||
attributes?: { key: string; value: Record<string, unknown> }[]
|
||||
}[]
|
||||
}[]
|
||||
}[]
|
||||
}
|
||||
|
||||
const servers: Server[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
for (const server of servers.splice(0)) {
|
||||
server.close()
|
||||
server.closeAllConnections()
|
||||
}
|
||||
})
|
||||
|
||||
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', () => {
|
||||
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)
|
||||
server.listen(0, '127.0.0.1')
|
||||
await once(server, 'listening')
|
||||
const address = server.address()
|
||||
if (address === null || typeof address === 'string') throw new Error('no port')
|
||||
return { url: `http://127.0.0.1:${address.port}/v1/logs`, captures }
|
||||
}
|
||||
|
||||
async function boot(url: string) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(TelemetryOtel, {
|
||||
exporter: { url, headers: { authorization: 'Bearer test-token' } },
|
||||
})
|
||||
return { ctx, fiber }
|
||||
}
|
||||
|
||||
function allRecords(captures: Capture[]) {
|
||||
return captures.flatMap(c => c.body.resourceLogs.flatMap(r => r.scopeLogs.flatMap(s =>
|
||||
s.logRecords.map(record => ({ scope: s.scope.name, record })))))
|
||||
}
|
||||
|
||||
describe('TelemetryOtel wire', () => {
|
||||
it('ships session records and the ops shutdown marker through the real SDK pipeline', async () => {
|
||||
const { url, captures } = await mockCollector()
|
||||
const { ctx, fiber } = await boot(url)
|
||||
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' } })
|
||||
await fiber.dispose()
|
||||
|
||||
expect(captures.length).toBeGreaterThan(0)
|
||||
const first = captures[0]!
|
||||
const authorization: string | undefined = first.headers.authorization
|
||||
expect(authorization).toBe('Bearer test-token')
|
||||
|
||||
const resource = first.body.resourceLogs[0]!.resource.attributes
|
||||
expect(resource).toContainEqual({ key: 'service.name', value: { stringValue: 'deepseek-harness' } })
|
||||
|
||||
const records = allRecords(captures)
|
||||
const ledger = records.filter(r => r.scope === '@deepseek-ai/dsh-session-telemetry-otel')
|
||||
const ops = records.filter(r => r.scope === '@deepseek-ai/dsh-session-telemetry-otel/ops')
|
||||
|
||||
const start = ledger.find(r => r.record.attributes?.some(a => a.key === 'event.type' && a.value.stringValue === 'turn/start'))
|
||||
expect(start).toBeDefined()
|
||||
expect(start?.record.severityNumber).toBe(9)
|
||||
expect(BigInt(start!.record.timeUnixNano)).toBe(BigInt(session.events[0]!.time) * 1_000_000n)
|
||||
expect(start?.record.attributes).toContainEqual({ key: 'session.cwd', value: { stringValue: '/tmp/w' } })
|
||||
|
||||
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(ops).toHaveLength(1)
|
||||
expect(ops[0]!.record.attributes).toContainEqual({ key: 'telemetry.op', value: { stringValue: 'shutdown' } })
|
||||
})
|
||||
|
||||
it('drains records enqueued after a timer export began: dispose during an in-flight batch', async () => {
|
||||
// The backend implements NO flush() — the batch processor exports on its
|
||||
// own cadence, and shutdown's internal drain is complete exactly because
|
||||
// nothing in the process calls forceFlush() concurrently (the SDK's
|
||||
// concurrent-flush guard skips draining otherwise). Pin that: hold the
|
||||
// collector's response to the timer-triggered export open across
|
||||
// disposal, and the dispose-time shutdown marker (enqueued after that
|
||||
// batch's snapshot) must still arrive.
|
||||
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 },
|
||||
processor: { scheduledDelayMillis: 10 },
|
||||
})
|
||||
const session = ctx.sessions.create(SessionId('drain'), { meta: {} })
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
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 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()
|
||||
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 leaves the seam flush hint unimplemented', async () => {
|
||||
const { url, captures } = await mockCollector()
|
||||
const { ctx, fiber } = await boot(url)
|
||||
const session = ctx.sessions.create(SessionId('warn'), { meta: {} })
|
||||
session.append('prompt/blocked', { content: [], source: { kind: 'user' }, reason: 'vetoed' })
|
||||
// No flush(): the coordinator's optional-call forwarding no-ops, and the
|
||||
// batch processor owns export cadence end to end (see the backend note).
|
||||
expect('flush' in ctx.telemetry && ctx.telemetry.flush !== undefined).toBe(false)
|
||||
await fiber.dispose()
|
||||
const blocked = allRecords(captures).find(r =>
|
||||
r.record.attributes?.some(a => a.key === 'event.type' && a.value.stringValue === 'prompt/blocked'))
|
||||
expect(blocked?.record.severityNumber).toBe(13)
|
||||
})
|
||||
})
|
||||
|
||||
describe('TelemetryOtel config fails loud', () => {
|
||||
it.each([
|
||||
[{}, /exporter\.url is required/],
|
||||
[{ exporter: { url: '' } }, /exporter\.url is required/],
|
||||
[{ exporter: { url: 'not a url' } }, /not a valid URL/],
|
||||
[{ exporter: { url: 'ftp://collector' } }, /must be http\(s\)/],
|
||||
// The SDK accepts a non-positive batch size but its shutdown drain then
|
||||
// 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/],
|
||||
])('rejects %j at plugin load', async (config, message) => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await expect(ctx.plugin(TelemetryOtel, config as Config)).rejects.toThrow(message)
|
||||
})
|
||||
})
|
||||
|
||||
describe('dsh-session-telemetry-otel real-load-path guard', () => {
|
||||
it('keeps the Service class with inject/Config through unwrapExports', async () => {
|
||||
const module = await import('../src/index.ts')
|
||||
const loader = Object.create(Loader.prototype) as Loader
|
||||
const unwrapped = loader.unwrapExports(module) as typeof TelemetryOtel
|
||||
expect(unwrapped).toBe(TelemetryOtel)
|
||||
expect(unwrapped.inject).toEqual(['sessions'])
|
||||
expect(typeof unwrapped.Config).toBe('function')
|
||||
})
|
||||
|
||||
it('boots through the unwrapped class and registers ctx.telemetry', async () => {
|
||||
const { url } = await mockCollector()
|
||||
const module = await import('../src/index.ts')
|
||||
const loader = Object.create(Loader.prototype) as Loader
|
||||
const unwrapped = loader.unwrapExports(module) as Parameters<Context['plugin']>[0]
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(unwrapped, { exporter: { url } })
|
||||
expect(ctx.telemetry).toBeInstanceOf(TelemetryOtel)
|
||||
await fiber.dispose()
|
||||
})
|
||||
})
|
||||
33
packages/telemetry/session-telemetry-otel/tsconfig.json
Normal file
33
packages/telemetry/session-telemetry-otel/tsconfig.json
Normal file
@@ -0,0 +1,33 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib/types"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/schemastery"
|
||||
},
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../../llm/llm"
|
||||
},
|
||||
{
|
||||
"path": "../session-telemetry"
|
||||
},
|
||||
{
|
||||
"path": "../../support/invariants"
|
||||
}
|
||||
]
|
||||
}
|
||||
6
packages/telemetry/session-telemetry/README.i18n.yaml
Normal file
6
packages/telemetry/session-telemetry/README.i18n.yaml
Normal file
@@ -0,0 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# 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/README.md
|
||||
README.md: a3820ee8e91a5c513c08df76b149cf3a670ee2ca
|
||||
README.zh.md: d607a4ab4d953242ede53ff0b5e50bdf69e29ea4
|
||||
42
packages/telemetry/session-telemetry/README.md
Normal file
42
packages/telemetry/session-telemetry/README.md
Normal file
@@ -0,0 +1,42 @@
|
||||
# @deepseek-ai/dsh-session-telemetry
|
||||
|
||||
English | [中文](README.zh.md)
|
||||
|
||||
The telemetry seam: the CAPTURE side of session-event reporting, behind a backend contract any reporting SDK satisfies with zero bending. The boundary axiom that shapes everything here: **this package's aspect ends at `emit()`** — batching, retry, queueing, and loss policy belong to the backend's SDK and are neither specified nor wrapped. Rationale and rejected alternatives: [the revival Agent Note](../../../.agents/notes/implemented/feature/2026-07-23-session-telemetry-otel-revival.md).
|
||||
|
||||
## The backend contract
|
||||
|
||||
`TelemetryBackend` is three members: `emit(record)` (MUST be a non-blocking enqueue — it runs synchronously on the `session/event` hot path), optional `flush()` (a turn-boundary hint, fire-and-forget; most backends leave it unimplemented and let their SDK's batching cadence govern export timing — an implementer owns the interaction between concurrent flushes and `shutdown()`'s drain), and `shutdown()` (the lifecycle forward: drain-and-quiesce, awaited at dispose). `Telemetry` is its service-registered form under the `telemetry` context key — one implementation per context, duplicate load throws. A backend composes `TelemetryCoordinator` in its constructor.
|
||||
|
||||
## 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 from the construction boundary — constructor seeds from fork/resume never re-emit on the firehose and never re-export), `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` (emit the session's `shutdown` operational record at its own termination edge — where receivers key crash detection — then retire it, so a long-lived backend neither retains closed sessions nor re-marks them at unload), `agent/error` (the one live-bus relay; turn-enclosure structurally bars those errors from the log), a dispose effect (mark each session still alive at teardown, 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
|
||||
|
||||
Every record passes the `telemetry/record` waterfall between projection and `emit()` — the seam's scrubbing extension point. The seam ships NO rules of its own: the innermost `next()` passes the record through unchanged, so with no listener mounted records reach the backend exactly as captured, and exported data is precisely as clean as the rules a deployment mounts. Listeners stack by transforming `next()`'s return value; returning without `next()` replaces everything beneath, and a throwing listener withholds that one record fail-closed inside the coordinator's containment. Redaction applies to the exported copy only; the canonical session log is never rewritten.
|
||||
|
||||
## The handoff cursor
|
||||
|
||||
A module-scope `WeakMap<Session, seq>` marks the highest seq HANDED OFF (not delivered) per session, advanced at emit time. It survives reloads that do not re-evaluate this module — config re-applies and backend source reloads, which is where iteration happens; that asymmetry is why the cursor lives in the seam. On re-adoption the coordinator re-hands only events past the cursor (events at or below it still rebuild the chunk-projection state); a missing cursor safely degrades to a re-hand from the session's construction boundary (`Session.firstLiveSeq` — seq 0 for a session born in this process), absorbed by receiver-side dedupe on `(session.id, event.seq)`. Constructor seeds never re-export: a resumed session's history shipped from the previous process under the same id, and a fork's inherited prefix lives in the parent's stream (receivers stitch on `session.parent_id` + `session.seed_length`). The accepted cost, consistent with at-most-once delivery: a resume does not backfill records a previous process failed to deliver — a deployment with a backfill requirement needs the deferred outbox, not replay. This is a deliberate, narrow exception to the registrations-are-effects discipline: entries die with their sessions, the value is a monotonic watermark, and losing it is never an error.
|
||||
|
||||
## The fixed chunk projection
|
||||
|
||||
Only the first `assistant/chunk` of each `(turn, step)` ships; the rest are dropped at capture and never advance the cursor. That one chunk is the stream-started signal: `step/start` + first-chunk presence + `assistant/message` presence + the `turn/end` reason distinguish "the request never started" from "the stream died midway" without chunk volume, and time-to-first-token stays computable. Chunk elision makes `seq` gaps routine on the wire — a gap is never a loss signal. Every other event type, including ones merged by plugins this package never heard of, passes through whole.
|
||||
|
||||
## The logical record
|
||||
|
||||
`TelemetryRecord`: `channel` (`ledger` | `ops`), `time` (epoch ms), `severity` (pre-mapped: ERROR for `tool/result.isError` and `turn/end` error reasons; WARN for `prompt/blocked`; INFO otherwise, including plugin-merged event types whose outcome semantics stay with their owners), identity-only `attributes` (`session.id`, `event.type`, `event.seq`, plus `session.cwd`/`session.parent_id`/`session.seed_length` when the header has them), and the complete deep-copied `event.data` as `body` — post-redaction. Operational records carry `telemetry.op` (`agent-error` | `shutdown`) and `session.id`, and deliberately NO `event.seq`/`event.type` — signals to alert on, not entries to sum. Delivery downstream of the handoff is the backend SDK's; duplicates remain possible (cursor-less re-adoption, SDK retries), so receivers dedupe on `(session.id, event.seq)`.
|
||||
|
||||
## Model Experience
|
||||
|
||||
None, as the seam only observes the session stream and hands redacted copies to a reporting backend; it never contributes to a model request.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
None; this package neither assembles nor sends a provider request.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Best-effort delivery** — the cursor marks handed-off, not delivered; a session torn down inside a reload window cannot be re-adopted; whatever sits in a backend queue at crash time is lost. A durable outbox (spool, per-sink cursors, at-least-once) is deferred until a deployment states a crash-loss requirement — see [the revival Agent Note](../../../.agents/notes/implemented/feature/2026-07-23-session-telemetry-otel-revival.md).
|
||||
- **No built-in redaction rules** — with no `telemetry/record` listener mounted, records leave the process exactly as captured, including any credentials embedded in file contents or command output; a deployment exporting to a shared collector owns its rule set.
|
||||
42
packages/telemetry/session-telemetry/README.zh.md
Normal file
42
packages/telemetry/session-telemetry/README.zh.md
Normal file
@@ -0,0 +1,42 @@
|
||||
# @deepseek-ai/dsh-session-telemetry
|
||||
|
||||
[English](README.md) | 中文
|
||||
|
||||
遥测（telemetry）seam：会话事件上报的捕获侧，隔在一个后端契约之后，任何上报 SDK 都无需变形即可满足该契约。塑造本包一切设计的边界公理：**本包的职责止于 `emit()`**。批处理、重试、排队与丢失策略都属于后端自身的 SDK，本包既不为其立规，也不做包装。设计依据与被否决的替代方案见[复活 Agent Note（agent 决策记录）](../../../.agents/notes/implemented/feature/2026-07-23-session-telemetry-otel-revival.md)。
|
||||
|
||||
## 后端契约
|
||||
|
||||
`TelemetryBackend` 只有三个成员：`emit(record)`（必须是非阻塞入队；它在 `session/event` 热路径上同步执行）、可选的 `flush()`（轮次边界提示，触发后不等待结果；多数后端不实现它，而由其 SDK 的批处理节奏决定导出时机；并发 flush 与 `shutdown()` 的排空之间的交互由实现方自行负责）、以及 `shutdown()`（生命周期转发点：排空并完全停稳，在 dispose（资源释放）时被等待）。`Telemetry` 是它注册在 `telemetry` 上下文键下的服务形态：每个上下文只允许一个实现，重复加载会抛出异常。后端在其构造函数中组合 `TelemetryCoordinator`。
|
||||
|
||||
## 捕获点
|
||||
|
||||
协调器的全部注册都经由组合方 fiber 的 effect 完成：`session/created`（收养：记录 header，并经投影从构造边界起回读日志；来自 fork 或恢复的构造函数种子绝不会在 firehose 上再次发出，也绝不会再次导出）、`session/event`（投影、深拷贝、脱敏、交接；零 I/O）、`session/flush`（转发可选的 `flush()` 提示并返回 void；循环所等待的并行任务绝不能等待遥测）、`session/disposed`（在会话自身的终止边缘发出该会话的 `shutdown` 运维记录，接收端正是在这个边缘锚定崩溃检测；随后将该会话退役，因此长生命周期的后端既不会保留已关闭的会话，也不会在卸载时再次标记它们）、`agent/error`（唯一的实时总线转发；轮次封闭机制在结构上决定了这些错误进不了日志）、一个 dispose effect（拆卸时先标记每个仍存活的会话，再等待后端的 `shutdown()`；失败只发出警告而不抛出），以及对 `ctx.sessions.list()` 的收养扫描（热重载不会重放 `session/created`）。
|
||||
|
||||
## 脱敏 waterfall
|
||||
|
||||
每条记录在投影与 `emit()` 之间都要经过 `telemetry/record` waterfall（瀑布式事件），这是该 seam 的擦除扩展点。seam 自身不带任何规则：最内层的 `next()` 原样透传记录，因此未挂载监听器时，记录以捕获时的原样到达后端；导出数据能干净到什么程度，恰恰取决于部署方挂载了什么规则。监听器通过变换 `next()` 的返回值来堆叠；不调用 `next()` 就返回，即替换其下方的全部逻辑；抛出异常的监听器会在协调器的隔离范围内以 fail-closed 方式扣下这一条记录。脱敏只作用于导出副本；权威会话日志永不改写。
|
||||
|
||||
## handoff 游标
|
||||
|
||||
一个模块作用域的 `WeakMap<Session, seq>` 记录每个会话已交接（而非已投递）的最高 seq，在 emit 时推进。游标在不重新求值本模块的重载（配置重新应用、后端源码重载）中存活，而迭代恰恰发生在这类重载中；这种不对称正是游标放在 seam 一侧的原因。重新收养时，协调器只重新交接游标之后的事件（游标及其之前的事件仍用于重建分片投影状态）；游标缺失时安全退化为从会话构造边界起的重新交接（`Session.firstLiveSeq`，对在本进程中诞生的会话即 seq 0），由接收端基于 `(session.id, event.seq)` 的去重吸收。构造函数种子绝不会再次导出：恢复会话的历史已由上一个进程以同一 id 发出，fork 继承的前缀则位于父会话的流中（接收端基于 `session.parent_id` + `session.seed_length` 拼接）。由此接受的代价与至多一次（at-most-once）投递一致：恢复不会回填上一个进程未能投递的记录；有回填要求的部署需要的是已推迟的 outbox，而不是回放。这是对「注册即 effect」纪律的一次有意且范围极窄的例外：条目随其会话消亡，值是单调水位线，丢失它绝不是错误。
|
||||
|
||||
## 固定分片投影
|
||||
|
||||
每个 `(turn, step)` 只发出第一条 `assistant/chunk`；其余分片在捕获时丢弃，且绝不推进游标。这一条分片就是「流已开始」的信号：`step/start`、首分片是否存在、`assistant/message` 是否存在，加上 `turn/end` 的原因，无需分片流量即可区分「请求从未开始」与「流中途夭折」，首个 token 延迟（time-to-first-token）也仍然可以计算。分片省略使导出流中的 `seq` 缺口成为常态：缺口绝不是丢失信号。其余所有事件类型都会完整透传，包括本包从未听说过的插件所合并的事件类型。
|
||||
|
||||
## 逻辑记录
|
||||
|
||||
`TelemetryRecord` 包含：`channel`（`ledger` | `ops`）、`time`（epoch 毫秒）、`severity`（预先映射好的严重级别：`tool/result.isError` 与 `turn/end` 的错误原因映射为 ERROR，`prompt/blocked` 映射为 WARN，其余为 INFO，包括结果语义仍归其所有者的插件合并事件类型）、只含身份信息的 `attributes`（`session.id`、`event.type`、`event.seq`，header 中存在时再加 `session.cwd`/`session.parent_id`/`session.seed_length`），以及作为 `body` 的完整深拷贝 `event.data`，且以脱敏后的内容为准。运维记录携带 `telemetry.op`（`agent-error` | `shutdown`）和 `session.id`，并刻意不带 `event.seq`/`event.type`：它们是用来告警的信号，不是用来累加的条目。交接之后的投递由后端 SDK 负责；重复仍然可能出现（无游标的重新收养、SDK 重试），因此接收端基于 `(session.id, event.seq)` 去重。
|
||||
|
||||
## 模型体验
|
||||
|
||||
无。该 seam 只观察会话流，并把脱敏后的副本交给上报后端；它绝不向模型请求贡献任何内容。
|
||||
|
||||
#### KV Cache 影响
|
||||
|
||||
无；本包既不组装也不发送提供方请求。
|
||||
|
||||
## 已知限制与延期工作
|
||||
|
||||
- **尽力而为的投递**：游标标记的是已交接而非已投递；在重载窗口内被拆除的会话无法重新收养；崩溃时留在后端队列中的内容会丢失。持久 outbox（spool、每 sink 游标、at-least-once）推迟到有部署方提出明确的崩溃丢失要求时再实现；见[复活 Agent Note](../../../.agents/notes/implemented/feature/2026-07-23-session-telemetry-otel-revival.md)。
|
||||
- **不内置脱敏规则**：未挂载 `telemetry/record` 监听器时，记录以捕获时的原样离开进程，包括文件内容或命令输出中内嵌的任何凭据；向共享 collector 导出的部署方自行负责其规则集。
|
||||
41
packages/telemetry/session-telemetry/package.json
Normal file
41
packages/telemetry/session-telemetry/package.json
Normal file
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-session-telemetry",
|
||||
"description": "Telemetry seam for the DeepSeek Harness: session-event capture, projection, redaction, and handoff to a reporting backend",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "lib/index.js",
|
||||
"types": "lib/types/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./lib/types/index.d.ts",
|
||||
"default": "./lib/index.js"
|
||||
},
|
||||
"./invariant": {
|
||||
"types": "./lib/types/invariant.d.ts",
|
||||
"default": "./lib/invariant.js"
|
||||
},
|
||||
"./src/*": "./src/*",
|
||||
"./package.json": "./package.json"
|
||||
},
|
||||
"files": [
|
||||
"lib/index.js",
|
||||
"lib/invariant.js",
|
||||
"lib/types/**/*.d.ts",
|
||||
"lib/types/**/*.d.ts.map",
|
||||
"src"
|
||||
],
|
||||
"license": "BSD-3-Clause",
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-invariants": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
}
|
||||
}
|
||||
280
packages/telemetry/session-telemetry/src/coordinator.ts
Normal file
280
packages/telemetry/session-telemetry/src/coordinator.ts
Normal file
@@ -0,0 +1,280 @@
|
||||
/**
|
||||
* Capture coordinator: the seam's upstream half. Subscribes to the session
|
||||
* firehose plus the one live-bus relay (`agent/error`), applies the fixed
|
||||
* chunk projection, builds logical records, runs each through the
|
||||
* `telemetry/record` waterfall (deployment-mounted redaction rules;
|
||||
* pass-through when none), and hands the result to the backend — synchronously, with every
|
||||
* handler self-contained so a failing backend can never starve other
|
||||
* subscribers (cordis `emit` is stop-on-throw) or touch the agent loop.
|
||||
* Composed by a backend in its constructor.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-session-telemetry/coordinator
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import type { TelemetryBackend, TelemetryRecord, TelemetrySeverity } from './index.ts'
|
||||
|
||||
/**
|
||||
* The handoff cursor: per session, the highest `seq` handed to a backend.
|
||||
* Deliberately MODULE-scope ambient state — a narrow, documented exception
|
||||
* to the registrations-are-effects discipline: cordis has no HMR
|
||||
* state-handover API, and keying by the `Session` object (which belongs to
|
||||
* the session store and outlives any telemetry fiber) is the only in-process
|
||||
* lifetime that lets a re-adopting fiber resume instead of re-handing
|
||||
* history. Entries die with their sessions; a missing entry safely means
|
||||
* "re-hand everything". Advanced only at emit time — the cursor marks
|
||||
* handed-off, not delivered.
|
||||
*/
|
||||
const handoffCursor = new WeakMap<Session, number>()
|
||||
|
||||
/**
|
||||
* Install the telemetry capture side onto a context for one backend.
|
||||
*
|
||||
* 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`). A `session/disposed` emits the session's `shutdown`
|
||||
* operational record — the marker rides the session's own termination edge,
|
||||
* where receivers key crash detection — and retires it from the adopted set,
|
||||
* so a long-lived backend neither retains closed sessions (and their frozen
|
||||
* event logs) nor re-marks them at unload. Disposal marks the sessions still
|
||||
* alive at teardown (their own edge would fire unobserved) 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 and still live, for double-adoption
|
||||
* protection and the teardown sweep of unmarked sessions;
|
||||
* `session/disposed` marks and 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>>()
|
||||
|
||||
/**
|
||||
* @param ctx - the composing backend's context; listeners bind to its fiber.
|
||||
* @param backend - the backend receiving records; owned elsewhere, never disposed here beyond `shutdown()` forwarding.
|
||||
*/
|
||||
constructor(
|
||||
private readonly ctx: Context,
|
||||
private readonly backend: TelemetryBackend,
|
||||
) {
|
||||
ctx.on('session/created', (session) => {
|
||||
this.adopt(session)
|
||||
})
|
||||
// The session's own termination edge: emit the shutdown marker HERE —
|
||||
// receivers classify a session with activity and no marker as crashed,
|
||||
// so a normally closed session in a long-running host must get its
|
||||
// marker at disposal, not never. Then retire: the projection/cursor
|
||||
// WeakMaps die with the Session object; only the strong adopted set
|
||||
// needs the explicit release.
|
||||
ctx.on('session/disposed', (session) => {
|
||||
this.contain(() => {
|
||||
if (!this.adopted.delete(session)) return
|
||||
this.handOff(shutdownRecord(session))
|
||||
})
|
||||
})
|
||||
ctx.on('session/event', (session, event) => {
|
||||
this.contain(() => {
|
||||
this.capture(session, event)
|
||||
})
|
||||
})
|
||||
// Parallel listeners are awaited by the loop at turn end; returning void
|
||||
// (not the SDK's flush promise) is the turn-latency contract.
|
||||
ctx.on('session/flush', (session) => {
|
||||
this.contain(() => {
|
||||
this.hintFlush(session)
|
||||
})
|
||||
})
|
||||
ctx.on('agent/error', (agent, turn, step, error) => {
|
||||
this.contain(() => {
|
||||
this.relayAgentError(agent, turn, step, error)
|
||||
})
|
||||
})
|
||||
ctx.effect(() => async () => {
|
||||
// Sessions still adopted here are alive through a whole-application
|
||||
// teardown (their own disposal edge will fire after telemetry is gone,
|
||||
// unobserved) — mark them now so the receiver sees a clean stop of
|
||||
// observation rather than a crash-shaped silence.
|
||||
for (const session of this.adopted) {
|
||||
this.contain(() => {
|
||||
this.handOff(shutdownRecord(session))
|
||||
})
|
||||
}
|
||||
try {
|
||||
await this.backend.shutdown()
|
||||
} catch (error) {
|
||||
this.ctx.logger.warn(`telemetry: backend shutdown failed: ${String(error)}`)
|
||||
}
|
||||
}, 'telemetry capture')
|
||||
for (const session of ctx.sessions.list()) {
|
||||
this.adopt(session)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Adopt a session: replay its log THROUGH the projection from the handoff
|
||||
* cursor, then rely on the firehose for everything after. When no cursor
|
||||
* survived, replay starts at the session's construction boundary
|
||||
* (`firstLiveSeq`), not seq 0: constructor seeds never publish on the
|
||||
* firehose, and their content already left the process under another
|
||||
* identity — the same id in a previous process (resume) or the parent's
|
||||
* stream (fork, stitched by receivers via `session.seed_length`). Events
|
||||
* at or below the start still feed the projection state (first-chunk
|
||||
* tracking) without being re-handed, so a resumed fiber drops mid-step
|
||||
* chunk continuations exactly like the fiber that saw the step begin. The
|
||||
* cost, accepted with the seam's at-most-once stance: a resume no longer
|
||||
* backfills records a previous process failed to deliver.
|
||||
* @param session - the live session to adopt; a second adoption is a no-op.
|
||||
*/
|
||||
private adopt(session: Session): void {
|
||||
if (this.adopted.has(session)) return
|
||||
this.adopted.add(session)
|
||||
const cursor = handoffCursor.get(session) ?? session.firstLiveSeq - 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. */
|
||||
private track(session: Session, event: SessionEvent): void {
|
||||
if (event.type === 'assistant/chunk') {
|
||||
this.seen(session).add(`${event.data.turn}:${event.data.step}`)
|
||||
}
|
||||
}
|
||||
|
||||
/** Project one event and hand it to the backend, advancing the cursor on handoff. */
|
||||
private capture(session: Session, event: SessionEvent): void {
|
||||
if (event.type === 'assistant/chunk') {
|
||||
const key = `${event.data.turn}:${event.data.step}`
|
||||
const seen = this.seen(session)
|
||||
// Fixed chunk projection: only the first chunk of each (turn, step)
|
||||
// ships — the stream-started signal; content is byte-complete in the
|
||||
// step's assembled assistant/message. Dropped chunks do not advance
|
||||
// the cursor, so re-adoption re-drops them deterministically.
|
||||
if (seen.has(key)) return
|
||||
seen.add(key)
|
||||
}
|
||||
this.handOff({
|
||||
channel: 'ledger',
|
||||
time: event.time,
|
||||
severity: severityOf(event),
|
||||
attributes: identityOf(session, event),
|
||||
// The live event object is mutable and the backend serializes later;
|
||||
// append-time validation guarantees this clone cannot throw.
|
||||
body: structuredClone(event.data),
|
||||
})
|
||||
handoffCursor.set(session, event.seq)
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the `telemetry/record` waterfall over one record and hand the result
|
||||
* to the backend. The innermost `next` passes the record through unchanged
|
||||
* — the seam ships no rules; exported data is as clean as the listeners a
|
||||
* deployment mounts. Callers run inside {@link contain}, so a throwing
|
||||
* rule withholds the record instead of reaching the loop (fail-closed).
|
||||
*/
|
||||
private handOff(record: TelemetryRecord): void {
|
||||
this.backend.emit(this.ctx.waterfall('telemetry/record', record, () => record))
|
||||
}
|
||||
|
||||
/** Forward the turn-end boundary to the backend's optional flush hint. */
|
||||
private hintFlush(session: Session): void {
|
||||
if (this.adopted.has(session)) this.backend.flush?.()
|
||||
}
|
||||
|
||||
/** Relay one `agent/error` bus emission as an `agent-error` operational record. */
|
||||
private relayAgentError(agent: Agent, turn: number, step: number, error: Error): void {
|
||||
this.handOff({
|
||||
channel: 'ops',
|
||||
time: Date.now(),
|
||||
severity: 'error',
|
||||
attributes: {
|
||||
'telemetry.op': 'agent-error',
|
||||
'session.id': String(agent.session.id),
|
||||
'agent.id': agent.id,
|
||||
'error.name': error.name,
|
||||
turn,
|
||||
step,
|
||||
},
|
||||
body: { name: error.name, message: error.message },
|
||||
})
|
||||
}
|
||||
|
||||
/** Lazily create the per-session first-chunk tracking set. */
|
||||
private seen(session: Session): Set<string> {
|
||||
let set = this.chunkSeen.get(session)
|
||||
if (!set) this.chunkSeen.set(session, set = new Set())
|
||||
return set
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one capture-side step with its exception contained: cordis `emit`
|
||||
* is stop-on-throw, so a throwing listener would starve every subscriber
|
||||
* registered after this plugin — nothing from the backend may escape.
|
||||
*/
|
||||
private contain(step: () => void): void {
|
||||
try {
|
||||
step()
|
||||
} catch (error) {
|
||||
this.ctx.logger.warn(`telemetry: capture step failed: ${String(error)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the per-session clean-exit marker: emitted at the session's own
|
||||
* disposal edge, or at coordinator dispose for sessions still alive then.
|
||||
*/
|
||||
function shutdownRecord(session: Session): TelemetryRecord {
|
||||
return {
|
||||
channel: 'ops',
|
||||
time: Date.now(),
|
||||
severity: 'info',
|
||||
attributes: { 'telemetry.op': 'shutdown', 'session.id': String(session.id) },
|
||||
body: { op: 'shutdown' },
|
||||
}
|
||||
}
|
||||
|
||||
/** Map an event's own outcome flag to the pre-baked alerting severity. */
|
||||
function severityOf(event: SessionEvent): TelemetrySeverity {
|
||||
switch (event.type) {
|
||||
case 'tool/result':
|
||||
return event.data.isError ? 'error' : 'info'
|
||||
case 'turn/end':
|
||||
return event.data.reason.kind === 'error' ? 'error' : 'info'
|
||||
case 'prompt/blocked':
|
||||
return 'warn'
|
||||
default:
|
||||
// Merge-extensible fall-through (no assertNever): event types this seam
|
||||
// does not depend on — including plugin-merged ones it never heard of —
|
||||
// pass through as info; their owners' outcome semantics stay theirs.
|
||||
return 'info'
|
||||
}
|
||||
}
|
||||
|
||||
/** Build the minimal identity attributes: envelope plus self-contained header facts. */
|
||||
function identityOf(session: Session, event: SessionEvent): Record<string, string | number> {
|
||||
const attributes: Record<string, string | number> = {
|
||||
'session.id': String(session.id),
|
||||
'event.type': event.type,
|
||||
'event.seq': event.seq,
|
||||
}
|
||||
const { cwd, parentSession, seedLength } = session.header
|
||||
if (cwd !== undefined) attributes['session.cwd'] = cwd
|
||||
if (parentSession !== undefined) attributes['session.parent_id'] = String(parentSession)
|
||||
// The durable fork boundary: a forked stream starts here, and its prefix
|
||||
// lives in the parent's stream — receivers stitch on (parent_id, seed_length).
|
||||
if (seedLength !== undefined) attributes['session.seed_length'] = seedLength
|
||||
return attributes
|
||||
}
|
||||
156
packages/telemetry/session-telemetry/src/index.ts
Normal file
156
packages/telemetry/session-telemetry/src/index.ts
Normal file
@@ -0,0 +1,156 @@
|
||||
/**
|
||||
* Telemetry seam for the DeepSeek Harness.
|
||||
*
|
||||
* The seam owns the CAPTURE side of session-event reporting — which records
|
||||
* exist (the chunk projection), what they carry (the logical record), when
|
||||
* they are handed over (adoption, the per-append firehose, lifecycle
|
||||
* forwarding), and the HMR handoff cursor. Everything downstream of
|
||||
* {@link Telemetry.emit} — batching, retry, queueing, loss policy — is the
|
||||
* reporting SDK's territory and is deliberately not modelled here. The
|
||||
* design and its trade-offs are pinned in
|
||||
* .agents/notes/implemented/feature/2026-07-23-session-telemetry-otel-revival.md.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-session-telemetry
|
||||
*/
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
|
||||
declare module 'cordis' {
|
||||
interface Context {
|
||||
telemetry: Telemetry
|
||||
}
|
||||
|
||||
interface Events {
|
||||
/**
|
||||
* Transform one outbound record before it reaches the backend. This
|
||||
* waterfall is the seam's redaction extension point. It ships NO rules
|
||||
* of its own: the
|
||||
* innermost `next()` passes the record through unchanged, and with no
|
||||
* listener mounted records reach the backend as captured, so exported
|
||||
* data is exactly as clean as the rules a deployment mounts. Listeners
|
||||
* stack by transforming `next()`'s return value; returning without
|
||||
* `next()` replaces everything beneath. Dispatched synchronously on the
|
||||
* capture hot path inside the coordinator's containment: a throwing
|
||||
* listener withholds that one record (fail-closed) and never reaches the
|
||||
* agent loop. Redaction applies to the exported copy only; the canonical
|
||||
* session log is never rewritten.
|
||||
* @param record - the candidate record, already the coordinator's own deep
|
||||
* copy; listeners return a (possibly new) record and must not mutate it.
|
||||
* @mode waterfall
|
||||
*/
|
||||
'telemetry/record'(record: TelemetryRecord, next: () => TelemetryRecord): TelemetryRecord
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Severity of a telemetry record, pre-mapped at capture so a receiver can
|
||||
* alert with zero configuration: `error` for events whose own outcome flag
|
||||
* says so (`tool/result.isError`, `turn/end` error reasons) and for
|
||||
* `agent-error` operational records, `warn` for `prompt/blocked`, `info`
|
||||
* for everything else — including event types merged by other packages,
|
||||
* whose outcome semantics stay with their owners.
|
||||
*/
|
||||
export type TelemetrySeverity = 'info' | 'warn' | 'error'
|
||||
|
||||
/**
|
||||
* One logical record handed to a backend — the seam's whole outbound
|
||||
* vocabulary. Ledger records mirror session-log events one-to-one;
|
||||
* operational records (`channel: 'ops'`) carry the two signals with no log
|
||||
* home (`agent-error`, `shutdown`) and deliberately omit `event.seq`-style
|
||||
* identity so they can never be mistaken for ledger rows.
|
||||
*/
|
||||
export interface TelemetryRecord {
|
||||
/** Ledger (session-log mirror) or ops (operational signal) channel; backends keep the two under separate instrumentation scopes. */
|
||||
channel: 'ledger' | 'ops'
|
||||
/** Unix epoch milliseconds — the source event's append time for ledger records, the emission time for ops records. */
|
||||
time: number
|
||||
/** Pre-mapped alerting severity; see {@link TelemetrySeverity}. */
|
||||
severity: TelemetrySeverity
|
||||
/**
|
||||
* Identity attributes, deliberately minimal: ledger records carry
|
||||
* `session.id`, `event.type`, `event.seq`, plus `session.cwd` /
|
||||
* `session.parent_id` when the header has them; ops records carry
|
||||
* `telemetry.op`, `session.id`, and (for `agent-error`) `agent.id`,
|
||||
* `turn`, `step`, `error.name`. Anything recoverable from the body is
|
||||
* intentionally NOT duplicated here.
|
||||
*/
|
||||
attributes: Record<string, string | number>
|
||||
/**
|
||||
* The complete payload: a deep copy of the session event's `data` for
|
||||
* ledger records (JSON-serializable by `Session.append`'s own
|
||||
* validation), or the op payload for ops records. Never mutated after
|
||||
* handoff.
|
||||
*/
|
||||
body: unknown
|
||||
}
|
||||
|
||||
/**
|
||||
* The backend contract the coordinator hands records to — the minimum any
|
||||
* reporting SDK satisfies with zero bending. {@link Telemetry} is its
|
||||
* service-registered form; tests compose the coordinator with a bare
|
||||
* implementation of this interface.
|
||||
*/
|
||||
export interface TelemetryBackend {
|
||||
/**
|
||||
* Hand one record to the backend's pipeline. MUST be a non-blocking
|
||||
* enqueue — the coordinator calls this synchronously from the
|
||||
* `session/event` hot path, so anything slower than a queue push would tax
|
||||
* the agent loop. Errors thrown here are contained by the coordinator and
|
||||
* logged; they never reach the loop.
|
||||
* @param record - the logical record to report; owned by the backend after the call.
|
||||
*/
|
||||
emit(record: TelemetryRecord): void
|
||||
/**
|
||||
* Optional hint that a natural boundary (turn end) passed — a backend may
|
||||
* forward it to its SDK's flush so records land at turn boundaries. Called
|
||||
* fire-and-forget; implementations must not block and must not throw
|
||||
* meaningfully (the coordinator contains exceptions). Most backends should
|
||||
* leave this unimplemented and let their SDK's own batching cadence govern
|
||||
* export timing: a backend that does implement it owns the interaction
|
||||
* between its concurrent flushes and {@link shutdown}'s drain (the OTel
|
||||
* backend removed its implementation for exactly that hazard — see the
|
||||
* revival Agent Note).
|
||||
*/
|
||||
flush?(): void
|
||||
/**
|
||||
* Forward the fiber's disposal to the SDK: flush whatever is queued and
|
||||
* 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>
|
||||
}
|
||||
|
||||
/**
|
||||
* The backend contract in its loadable form: one implementation per context —
|
||||
* the cordis `Service` registration under the `telemetry` key throws on a
|
||||
* duplicate, cordis' standard behavior. A backend composes a
|
||||
* {@link TelemetryCoordinator} in its constructor to install the capture side.
|
||||
*/
|
||||
export abstract class Telemetry extends Service implements TelemetryBackend {
|
||||
constructor(ctx: Context) {
|
||||
super(ctx, 'telemetry')
|
||||
}
|
||||
|
||||
/**
|
||||
* See {@link TelemetryBackend.emit} — the seam declaration is the contract's one home.
|
||||
* @param record - the logical record to report; owned by the backend after the call.
|
||||
*/
|
||||
abstract emit(record: TelemetryRecord): void
|
||||
|
||||
/** See {@link TelemetryBackend.flush}. */
|
||||
flush?(): void
|
||||
|
||||
/**
|
||||
* See {@link TelemetryBackend.shutdown}.
|
||||
* @returns resolves when the backend's pipeline has quiesced.
|
||||
*/
|
||||
abstract shutdown(): Promise<void>
|
||||
}
|
||||
|
||||
export { TelemetryCoordinator } from './coordinator.ts'
|
||||
32
packages/telemetry/session-telemetry/src/invariant.ts
Normal file
32
packages/telemetry/session-telemetry/src/invariant.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
/**
|
||||
* Package-owned invariant companion for `@deepseek-ai/dsh-session-telemetry`.
|
||||
* @module @deepseek-ai/dsh-session-telemetry/invariant
|
||||
*/
|
||||
|
||||
/* jscpd:ignore-start */
|
||||
import type { Context } from 'cordis'
|
||||
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
|
||||
|
||||
const PACKAGE_NAME = '@deepseek-ai/dsh-session-telemetry'
|
||||
|
||||
/** Cordis companion plugin name. */
|
||||
export const name = 'session-telemetry-invariant'
|
||||
/** Service required before the companion can reserve package ownership. */
|
||||
export const inject = ['invariants']
|
||||
|
||||
/**
|
||||
* No runtime invariant: the seam's whole output is the backend handoff — a
|
||||
* synchronous `emit()` call outside every authoritative event stream — and its
|
||||
* capture side never appends session events, so no event/data relation exists
|
||||
* for an independent companion to observe.
|
||||
*/
|
||||
const install: InvariantInstaller = () => {}
|
||||
|
||||
/**
|
||||
* Register this package's invariant companion.
|
||||
* @param ctx - Cordis context carrying the invariant service.
|
||||
* @returns the installed registration's disposer after setup succeeds.
|
||||
*/
|
||||
export const apply = (ctx: Context): Promise<() => void> =>
|
||||
Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
|
||||
/* jscpd:ignore-end */
|
||||
116
packages/telemetry/session-telemetry/tests/redact.spec.ts
Normal file
116
packages/telemetry/session-telemetry/tests/redact.spec.ts
Normal file
@@ -0,0 +1,116 @@
|
||||
/**
|
||||
* The `telemetry/record` waterfall contract: pass-through when no listener is
|
||||
* mounted, listener stacking and replacement, ops-record coverage, the
|
||||
* untouched canonical log, and the fail-closed containment of a throwing rule.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
TelemetryCoordinator,
|
||||
type TelemetryBackend,
|
||||
type TelemetryRecord,
|
||||
} from '../src/index.ts'
|
||||
|
||||
const FIXTURE_SECRET = 'sk-fixture1234567890'
|
||||
|
||||
class CollectingBackend implements TelemetryBackend {
|
||||
records: TelemetryRecord[] = []
|
||||
emit(record: TelemetryRecord): void {
|
||||
this.records.push(record)
|
||||
}
|
||||
async shutdown(): Promise<void> {}
|
||||
}
|
||||
|
||||
async function setup() {
|
||||
const backend = new CollectingBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
return { ctx, backend, fiber }
|
||||
}
|
||||
|
||||
describe('telemetry/record waterfall', () => {
|
||||
it('passes records through unchanged when no listener is mounted', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const session = ctx.sessions.create(SessionId('w'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: `key ${FIXTURE_SECRET}` }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const body = backend.records[0]!.body as { content: { text: string }[] }
|
||||
expect(body.content[0]!.text).toBe(`key ${FIXTURE_SECRET}`)
|
||||
})
|
||||
|
||||
it('applies a mounted rule to every outbound record, ops records included', async () => {
|
||||
const { ctx, backend, fiber } = await setup()
|
||||
ctx.on('telemetry/record', (_record, next) => {
|
||||
const record = next()
|
||||
return { ...record, body: { scrubbed: true } }
|
||||
})
|
||||
const session = ctx.sessions.create(SessionId('rule'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: FIXTURE_SECRET }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
expect(backend.records[0]!.body).toEqual({ scrubbed: true })
|
||||
// The dispose-time shutdown ops record passes through the same waterfall.
|
||||
await fiber.dispose()
|
||||
const ops = backend.records.filter(record => record.channel === 'ops')
|
||||
expect(ops).toHaveLength(1)
|
||||
expect(ops[0]!.body).toEqual({ scrubbed: true })
|
||||
})
|
||||
|
||||
it('keeps the canonical log untouched by a mounted rule', async () => {
|
||||
const { ctx } = await setup()
|
||||
ctx.on('telemetry/record', (_record, next) => ({ ...next(), body: null }))
|
||||
const session = ctx.sessions.create(SessionId('log'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: FIXTURE_SECRET }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const logged = session.events[0]!.data as { content: { text: string }[] }
|
||||
expect(logged.content[0]!.text).toBe(FIXTURE_SECRET)
|
||||
})
|
||||
|
||||
it('stacks listeners outermost-first around next()', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const order: string[] = []
|
||||
ctx.on('telemetry/record', (_record, next) => {
|
||||
order.push('outer-before')
|
||||
const record = next()
|
||||
order.push('outer-after')
|
||||
return { ...record, attributes: { ...record.attributes, outer: 1 } }
|
||||
})
|
||||
ctx.on('telemetry/record', (_record, next) => {
|
||||
order.push('inner')
|
||||
const record = next()
|
||||
return { ...record, attributes: { ...record.attributes, inner: 1 } }
|
||||
})
|
||||
const session = ctx.sessions.create(SessionId('stack'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
expect(order).toEqual(['outer-before', 'inner', 'outer-after'])
|
||||
expect(backend.records[0]!.attributes).toMatchObject({ outer: 1, inner: 1 })
|
||||
})
|
||||
|
||||
it('a listener that skips next() replaces everything beneath it', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const inner = { called: false }
|
||||
ctx.on('telemetry/record', () => ({ channel: 'ops', time: 0, severity: 'info', attributes: {}, body: 'replaced' } satisfies TelemetryRecord))
|
||||
ctx.on('telemetry/record', (_record, next) => {
|
||||
inner.called = true
|
||||
return next()
|
||||
})
|
||||
const session = ctx.sessions.create(SessionId('veto'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
expect(backend.records[0]!.body).toBe('replaced')
|
||||
expect(inner.called).toBe(false)
|
||||
})
|
||||
|
||||
it('a throwing rule withholds the record fail-closed without disturbing the log', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
ctx.on('telemetry/record', () => {
|
||||
throw new Error('rule exploded')
|
||||
})
|
||||
const session = ctx.sessions.create(SessionId('closed'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
expect(backend.records).toHaveLength(0)
|
||||
expect(session.events).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
424
packages/telemetry/session-telemetry/tests/telemetry.spec.ts
Normal file
424
packages/telemetry/session-telemetry/tests/telemetry.spec.ts
Normal file
@@ -0,0 +1,424 @@
|
||||
/**
|
||||
* Coordinator semantics against a bare fake backend — the RFC's named unit
|
||||
* tier for the seam: adoption (fresh, seeded, re-adoption via the handoff
|
||||
* cursor), the fixed chunk projection, deep-copy isolation, turn-latency and
|
||||
* dispose-ordering pins, failure containment, and the `agent/error` relay.
|
||||
*/
|
||||
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import SessionStore, { SessionId, type Session, type SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { TelemetryCoordinator, type TelemetryBackend, type TelemetryRecord } from '../src/index.ts'
|
||||
|
||||
declare module '@deepseek-ai/dsh-session' {
|
||||
interface SessionEventMap {
|
||||
/**
|
||||
* Test-only merged event proving unknown types flow through unchanged.
|
||||
* @mode emit
|
||||
* @param payload - opaque test payload
|
||||
*/
|
||||
'telemetry-test/opaque': { payload: { nested: string[] } }
|
||||
}
|
||||
}
|
||||
|
||||
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'])}`)
|
||||
}
|
||||
|
||||
flush = vi.fn()
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
this.calls.push('shutdown')
|
||||
await new Promise(resolve => setTimeout(resolve, 5))
|
||||
if (this.shutdownError) throw this.shutdownError
|
||||
this.shutdownResolved = true
|
||||
}
|
||||
|
||||
ledger(): TelemetryRecord[] {
|
||||
return this.records.filter(r => r.channel === 'ledger')
|
||||
}
|
||||
}
|
||||
|
||||
async function setup(backend: FakeBackend = new FakeBackend()) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
return { ctx, backend, fiber }
|
||||
}
|
||||
|
||||
function liveSession(ctx: Context, id = `s-${Math.random().toString(36).slice(2)}`): Session {
|
||||
return ctx.sessions.create(SessionId(id), { meta: {} })
|
||||
}
|
||||
|
||||
function appendTurn(session: Session): void {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'hello' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
}
|
||||
|
||||
describe('TelemetryCoordinator capture', () => {
|
||||
it('hands every appended event over with envelope identity and cloned body', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const session = liveSession(ctx, 'cap')
|
||||
appendTurn(session)
|
||||
|
||||
const start = backend.ledger()[0]!
|
||||
const message = backend.ledger()[1]!
|
||||
expect(start.attributes).toMatchObject({ 'session.id': 'cap', 'event.type': 'turn/start', 'event.seq': 0 })
|
||||
expect(start.time).toBe(session.events[0]!.time)
|
||||
expect(start.severity).toBe('info')
|
||||
expect(message.attributes['event.seq']).toBe(1)
|
||||
// Deep-copy isolation: mutating the handed-off body never reaches the log.
|
||||
;(message.body as { content: { text: string }[] }).content[0]!.text = 'tampered'
|
||||
const logged = session.events[1] as SessionEvent<'user/message'>
|
||||
expect(logged.data.content[0]).toMatchObject({ text: 'hello' })
|
||||
})
|
||||
|
||||
it('stamps header facts on every record when present', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const parent = SessionId('parent')
|
||||
const session = ctx.sessions.create(SessionId('child'), { meta: { cwd: '/tmp/proj', parentSession: parent } })
|
||||
appendTurn(session)
|
||||
for (const record of backend.ledger()) {
|
||||
expect(record.attributes['session.cwd']).toBe('/tmp/proj')
|
||||
expect(record.attributes['session.parent_id']).toBe('parent')
|
||||
}
|
||||
})
|
||||
|
||||
it('maps outcome flags to severity, unknown types falling through as info', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const session = liveSession(ctx)
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('tool/result', { turn: 1, step: 1, callId: 'c1' as never, content: [], isError: true }, { surfaceOp: 'append' })
|
||||
session.append('tool/result', { turn: 1, step: 1, callId: 'c2' as never, content: [], isError: false }, { surfaceOp: 'append' })
|
||||
session.append('prompt/blocked', { content: [], source: { kind: 'user' }, reason: 'vetoed' })
|
||||
session.append('telemetry-test/opaque', { payload: { nested: [] } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, message: 'boom' } })
|
||||
const severities = backend.ledger().map(r => [r.attributes['event.type'], r.severity])
|
||||
expect(severities).toEqual([
|
||||
['turn/start', 'info'],
|
||||
['tool/result', 'error'],
|
||||
['tool/result', 'info'],
|
||||
['prompt/blocked', 'warn'],
|
||||
['telemetry-test/opaque', 'info'],
|
||||
['turn/end', 'error'],
|
||||
])
|
||||
})
|
||||
|
||||
it('passes unknown merged event types through unchanged', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const session = liveSession(ctx)
|
||||
session.append('telemetry-test/opaque', { payload: { nested: ['a', 'b'] } })
|
||||
const record = backend.ledger()[0]!
|
||||
expect(record.attributes['event.type']).toBe('telemetry-test/opaque')
|
||||
expect(record.severity).toBe('info')
|
||||
expect(record.body).toEqual({ payload: { nested: ['a', 'b'] } })
|
||||
})
|
||||
|
||||
it('ships only the first chunk of each (turn, step), per session', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const a = liveSession(ctx, 'a')
|
||||
const b = liveSession(ctx, 'b')
|
||||
const chunk = (s: Session, turn: number, step: number, text: string) =>
|
||||
s.append('assistant/chunk', { turn, step, chunk: { type: 'text-delta', index: 0, text } })
|
||||
chunk(a, 1, 1, 'a11-first')
|
||||
chunk(a, 1, 1, 'a11-second')
|
||||
chunk(a, 1, 2, 'a12-first')
|
||||
chunk(b, 1, 1, 'b11-first')
|
||||
chunk(b, 1, 1, 'b11-second')
|
||||
const shipped = backend.ledger().map(r => [r.attributes['session.id'], (r.body as { chunk: { text: string } }).chunk.text])
|
||||
expect(shipped).toEqual([
|
||||
['a', 'a11-first'],
|
||||
['a', 'a12-first'],
|
||||
['b', 'b11-first'],
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
describe('TelemetryCoordinator adoption', () => {
|
||||
it('starts export at the construction boundary: seeded history never re-exports', async () => {
|
||||
const backend = new FakeBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const parent = liveSession(ctx, 'seed-parent')
|
||||
appendTurn(parent)
|
||||
const child = ctx.sessions.create(SessionId('seeded'), { seed: [...parent.events], meta: {} })
|
||||
await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
// The live parent (no constructor seed) replays in full; the child's
|
||||
// inherited prefix already left the process under another identity (the
|
||||
// parent's id here; the same id in a previous process for a resume) and
|
||||
// must not be re-exported — only its live suffix ships.
|
||||
const seqs = backend.ledger().map(r => [r.attributes['session.id'], r.attributes['event.seq']])
|
||||
expect(seqs).toEqual(expect.arrayContaining([['seed-parent', 0], ['seed-parent', 1]]))
|
||||
expect(seqs.filter(([id]) => id === 'seeded')).toEqual([])
|
||||
child.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
expect(backend.ledger().map(r => [r.attributes['session.id'], r.attributes['event.seq']]))
|
||||
.toEqual(expect.arrayContaining([['seeded', 2]]))
|
||||
})
|
||||
|
||||
it('resume shape: a full-log seed exports nothing yet still rebuilds the chunk projection', async () => {
|
||||
const backend = new FakeBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const donor = ctx.sessions.create(SessionId('donor'), { meta: {} })
|
||||
donor.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
donor.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'first' } })
|
||||
const resumed = ctx.sessions.create(SessionId('resumed'), { seed: [...donor.events], meta: {} })
|
||||
await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
const ofResumed = () => backend.ledger()
|
||||
.filter(r => r.attributes['session.id'] === 'resumed')
|
||||
.map(r => r.attributes['event.seq'])
|
||||
expect(ofResumed()).toEqual([])
|
||||
// The seed fed the projection: the (turn 1, step 1) first chunk already
|
||||
// shipped from the original process, so its continuation is re-dropped…
|
||||
resumed.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'continuation' } })
|
||||
expect(ofResumed()).toEqual([])
|
||||
// …while a new step's first chunk exports normally.
|
||||
resumed.append('assistant/chunk', { turn: 1, step: 2, chunk: { type: 'text-delta', index: 0, text: 'next step' } })
|
||||
expect(ofResumed()).toEqual([3])
|
||||
})
|
||||
|
||||
it('stamps session.seed_length from the header so receivers can stitch fork streams', async () => {
|
||||
const backend = new FakeBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const parent = liveSession(ctx, 'stitch-parent')
|
||||
appendTurn(parent)
|
||||
const child = ctx.sessions.create(SessionId('stitch-child'), {
|
||||
seed: [...parent.events],
|
||||
meta: { parentSession: SessionId('stitch-parent'), seedLength: 2 },
|
||||
})
|
||||
await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
child.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
const record = backend.ledger().find(r => r.attributes['session.id'] === 'stitch-child')!
|
||||
expect(record.attributes['session.parent_id']).toBe('stitch-parent')
|
||||
expect(record.attributes['session.seed_length']).toBe(2)
|
||||
})
|
||||
|
||||
it('adopts exactly once when created fires after the sweep', async () => {
|
||||
const backend = new FakeBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
// The enter/announce window: prepare+enter puts the session in the store
|
||||
// (visible to the constructor sweep) before `session/created` fires, so a
|
||||
// coordinator loaded inside that window sees the session twice — sweep
|
||||
// first, created second. The second adoption must be a no-op.
|
||||
const session = ctx.sessions.prepare(SessionId('overlap'))
|
||||
appendTurn(session)
|
||||
ctx.sessions.enter(session)
|
||||
await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
expect(backend.ledger()).toHaveLength(2)
|
||||
ctx.sessions.announce(session)
|
||||
expect(backend.ledger()).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('resumes from the handoff cursor across a reload, re-dropping mid-step chunks', async () => {
|
||||
const backend = new FakeBackend()
|
||||
const { ctx, fiber } = await setup(backend)
|
||||
const session = liveSession(ctx, 'hmr')
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'first' } })
|
||||
expect(backend.ledger()).toHaveLength(2)
|
||||
|
||||
await fiber.dispose()
|
||||
// The reload window: appends while no telemetry listener is registered.
|
||||
session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'mid-step continuation' } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
|
||||
const second = new FakeBackend()
|
||||
await ctx.plugin({
|
||||
name: 'fake-telemetry-2',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, second),
|
||||
})
|
||||
// Only the window events past the cursor are re-handed, and the mid-step
|
||||
// continuation is re-dropped because ≤cursor events rebuilt the projection.
|
||||
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()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = liveSession(ctx, 'fresh')
|
||||
appendTurn(session)
|
||||
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, 1])
|
||||
})
|
||||
})
|
||||
|
||||
describe('TelemetryCoordinator lifecycle and containment', () => {
|
||||
it('forwards session/flush as a hint without awaiting backend work', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const session = liveSession(ctx)
|
||||
let settled = false
|
||||
backend.flush.mockImplementation(() => {
|
||||
// The backend may kick off arbitrary async work; the loop's parallel must not wait for it.
|
||||
void new Promise(resolve => setTimeout(resolve, 50)).then(() => { settled = true })
|
||||
})
|
||||
await ctx.parallel('session/flush', session)
|
||||
expect(backend.flush).toHaveBeenCalledTimes(1)
|
||||
expect(settled).toBe(false)
|
||||
})
|
||||
|
||||
it('ignores flush hints for sessions it never adopted', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const stranger = ctx.sessions.prepare(SessionId('stranger'), { meta: {} })
|
||||
await ctx.parallel('session/flush', stranger)
|
||||
expect(backend.flush).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('emits no marker for a session whose announcement was vetoed before adoption', async () => {
|
||||
const backend = new FakeBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
// A listener registered BEFORE the coordinator vetoes publication: the
|
||||
// store still emits the paired `session/disposed` for rollback, but the
|
||||
// coordinator never saw `session/created` — a marker for a session the
|
||||
// receiver saw no activity from would be noise, not signal.
|
||||
ctx.on('session/created', () => {
|
||||
throw new Error('vetoed by an earlier listener')
|
||||
})
|
||||
await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
expect(() => ctx.sessions.create(SessionId('vetoed'), { meta: {} })).toThrow('vetoed')
|
||||
expect(backend.records.filter(r => r.channel === 'ops')).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('emits each adopted session’s shutdown record before awaiting backend shutdown', async () => {
|
||||
const { ctx, backend, fiber } = await setup()
|
||||
liveSession(ctx, 's1')
|
||||
liveSession(ctx, 's2')
|
||||
await fiber.dispose()
|
||||
expect(backend.calls).toEqual(['emit:shutdown', 'emit:shutdown', 'shutdown'])
|
||||
expect(backend.shutdownResolved).toBe(true)
|
||||
const ops = backend.records.filter(r => r.channel === 'ops')
|
||||
expect(ops.map(r => r.attributes['session.id']).sort()).toEqual(['s1', 's2'])
|
||||
expect(ops.every(r => r.attributes['telemetry.op'] === 'shutdown' && r.severity === 'info')).toBe(true)
|
||||
expect(ops.every(r => !('event.seq' in r.attributes) && !('event.type' in r.attributes))).toBe(true)
|
||||
})
|
||||
|
||||
it('emits the shutdown marker at the session’s own disposal edge, then retires it', 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 termination
|
||||
// edge. The marker must ride THAT edge (receivers classify a session with
|
||||
// activity and no marker as crashed, so a normally closed session in a
|
||||
// long-running host must not look like a crash), and the session retires
|
||||
// from the adopted set so unload neither retains it nor re-marks it.
|
||||
const owner = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
inner.sessions.create(SessionId('ephemeral'), { meta: {} })
|
||||
}, { inject: ['sessions'] }))
|
||||
await owner.dispose()
|
||||
const atEdge = backend.records.filter(r => r.channel === 'ops')
|
||||
expect(atEdge.map(r => r.attributes['session.id'])).toEqual(['ephemeral'])
|
||||
expect(atEdge[0]!.attributes['telemetry.op']).toBe('shutdown')
|
||||
await fiber.dispose()
|
||||
const ops = backend.records.filter(r => r.channel === 'ops')
|
||||
expect(ops.map(r => r.attributes['session.id'])).toEqual(['ephemeral', 'survivor'])
|
||||
})
|
||||
|
||||
it('warns instead of throwing when backend shutdown fails', async () => {
|
||||
const backend = new FakeBackend()
|
||||
backend.shutdownError = new Error('exporter unreachable')
|
||||
const { ctx, fiber } = await setup(backend)
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
liveSession(ctx)
|
||||
await expect(fiber.dispose()).resolves.not.toThrow()
|
||||
expect(warn.mock.calls.some(args => String(args[0]).includes('shutdown failed'))).toBe(true)
|
||||
})
|
||||
|
||||
it('contains emit failures: the append succeeds and capture heals', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const session = liveSession(ctx)
|
||||
backend.emitError = new Error('backend broke')
|
||||
expect(() => session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })).not.toThrow()
|
||||
expect(warn).toHaveBeenCalled()
|
||||
backend.emitError = undefined
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
expect(backend.ledger().map(r => r.attributes['event.type'])).toEqual(['turn/end'])
|
||||
})
|
||||
|
||||
it('relays agent/error as an ops record with identity and structured name', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
const session = liveSession(ctx, 'erring')
|
||||
// Only the members the relay reads; the full Agent surface is irrelevant here.
|
||||
const agent = { id: 'agent-1', session } as Agent
|
||||
ctx.emit('agent/error', agent, 3, 2, new TypeError('adapter exploded'))
|
||||
const record = backend.records.find(r => r.channel === 'ops')!
|
||||
expect(record.severity).toBe('error')
|
||||
expect(record.attributes).toMatchObject({
|
||||
'telemetry.op': 'agent-error',
|
||||
'session.id': 'erring',
|
||||
'agent.id': 'agent-1',
|
||||
'error.name': 'TypeError',
|
||||
turn: 3,
|
||||
step: 2,
|
||||
})
|
||||
expect(record.body).toEqual({ name: 'TypeError', message: 'adapter exploded' })
|
||||
})
|
||||
})
|
||||
27
packages/telemetry/session-telemetry/tsconfig.json
Normal file
27
packages/telemetry/session-telemetry/tsconfig.json
Normal file
@@ -0,0 +1,27 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib/types"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../../core/agent"
|
||||
},
|
||||
{
|
||||
"path": "../../support/invariants"
|
||||
}
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user