@adhd/sox-telemetry

A shared tracing/metrics substrate: a durable JSONL log sink, a log/withTimedEvent API that makes hangs visible (the "start" line lands before the work even begins, so a process that never returns is already on disk), a stage catalog for measuring wait-vs-work on contended resources, a self-check surface you can poll instead of grepping logs, and an opt-in bridge to the real OpenTelemetry SDK. It is the one place in a codebase that should own @opentelemetry/* — every other package can log/trace through this facade without paying the SDK's load cost unless initTelemetry({ otel: true }) actually asks for it.

pnpm add @adhd/sox-telemetry

Quick start

import { initTelemetry, log, withTimedEvent } from '@adhd/sox-telemetry';

const handle = initTelemetry({
  service: 'my-service',
  role: 'live-service', // required, closed union: 'live-service' | 'test' | 'cli' | 'harness'
  logDir: './logs',      // defaults to ~/.adhd/sox-ecosystem/<service>/logs
});

log.info('startup.begin', { pid: process.pid });

await withTimedEvent('db.migrate', { table: 'users' }, async () => {
  // ... do the work; a "db.migrate.start" line is already durable on disk
  // before this callback even runs, so a hang here is visible immediately.
});

console.log('records are landing in', handle.currentLogFilePath());
await handle.flush();
handle.close();

Every record — from log.*, from withTimedEvent, from a stage, from an OTel span — carries service, role, trace_id, pid, ts (ISO-8601), and level, stamped once by initTelemetry rather than per call site. Calling any emitter before initTelemetry() is safe: it falls back to a service: 'unlabeled', no-op sink rather than throwing, and prints one stderr warning on first use so an unwired composition root doesn't fail silently forever.

API reference

Initialization

function initTelemetry(opts: InitTelemetryOptions): TelemetryHandle;

interface InitTelemetryOptions {
  service: string;
  role: Role;                    // 'live-service' | 'test' | 'cli' | 'harness' — required
  logSink?: LogSink;              // 'file' (default) | 'stderr' | 'none' — never 'stdout'
  logDir?: string;                 // default: ~/.adhd/sox-ecosystem/<service>/logs
  maxBytes?: number;                // size-based log rotation threshold
  maxFiles?: number;                 // rotated files retained
  durable?: boolean;                  // writeSync durability, default true
  otel?: boolean;                       // bring up the real OTel SDK; default on for
                                         // 'live-service'/'cli', off for 'test'/'harness'
  snapshotEveryRecords?: number;         // activity-triggered metrics snapshot cadence, default 1000
}

interface TelemetryHandle {
  readonly service: string;
  readonly role: Role;
  currentLogFilePath(): string | null; // null means "no file sink configured" — the only meaning it has
  flush(): Promise<void>;              // await any buffered writes
  otelReady(): Promise<void>;          // resolves once OTel bring-up settles; never rejects
  close(): void;
}

function currentRuntimeState(): Readonly<{ service: string; role: Role; logSink: LogSink }>;

/** Resolve the role a composition root should actually pass to initTelemetry(),
 *  correcting a structural default toward 'harness' when an out-of-process
 *  integration harness set SOX_TELEMETRY_HARNESS=1 on this process's env. */
function resolveProcessRole(structuralDefault: Role): Role;

Sharing telemetry between libraries — acquireTelemetry

initTelemetry is last-wins: each call closes the previous sink, shuts the previous OTel provider down and installs a new configuration. Use it at a composition root that is the sole owner. A library or plugin that may share a process with another telemetry user should acquire instead:

function acquireTelemetry(opts: InitTelemetryOptions): TelemetryLease;

interface TelemetryLease extends TelemetryHandle {
  readonly owned: boolean;               // true: this call brought telemetry up; false: it joined an existing instance
  readonly conflicts: readonly string[]; // opts that differ from the owner's (service | role | logSink | logDir | otel)
  release(): Promise<void>;              // idempotent; resolves after teardown when it was the LAST lease; close() is an alias
}

const a = acquireTelemetry({ service: 'agent-mcp', role: 'live-service' });
const b = acquireTelemetry({ service: 'agent-mcp', role: 'live-service' }); // joins; options ignored
await a.otelReady();   // ONE provider, registered once
await a.release();     // provider stays up: b still holds a lease
await b.release();     // last lease: final snapshot, then sinks + provider close

The first acquirer owns the configuration; later acquirers' options are ignored (a mismatch is reported in lease.conflicts and logged once as telemetry.acquire.config_conflict). The owner record lives on the shared globalThis runtime slot, so two installed copies of this package share one owner and one reference count. A raw initTelemetry call supersedes the owner (it always did); the next acquireTelemetry then owns again, and leases of the superseded epoch cannot close the new one. handle.close() from initTelemetry is also generation-scoped now: a handle that was superseded by a later initTelemetry no longer tears down the live configuration.

