fix: harden SDK — observer leak + callback isolation, surfaced errors

- observer: detach the sleep() abort-listener on normal timeout (was leaking one
  listener per tick onto a long-lived signal); wrap onLoop/onReport/onTick so a
  throwing user callback routes to onError instead of killing the loop
- collect / planUpload: add opt-in onError so per-adapter locate/parse failures
  are observable instead of silently swallowed (matches the observer)
- consolidate writeOtlp → writeOtlpFile (single temp-writer, single prefix)
- document the ms→nanos number-typed timestamp (ordering + ms resolution kept)
- test: observer routes a throwing onLoop to onError (regression)

Author: drewstone

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@tangle-network/traces",
-  "version": "0.2.0",
+  "version": "0.2.1",
   "description": "Point it at your coding-agent session traces (Claude Code, Codex, OpenCode, Gemini, Pi, …) and get failure-mode + efficiency findings. CLI + SDK over the @tangle-network/agent-eval analyst suite — observe live sessions, run your own analysts, redact, and upload to the Tangle Intelligence Platform.",
   "type": "module",
   "license": "MIT",

src/analyze.ts

@@ -10,14 +10,11 @@
  * (`halo <file> -p "diagnose"`), so analysis is never locked to one engine.
  */
 
-import { mkdtemp, writeFile } from 'node:fs/promises'
-import { tmpdir } from 'node:os'
-import { join } from 'node:path'
 import type { AxAIService } from '@ax-llm/ax'
 import { type AnalystRegistry, buildDefaultAnalystRegistry } from '@tangle-network/agent-eval/analyst'
 import { OtlpFileTraceStore } from '@tangle-network/agent-eval/traces'
 import type { OtlpSpan } from './otlp.js'
