refactor(telemetry): ship the redact waterfall without built-in rules
The seam keeps the telemetry/redact scrubbing interface but ships no rules of its own: the innermost next() passes records through unchanged, and deployments mount their rules as waterfall listeners. As an SDK we cannot know which patterns are secrets in a given deployment; a shipped list invites false confidence while catching only known shapes, and false positives would corrupt exported bodies. Mechanism stays with the seam, policy moves to the deployment; both READMEs and the Agent Note state the raw-export default plainly. The loader-composition e2e now mounts a deployment-style rule fixture and pins the same wire behavior: secret absent, placeholder present, canonical log untouched.
This commit is contained in:
kingwl
@@ -12,7 +12,7 @@ The coordinator registers, all through the composing fiber's effects: `session/c
|
||||
|
||||
## The redact waterfall
|
||||
|
||||
Every record passes the `telemetry/redact` waterfall between projection and `emit()` — nothing reaches a backend unredacted. The innermost `next()` applies the built-in conservative rule set (`applyDefaultRedaction`: credential shapes — API keys, GitHub/Slack tokens, AWS/Google keys, JWTs, PEM blocks, URL userinfo — replaced with `[REDACTED]` in body strings and string attribute values). Listeners stack stricter rules by transforming `next()`'s return value; returning without `next()` replaces the default rule set, and a throwing listener withholds that one record fail-closed inside the coordinator's containment. The built-in pattern list is a security invariant, deliberately not configurable from cordis.yml. Redaction applies to the exported copy only; the canonical session log is never rewritten.
|
||||
Every record passes the `telemetry/redact` 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
|
||||
|
||||
@@ -37,4 +37,4 @@ 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).
|
||||
- **Redaction is shape-based** — the default rules catch known credential shapes, not every secret; a deployment with stricter needs stacks `telemetry/redact` listeners, and exported data is only as clean as the mounted rules.
|
||||
- **No built-in redaction rules** — with no `telemetry/redact` 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.
|
||||
|
||||
@@ -2,10 +2,11 @@
|
||||
* 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/redact` waterfall, and hands the redacted copy 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.
|
||||
* `telemetry/redact` waterfall (deployment-mounted 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
|
||||
*/
|
||||
@@ -14,7 +15,6 @@ 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'
|
||||
import { applyDefaultRedaction } from './redact.ts'
|
||||
|
||||
/**
|
||||
* The handoff cursor: per session, the highest `seq` handed to a backend.
|
||||
@@ -145,13 +145,13 @@ export class TelemetryCoordinator {
|
||||
|
||||
/**
|
||||
* Run the `telemetry/redact` waterfall over one record and hand the result
|
||||
* to the backend. The innermost `next` applies the seam's conservative
|
||||
* default rules, so an unconfigured deployment still never exports raw
|
||||
* credential shapes; callers run inside {@link contain}, so a throwing
|
||||
* 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/redact', record, () => applyDefaultRedaction(record)))
|
||||
this.backend.emit(this.ctx.waterfall('telemetry/redact', record, () => record))
|
||||
}
|
||||
|
||||
/** Forward the turn-end boundary to the backend's optional flush hint. */
|
||||
|
||||
@@ -22,16 +22,17 @@ declare module 'cordis' {
|
||||
|
||||
interface Events {
|
||||
/**
|
||||
* Redact one outbound record before it reaches the backend. The innermost
|
||||
* `next()` applies the seam's conservative default rule set
|
||||
* (credential-shape scrubbing); listeners stack stricter rules by
|
||||
* transforming its return value, and returning without `next()` replaces
|
||||
* the default — the exported record is then only as clean as the
|
||||
* replacing rule. 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.
|
||||
* Redact one outbound record before it reaches the backend — the seam's
|
||||
* scrubbing extension point. The seam 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
|
||||
@@ -142,4 +143,3 @@ export abstract class Telemetry extends Service implements TelemetryBackend {
|
||||
}
|
||||
|
||||
export { TelemetryCoordinator } from './coordinator.ts'
|
||||
export { applyDefaultRedaction, REDACTION_PLACEHOLDER } from './redact.ts'
|
||||
|
||||
@@ -1,77 +0,0 @@
|
||||
/**
|
||||
* Conservative default redaction for outbound telemetry records.
|
||||
*
|
||||
* Session-event bodies carry file contents and command output that may embed
|
||||
* credentials; nothing may cross the seam to a backend unredacted. This module
|
||||
* is the innermost rule set of the `telemetry/redact` waterfall — always
|
||||
* applied unless an outer listener deliberately replaces the whole chain. It
|
||||
* scrubs credential-SHAPED substrings from every string in the record body,
|
||||
* leaving structure (keys, nesting, surrounding prose) intact. The pattern
|
||||
* list is a security invariant, deliberately not configurable; deployments
|
||||
* add stricter rules by stacking `telemetry/redact` listeners.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-session-telemetry/redact
|
||||
*/
|
||||
|
||||
import type { TelemetryRecord } from './index.ts'
|
||||
|
||||
/** Replacement text substituted for each detected credential-shaped span. */
|
||||
export const REDACTION_PLACEHOLDER = '[REDACTED]'
|
||||
|
||||
/**
|
||||
* Well-known credential shapes. A match anywhere inside a body string is
|
||||
* replaced; low-signal values (package names, versions, git SHAs, plain URLs)
|
||||
* deliberately stay untouched — they are the observability signal.
|
||||
*/
|
||||
const SECRET_PATTERNS: readonly RegExp[] = [
|
||||
/sk-(?:ant-)?[A-Za-z0-9_-]{10,}/g, // DeepSeek / OpenAI / Anthropic API keys
|
||||
/gh[pousr]_[A-Za-z0-9]{16,}/g, // GitHub personal/oauth/server/refresh tokens
|
||||
/github_pat_[A-Za-z0-9_]{20,}/g, // GitHub fine-grained PAT
|
||||
/xox[baprs]-[A-Za-z0-9-]{10,}/g, // Slack tokens
|
||||
/AKIA[0-9A-Z]{16}/g, // AWS access key id
|
||||
/AIza[0-9A-Za-z_-]{35}/g, // Google API key
|
||||
/eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}/g, // JWT
|
||||
/-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----/g, // PEM blocks
|
||||
/\b(?<scheme>[a-z][a-z0-9+.-]*):\/\/[^/\s:@]+:[^/\s:@]+@/g, // URL userinfo credentials
|
||||
]
|
||||
|
||||
/** Replace every known credential shape inside one string. */
|
||||
function scrub(text: string): string {
|
||||
let out = text
|
||||
for (const pattern of SECRET_PATTERNS) {
|
||||
out = out.replace(pattern, REDACTION_PLACEHOLDER)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
/**
|
||||
* Deep-scrub every string inside a lossless-JSON value, preserving structure.
|
||||
* The record body is the coordinator's own `structuredClone` — mutation-free
|
||||
* rebuilding keeps the exported copy independent of the canonical log either way.
|
||||
*/
|
||||
function scrubValue(value: unknown): unknown {
|
||||
if (typeof value === 'string') return scrub(value)
|
||||
if (Array.isArray(value)) return value.map(scrubValue)
|
||||
if (value !== null && typeof value === 'object') {
|
||||
const out: Record<string, unknown> = {}
|
||||
for (const [key, entry] of Object.entries(value)) out[key] = scrubValue(entry)
|
||||
return out
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply the conservative default rule set to one record — the innermost
|
||||
* `next` of the `telemetry/redact` waterfall. Attribute VALUES are scrubbed
|
||||
* alongside the body (identity attributes are seam-built and boring, but
|
||||
* `session.cwd` is caller-supplied); attribute keys are seam-owned constants.
|
||||
* @param record - the candidate record; not mutated.
|
||||
* @returns a redacted copy safe to hand to a backend.
|
||||
*/
|
||||
export function applyDefaultRedaction(record: TelemetryRecord): TelemetryRecord {
|
||||
const attributes: Record<string, string | number> = {}
|
||||
for (const [key, value] of Object.entries(record.attributes)) {
|
||||
attributes[key] = typeof value === 'string' ? scrub(value) : value
|
||||
}
|
||||
return { ...record, attributes, body: scrubValue(record.body) }
|
||||
}
|
||||
@@ -1,91 +1,19 @@
|
||||
/**
|
||||
* Default redaction rules and the `telemetry/redact` waterfall contract:
|
||||
* credential shapes scrubbed from bodies and attribute values, structure
|
||||
* preserved, canonical log untouched, listener stacking/replacement, and the
|
||||
* fail-closed containment of a throwing rule.
|
||||
* The `telemetry/redact` 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 {
|
||||
applyDefaultRedaction,
|
||||
REDACTION_PLACEHOLDER,
|
||||
TelemetryCoordinator,
|
||||
type TelemetryBackend,
|
||||
type TelemetryRecord,
|
||||
} from '../src/index.ts'
|
||||
|
||||
const SECRETS = {
|
||||
deepseek: 'sk-abcdef1234567890abcdef',
|
||||
anthropic: 'sk-ant-abcdef1234567890',
|
||||
githubPat: 'ghp_ABCDEFGHIJKLMNOPqrstuv12345678',
|
||||
finePat: 'github_pat_ABCDEFGHIJKLMNOPQRSTuvwx',
|
||||
slack: 'xoxb-1234567890-abcdefghij',
|
||||
aws: 'AKIAIOSFODNN7EXAMPLE',
|
||||
google: 'AIzaSyA-1234567890abcdefghijklmnopqrstu',
|
||||
jwt: 'eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.SflKxwRJSMeKKF2QT4fwpM',
|
||||
pem: '-----BEGIN RSA PRIVATE KEY-----\nMIIEow\n-----END RSA PRIVATE KEY-----',
|
||||
urlCreds: 'https://user:hunter2@internal.example.com/repo.git',
|
||||
} as const
|
||||
|
||||
function record(body: unknown, attributes: Record<string, string | number> = {}): TelemetryRecord {
|
||||
return { channel: 'ledger', time: 1, severity: 'info', attributes, body }
|
||||
}
|
||||
|
||||
describe('applyDefaultRedaction', () => {
|
||||
it('scrubs every known credential shape while preserving surrounding text', () => {
|
||||
for (const secret of Object.values(SECRETS)) {
|
||||
const out = applyDefaultRedaction(record(`before ${secret} after`))
|
||||
expect(out.body, secret).not.toContain(secret.includes('\n') ? 'MIIEow' : secret)
|
||||
expect(out.body).toContain('before ')
|
||||
expect(out.body).toContain(' after')
|
||||
expect(out.body).toContain(REDACTION_PLACEHOLDER)
|
||||
}
|
||||
})
|
||||
|
||||
it('scrubs URL userinfo credentials but leaves plain URLs alone', () => {
|
||||
const out = applyDefaultRedaction(record(`${SECRETS.urlCreds} and https://example.com/path`))
|
||||
expect(out.body).not.toContain('hunter2')
|
||||
expect(out.body).toContain('https://example.com/path')
|
||||
})
|
||||
|
||||
it('recurses through arrays and objects, preserving structure and non-strings', () => {
|
||||
const out = applyDefaultRedaction(record({
|
||||
list: [`key=${SECRETS.deepseek}`, 7, null, true],
|
||||
nested: { text: SECRETS.githubPat, count: 3 },
|
||||
}))
|
||||
expect(out.body).toEqual({
|
||||
list: [`key=${REDACTION_PLACEHOLDER}`, 7, null, true],
|
||||
nested: { text: REDACTION_PLACEHOLDER, count: 3 },
|
||||
})
|
||||
})
|
||||
|
||||
it('leaves low-signal values untouched', () => {
|
||||
const clean = {
|
||||
pkg: '@deepseek-ai/dsh-session-telemetry@0.0.1',
|
||||
sha: '342a4c3a9d3adf13cf4ad33b9f8d6e79170be5e2',
|
||||
prose: 'ordinary sentence with kebab-case-identifier',
|
||||
}
|
||||
expect(applyDefaultRedaction(record(clean)).body).toEqual(clean)
|
||||
})
|
||||
|
||||
it('scrubs string attribute values and keeps numeric ones', () => {
|
||||
const out = applyDefaultRedaction(record(null, {
|
||||
'session.cwd': `/home/${SECRETS.aws}/proj`,
|
||||
'event.seq': 4,
|
||||
}))
|
||||
expect(out.attributes['session.cwd']).toBe(`/home/${REDACTION_PLACEHOLDER}/proj`)
|
||||
expect(out.attributes['event.seq']).toBe(4)
|
||||
})
|
||||
|
||||
it('never mutates its input', () => {
|
||||
const input = record({ text: SECRETS.slack }, { 'session.cwd': SECRETS.aws })
|
||||
applyDefaultRedaction(input)
|
||||
expect((input.body as { text: string }).text).toBe(SECRETS.slack)
|
||||
expect(input.attributes['session.cwd']).toBe(SECRETS.aws)
|
||||
})
|
||||
})
|
||||
const FIXTURE_SECRET = 'sk-fixture1234567890'
|
||||
|
||||
class CollectingBackend implements TelemetryBackend {
|
||||
records: TelemetryRecord[] = []
|
||||
@@ -99,49 +27,80 @@ async function setup() {
|
||||
const backend = new CollectingBackend()
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin({
|
||||
const fiber = await ctx.plugin({
|
||||
name: 'fake-telemetry',
|
||||
inject: ['sessions'],
|
||||
apply: (inner: Context) => void new TelemetryCoordinator(inner, backend),
|
||||
})
|
||||
return { ctx, backend }
|
||||
return { ctx, backend, fiber }
|
||||
}
|
||||
|
||||
describe('telemetry/redact waterfall', () => {
|
||||
it('applies the default rules when no listener is registered', async () => {
|
||||
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 ${SECRETS.deepseek}` }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
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 ${REDACTION_PLACEHOLDER}`)
|
||||
expect(body.content[0]!.text).toBe(`key ${FIXTURE_SECRET}`)
|
||||
})
|
||||
|
||||
it('keeps the canonical log unredacted', async () => {
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create(SessionId('log'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: SECRETS.githubPat }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const logged = session.events[0]!.data as { content: { text: string }[] }
|
||||
expect(logged.content[0]!.text).toBe(SECRETS.githubPat)
|
||||
})
|
||||
|
||||
it('lets a listener stack a stricter rule on top of the defaults', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
it('applies a mounted rule to every outbound record, ops records included', async () => {
|
||||
const { ctx, backend, fiber } = await setup()
|
||||
ctx.on('telemetry/redact', (_record, next) => {
|
||||
const defaulted = next()
|
||||
return { ...defaulted, body: { shapeOnly: true } }
|
||||
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/redact', (_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/redact', (_record, next) => {
|
||||
order.push('outer-before')
|
||||
const record = next()
|
||||
order.push('outer-after')
|
||||
return { ...record, attributes: { ...record.attributes, outer: 1 } }
|
||||
})
|
||||
ctx.on('telemetry/redact', (_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: 'anything' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
expect(backend.records[0]!.body).toEqual({ shapeOnly: true })
|
||||
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 the default rules', async () => {
|
||||
it('a listener that skips next() replaces everything beneath it', async () => {
|
||||
const { ctx, backend } = await setup()
|
||||
ctx.on('telemetry/redact', record => record)
|
||||
const inner = { called: false }
|
||||
ctx.on('telemetry/redact', () => ({ channel: 'ops', time: 0, severity: 'info', attributes: {}, body: 'replaced' } satisfies TelemetryRecord))
|
||||
ctx.on('telemetry/redact', (_record, next) => {
|
||||
inner.called = true
|
||||
return next()
|
||||
})
|
||||
const session = ctx.sessions.create(SessionId('veto'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: SECRETS.slack }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const body = backend.records[0]!.body as { content: { text: string }[] }
|
||||
expect(body.content[0]!.text).toBe(SECRETS.slack)
|
||||
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 () => {
|
||||
|
||||
Reference in New Issue
Block a user