Process identity attributes

Every product tracing through this package stamps the same three dimensions so a process run can be aggregated by session, checkout and population:

type WorktreeKind = 'main' | 'linked' | 'none';

function identityAttributes(): { session_id: string; worktree: WorktreeKind; role: Role };
function processIdentity(): { readonly sessionId: string; readonly worktree: WorktreeKind }; // resolved once, frozen
function resolveProcessIdentity(env?: ProcessEnv, cwd?: string, sessionId?: string): ProcessIdentity; // pure
function classifyWorktree(cwd?: string): WorktreeKind;
function toOtelAttributes(fields: Record<string, unknown>): OtelAttributes; // keeps string|number|boolean only
function requireTelemetryPrefix(prefix: unknown, owner?: string): string;   // required, no default; throws if blank
function prefixAttributes(prefix: string, fields: Record<string, unknown>): OtelAttributes; // { op } -> { 'prefix.op' }

const prefix = requireTelemetryPrefix(opts.prefix, 'my-plugin');
await withSpan(`${prefix}.op`, { ...prefixAttributes(prefix, { op: id }), ...identityAttributes() }, run);

• session_id is ADHD_SESSION_ID when set, else a process-stable randomUUID(); it is shared across duplicate installed copies of the package.
• Identity keys are unprefixed by design (a prefix would split one session across products). A product's own keys take a required prefix — there is deliberately no default.
• Spread identityAttributes() last so a configured key can never shadow an identity dimension.
• There is no release.* helper: service.version and sox.artifact_sha256 already ride every span as OTel resource attributes, and the full { version, artifact_sha256, git_sha } rides every JSONL record envelope. (git_sha is envelope-only; it is not a resource attribute.)

Logging

interface LogFields { [key: string]: unknown; }

const log: {
  debug: (event: string, fields?: LogFields) => void;
  info: (event: string, fields?: LogFields) => void;
  warn: (event: string, fields?: LogFields) => void;
  error: (event: string, fields?: LogFields) => void;
};

/** Truncate a value for safe, size-bounded logging (SQL text, error messages). */
function truncateForLog(s: string, maxLen?: number): string; // default maxLen 500

/** Log `<event>.start` immediately, run fn, then log `.finish`/`.error` with
 *  an elapsed duration_ms. The start line reaches the sink before fn even runs. */
function withTimedEvent<T>(event: string, fields: LogFields, fn: () => Promise<T>): Promise<T>;

/** Wrap every named method of obj so it emits withTimedEvent automatically,
 *  bound to the real target — `this` inside a wrapped method is never a Proxy. */
function instrumentBoundary<T extends object>(
  obj: T,
  opts: { component: string; methods: readonly (keyof T & string)[] },
): T;

import { instrumentBoundary } from '@adhd/sox-telemetry';

const store = instrumentBoundary(rawStore, { component: 'user-store', methods: ['get', 'put'] });
// store.get(...) / store.put(...) now emit user-store.get.start/finish/error automatically;
// every other method on rawStore passes through untouched.

Stages — measuring wait vs. work on a contended resource

interface StageDeclaration { paths: readonly string[]; }
type StageMap = Record<string, StageDeclaration>;

function declareStages<T extends StageMap>(pkg: string, stages: T): StageCatalog<T>;

class StageCatalog<T extends StageMap> {
  withContendedStage<K extends keyof T & string, R>(
    stage: K,
    stagePath: T[K]['paths'][number],
    admit: () => Promise<void>,
    work: () => Promise<R>,
  ): Promise<R>;
}

Declaring a stage's code paths up front makes an unwired sibling path a visible finding in telemetrySelfCheck() (paths_with_zero_samples) instead of a metric that silently never fires:

import { declareStages } from '@adhd/sox-telemetry';

const stages = declareStages('embed-service', {
  embed: { paths: ['write', 'heal'] as const },
});

await stages.withContendedStage(
  'embed',
  'write',
  async () => { /* acquire a slot/lock — this is the "wait" phase */ },
  async () => { /* do the embedding work — this is the "work" phase */ return 'ok'; },
);

Self-check

function telemetrySelfCheck(): TelemetrySelfCheck;

interface TelemetrySelfCheck {
  window: 'since process start';
  role: Role;
  stages_declared: number;
  stages_with_zero_samples: string[];
  paths_with_zero_samples: string[];
  stages: StageSelfCheck[];
  children: ChildrenTelemetrySelfCheck;
  otel: { state: 'disabled' | 'pending' | 'ready' | 'failed'; spans_enabled: boolean };
  metric_persistence: { written: number; records_since: number; every_records: number; file: string | null };
}

A process's own health is a pull, not a log-scrape: telemetrySelfCheck() reports whether every declared stage/path has ever produced a sample, whether spawned children ever initialized telemetry, and whether the OTel SDK actually came up — each field distinguishes "never happened" from "not asked for" rather than collapsing both into silence.