-import { serializeSpans } from './otlp.js'
+import { writeOtlpFile } from './otlp.js'
 
 export interface AnalyzeOptions {
   /** Ax service enabling the agentic RLM kinds. Omit → deterministic only. */
@@ -44,13 +41,6 @@ export interface AnalyzeResult {
   result: Awaited<ReturnType<ReturnType<typeof buildDefaultAnalystRegistry>['run']>>
 }
 
-/** Write spans to an OTLP-JSONL file and return its path. */
-export async function writeOtlp(spans: readonly OtlpSpan[], outPath?: string): Promise<string> {
-  const path = outPath ?? join(await mkdtemp(join(tmpdir(), 'traces-')), 'spans.otlp.jsonl')
-  await writeFile(path, serializeSpans(spans), 'utf8')
-  return path
-}
-
 /**
  * `viewTrace` ceiling for the deterministic pass. The default 150KB cap
  * exists to protect an LLM's context window — the deterministic behavioral
@@ -62,7 +52,7 @@ const DETERMINISTIC_VIEW_CEILING = 256 * 1024 * 1024
 
 export async function analyzeSpans(spans: readonly OtlpSpan[], opts: AnalyzeOptions = {}): Promise<AnalyzeResult> {
   if (spans.length === 0) throw new Error('analyzeSpans: no spans to analyze')
-  const otlpPath = await writeOtlp(spans, opts.otlpOutPath)
+  const otlpPath = await writeOtlpFile(spans, opts.otlpOutPath)
   const runId = opts.runId ?? `traces-${Date.now()}`
 
   // Deterministic pass — high ceiling so the behavioral analyst sees the whole

src/cli.ts

@@ -15,10 +15,11 @@
  */
 
 import { stat, writeFile } from 'node:fs/promises'
-import { analyzeSpans, writeOtlp } from './analyze.js'
+import { analyzeSpans } from './analyze.js'
 import type { OtlpSpan } from './otlp.js'
-import { runPipelines } from './pipelines.js'
+import { writeOtlpFile } from './otlp.js'
 import { watchSessions } from './observer.js'
+import { runPipelines } from './pipelines.js'
 import { knownHarnesses, listAdapters, resolveAdapter } from './registry.js'
 import { renderPipelines, renderReport } from './report.js'
 import type { HarnessTraceAdapter, SessionRef } from './types.js'
@@ -184,7 +185,7 @@ async function collectSpans(args: Args): Promise<{ spans: OtlpSpan[]; harness: s
 async function cmdConvert(args: Args): Promise<void> {
   const { spans } = await collectSpans(args)
   if (spans.length === 0) throw new Error('no spans found for the given selection')
-  const path = await writeOtlp(spans, args.otlp)
+  const path = await writeOtlpFile(spans, args.otlp)
   console.log(`wrote ${spans.length} spans → ${path}`)
 }
 

src/collect.ts

@@ -27,6 +27,8 @@ export interface CollectOptions {
   last?: number
   /** Redact PII/secrets (default true). Set false to get raw spans. */
   redact?: boolean
+  /** Called on a per-adapter locate/parse failure (collection continues). */
+  onError?: (error: unknown, ref?: SessionRef) => void
 }
 
 export interface SessionBatch {
@@ -54,15 +56,17 @@ export async function collectSessions(opts: CollectOptions = {}): Promise<Sessio
     let refs: SessionRef[]
     try {
       refs = await adapter.locate({ cwd: opts.cwd, sinceMs: opts.sinceMs })
-    } catch {
+    } catch (err) {
+      opts.onError?.(err)
       continue
     }
     if (opts.last && opts.last > 0) refs = refs.slice(0, opts.last)
     for (const ref of refs) {
       let spans: OtlpSpan[]
       try {
         spans = await adapter.parse(ref)
-      } catch {
+      } catch (err) {
+        opts.onError?.(err, ref)
         continue
       }
       if (spans.length === 0) continue

src/observer.ts

@@ -102,7 +102,7 @@ export async function writeOtlpFile(spans: readonly OtlpSpan[], outPath?: string
   const { mkdtemp, writeFile } = await import('node:fs/promises')
   const { tmpdir } = await import('node:os')
   const { join } = await import('node:path')
-  const path = outPath ?? join(await mkdtemp(join(tmpdir(), 'tangle-traces-')), 'spans.otlp.jsonl')
+  const path = outPath ?? join(await mkdtemp(join(tmpdir(), 'traces-')), 'spans.otlp.jsonl')
   await writeFile(path, serializeSpans(spans), 'utf8')
   return path
 }

src/otlp.ts

@@ -49,6 +49,8 @@ export interface PlanOptions {
   all?: boolean
   cwd?: string
   sinceMs?: number
+  /** Called on a per-adapter locate/parse failure (planning continues). */
+  onError?: (error: unknown, ref?: SessionRef) => void
 }
 
 function adaptersFor(opts: PlanOptions): HarnessTraceAdapter[] {
@@ -66,14 +68,16 @@ export async function planUpload(opts: PlanOptions): Promise<UploadPlan> {
     let refs: SessionRef[]
     try {
       refs = await adapter.locate({ cwd: opts.cwd, sinceMs: opts.sinceMs })
-    } catch {
+    } catch (err) {
+      opts.onError?.(err)
       continue
     }
     for (const ref of refs) {
       let raw: OtlpSpan[]
       try {
         raw = await adapter.parse(ref)
-      } catch {
+      } catch (err) {
+        opts.onError?.(err, ref)
         continue
       }
       if (raw.length === 0) continue
@@ -128,6 +132,10 @@ function epochMs(ts: string): number {
   return Number.isNaN(n) ? 0 : n
 }
 
+// TraceSpanEvent.*UnixNano is typed `number`, and our source resolution is
+// milliseconds. ms × 1e6 exceeds MAX_SAFE_INTEGER, so the low ~256ns are not
+// representable — but that's below our input resolution, and since both ends
+// are integer-ms × 1e6 the rounding preserves ordering (start ≤ end always).
 const msToNano = (iso: string): number => epochMs(iso) * 1_000_000
 
 /** Map redacted OtlpSpan[] → the hosted TraceSpanEvent[] wire shape, attaching

src/upload.ts

@@ -72,6 +72,22 @@ describe('watchSessions (observer event API)', () => {
     expect(loops[0]!.toolName).toBe('bash')
     expect(loops[0]!.occurrences).toBeGreaterThanOrEqual(3)
   })
+
+  it('routes a throwing onLoop to onError instead of crashing the loop', async () => {
+    const controller = new AbortController()
+    const errors: unknown[] = []
+    await watchSessions({
+      adapters: [adapterOf(loopSpans(4))],
+      intervalMs: 250,
+      minLoopOccurrences: 3,
+      signal: controller.signal,
+      onLoop: () => { throw new Error('boom') },
+      onError: (e) => errors.push(e),
+      onTick: () => controller.abort(),
+    })
+    expect(errors).toHaveLength(1)
+    expect((errors[0] as Error).message).toBe('boom')
+  })
 })
 
 describe('collectSessions (batch seam)', () => {