Spans and metrics (OpenTelemetry, opt-in)

function withSpan<R>(name: string, attrs: OtelAttributes, fn: (span: OtelSpanHandle) => Promise<R>): Promise<R>;
function otelReady(): Promise<void>;
function snapshotMetrics(reason: 'activity' | 'pull' | 'shutdown' | 'interval'): Promise<void>;

interface OtelSpanHandle {
  setAttributes(attrs: OtelAttributes): void;
  recordError(err: unknown): void;
}

import { withSpan } from '@adhd/sox-telemetry';

await withSpan('checkout.process', { orderId }, async (span) => {
  span.setAttributes({ itemCount: items.length });
  // ... do the work; span.recordError(err) on catch if you want it on the span
});

When initTelemetry({ otel: true }) hasn't been called (or hasn't finished — the SDK loads via a dynamic import so it settles a few ms after initTelemetry returns), withSpan is fn plus one no-op object — the same call site works whether or not anything is actually collecting spans.

Trace-id propagation

function newTraceId(): string;                       // ulid — sortable, monotonic within a process
function currentTraceId(): string | undefined;
function traceIdOrNew(): string;
function withTrace<T>(traceId: string, fn: () => T): T;
function runWithNewTrace<T>(fn: (traceId: string) => T): T;

import { runWithNewTrace, log } from '@adhd/sox-telemetry';

runWithNewTrace((traceId) => {
  log.info('request.start', { traceId }); // every nested telemetry call in this
  // continuation (sync or awaited-async) shares the same active trace id.
});

The durable sink

class DurableJsonlSink {
  constructor(opts: JsonlSinkOptions);
  currentPath(): string;
  plannedPath(): string;
  reconfigure(opts: JsonlSinkOptions): void;
  flush(): Promise<void>;
  write(line: string): void;
  close(): void;
}

interface JsonlSinkOptions {
  dir: string;
  component: string;      // file-name prefix; role-qualify with '.', e.g. 'my-svc.live', not '-'
  maxBytes?: number;       // default 20 MB rotation threshold
  maxFiles?: number;       // default 7 retained
  durable?: boolean;       // writeSync (true, default) vs. a fire-and-forget stream
}

initTelemetry constructs one of these for you; reach for DurableJsonlSink directly only if you need a second, independently-rotated JSONL stream outside the main telemetry record flow. Writes are synchronous (durable: true) by default because a fire-and-forget stream loses buffered records on a hard kill — write() returning means the line is already on disk.

Raw log channel — createLogChannel, createRedactor, captureBody, withLogContext

A log channel is a second, independently retained JSONL stream beside the span/event stream. It is for raw records: structured ops events (logEvent) and wire records with full headers and bodies (logRaw). It never reads an environment variable; the caller resolves the directory.

const { createLogChannel, createRedactor, captureBody, withLogContext } = require('@adhd/sox-telemetry');
const os = require('node:os');
const path = require('node:path');

const redactor = createRedactor();                 // header values, key-shaped strings, secret-named keys
const channel = createLogChannel({
  dir: path.join(os.homedir(), '.adhd', 'my-service', 'logs'),
  component: 'my-service.api',                      // file prefix; separate parts with '.', never '-'
  durability: { mode: 'hybrid', syncMaxBytes: 65536 },
  redactor,
});
channel.registerSecret('my-live-api-key-1234');     // exact-value scrub (>= 8 chars) from now on

// Correlation ids set here are merged into every record written inside the callback,
// across awaits, even if two copies of this package are installed.
await withLogContext({ request_id: 'req-42' }, async () => {
  const body = captureBody(JSON.stringify({ prompt: 'hello' }), { maxBytes: 4 * 1024 * 1024 });
  channel.logRaw('provider.request', {
    method: 'POST',
    url: 'https://api.example.test/v1/chat',
    headers: { authorization: 'Bearer my-live-api-key-1234', accept: 'application/json' },
    body,                                           // -> body, body_bytes, body_sha256, truncated
  });
  channel.logEvent('provider.retry', { attempt: 2 }, 'warn');
});

await channel.close();                              // flushes the queue and pending gzips

The authorization header and the registered key are written as [REDACTED:<hash>:len=<n>]; request_id rides on both records.

File names. Each process writes its own file, so concurrent processes never share an open file:

<component>-<YYYY-MM-DD>.<fileLabel>.<pid>.jsonl                  live file
<component>-<YYYY-MM-DD>.<fileLabel>.<pid>.<epochMs>-<seq>.jsonl  rotated
<component>-<YYYY-MM-DD>.<fileLabel>.<pid>.<epochMs>-<seq>.jsonl.gz  rotated + compressed

fileLabel defaults to proc; fileScope: 'shared' restores the single shared file of DurableJsonlSink.

Defaults.

| Option | Default | Meaning | | --- | --- | --- | | maxBytes | 64 MiB | rotate the live file past this size | | gzipRotated | true | gzip rotated files off the hot path (about 100x on repetitive JSON) | | maxAgeDays | 14 | delete files older than this | | maxDirBytes | 2 GiB | delete oldest files until this component's files fit; the live file is never deleted | | bodyMaxBytes | 4 MiB | advisory per-body cap for callers of captureBody | | floodBreaker | 256 MiB per 10 min | past this logRaw volume, records become digest-only (body: null, body_sha256 kept) and one log.budget_tripped is written | | durability | { mode: 'hybrid', syncMaxBytes: 65536 } | see below | | queueMaxBytes | 16 MiB | hybrid queue bound; overflow drops the oldest and is counted in stats().dropped |

Durability modes.

• sync: every record is writeSynced on an O_APPEND descriptor. When logEvent/logRaw returns, the record survives kill -9 of the process.
• hybrid (default): records up to syncMaxBytes are written synchronously; larger ones go through a bounded queue flushed before the next synchronous record, on flush(), and on beforeExit/exit. A kill -9 can lose queued large records, never a small one that returned.

Redaction. The channel redacts both the structured fields and the serialized line, so no call site can forget it. Pass redactor: identityRedactor only in tests. initTelemetry({ redactor }) applies the same redactor to every log.* record and to span .start/.finish records (span attributes and captured args); with no redactor, output is unchanged.

Other exports. BodyCapture is the streaming form of captureBody (append(chunk), then finish()); currentLogContext() reads the merged context of the current async scope; channel.destination() is a pino-compatible destination (pino.multistream([{ stream: channel.destination() }])); channel.stats() reports written, dropped, serializeErrors, rotated, floodTripped.

Verification: npx nx run sox-telemetry:e2e drives 4 real processes through one directory, a kill -9 mid-write, gzip rotation replay, and a cross-copy run against the published 0.4.1.

Child processes and worker threads

const SOX_TELEMETRY_INIT: string; // env var name a parent writes and a child reads

function forkChild(modulePath: string, telemetry: InitTelemetryOptions, forkOpts?: ForkOptions): ChildProcess;
function spawnWorker(workerPath: string, telemetry: InitTelemetryOptions, workerOpts?: WorkerOptions): Worker;
function bootstrapChildTelemetry(defaults: InitTelemetryOptions): TelemetryHandle;
function childTelemetrySnapshot(): ChildTelemetrySnapshot;

interface ChildTelemetrySnapshot {
  service: string;
  role: Role;
  logSink: LogSink;
  filePath: string | null; // null iff logSink !== 'file'
  pid: number;
}

initTelemetry() only initializes the calling process's own state — a node:child_process.fork'd child or a worker_threads.Worker starts with fresh, uninitialized telemetry unless the parent uses forkChild/spawnWorker (which inject SOX_TELEMETRY_INIT into the child's env) and the child calls bootstrapChildTelemetry at its own entrypoint:

// parent.ts
import { forkChild } from '@adhd/sox-telemetry';
const child = forkChild('./worker.js', { service: 'my-service', role: 'live-service' });

// worker.js — the child's own composition root
import { bootstrapChildTelemetry } from '@adhd/sox-telemetry';
bootstrapChildTelemetry({ service: 'my-service-worker', role: 'live-service' });

A child that never calls bootstrapChildTelemetry shows up as unacked in the parent's telemetrySelfCheck().children rather than silently vanishing.

Gotchas

• Libraries and plugins should call acquireTelemetry, not initTelemetry: two initTelemetry calls in one process reconfigure each other (last-wins) while the OTel API's global registration is first-wins.
• role is required and has no default — pass whichever of 'live-service' | 'test' | 'cli' | 'harness' genuinely describes this process. There is no 'stdout' LogSink: writing telemetry to stdout is not offered as an option, because a stray stdout write can corrupt a process's own JSON-RPC/stdio protocol channel.
• otel: true costs real time and memory to bring up (SDK trace/metrics providers + context-manager); it defaults on for 'live-service'/'cli' and off for 'test'/'harness' for that reason. Pass it explicitly to opt a test in.
• currentLogFilePath() and metric_persistence.file return null for "no file sink configured" and a real path otherwise (even before the first write lands) — never ''. Treat null and a string as the only two states.
• otelReady() never rejects, even if OTel bring-up failed — a failed bring-up is reported through telemetrySelfCheck().otel.state ('failed') instead, because telemetry must not be able to prevent the service it's observing from starting.
• DurableJsonlSink never throws on a write fault (disk full, fd revoked, directory unwritable) — it drops the record silently rather than propagating, because a logging fault must never break or slow the operation it is observing.