@pleach/observe

@pleach/observe writes one audit row per LLM call to a backend you choose — Postgres, Supabase, your OpenTelemetry collector, or an in-memory buffer. It's a small, destination-flexible SDK — init, recordCall, subagent, and four destination factories — that sits in front of whatever agent loop you already run (the Vercel AI SDK, LangChain, the Anthropic or OpenAI SDK called directly, an in-house orchestrator), which makes it the brownfield entry point to the Pleach audit ledger.

The row is a strict subset of @pleach/core's AuditableCall v19 record plus the TokenCostRecord field set. A row written through the SDK is forward-compatible with the full runtime row, so a buyer who later adopts @pleach/core keeps the existing audit history without a migration.

When @pleach/observe fits

Two reader-shaped questions:

• Do you already have an agent loop you do not want to rewrite? Then observe is the brownfield hook. You add roughly 15 lines around your existing LLM calls and get the audit row, per-subagent attribution, and the destination of your choice.
• Are you starting fresh, or do you need replay determinism, family-locked routing, reactive channels, or checkpoint/restore? Then @pleach/core is the better fit — those are runtime properties, not row properties, and the SDK does not carry them by design.

The two paths are documented as a pair on Adoption paths. observe is the brownfield SDK; @pleach/core is the greenfield substrate.

Install

npm install @pleach/observe

@pleach/core is a peerDependency (^0.1.0 today; bumps to ^1 once @pleach/core@1.0.0 ships). Only the ProviderDecisionLedger TypeScript interface is referenced — the SDK does not pull a runtime implementation of the substrate.

Zero @opentelemetry/* runtime dependencies. The OTel destination is buyer-callback only: your OTel SDK stays on your side; the SDK hands you a GenAI-semantic-convention envelope and your callback ships it to the exporter you already configured.

Zero transport peer dependencies for Postgres / Supabase. Pass a buyer-constructed pg.Pool / pg.Client or SupabaseClient; the SDK uses minimal structural types so it never has to import the upstream packages.

Quickstart — paste and run

The Memory destination has no external dependencies; ideal for tests, local development, and first-look demos.

import { init, recordCall } from "@pleach/observe";
import { memory } from "@pleach/observe/destinations";

const dest = memory();
init({ destination: dest });

recordCall({
  turnId: "turn_001",
  providerId: "anthropic",
  callClass: "synthesize",
  family: "anthropic",
  model: "claude-opus-4-7",
  inputTokens: 1000,
  outputTokens: 250,
  costUSD: 0.0125,
  startedAt: Date.now(),
  completedAt: Date.now() + 1200,
});

console.log(dest.rows.length); // 1
console.log(dest.rows[0].model); // "claude-opus-4-7"

That is the entire surface for a one-shot write: one init, one recordCall. Swap memory() for any of the three other destinations below to ship the same row to Postgres, Supabase, or your OTel collector.

The module-level API

Four module-level entry points. init is called once per process; the other three are bound to the module singleton that init creates.

init(config: ObserveConfig): void
recordCall(row: ObserveRow): void
getRecorder(): ObserveRecorder    // throws if called before init()
subagent(name: string): ObserveRecorder & { run: <T>(cb: () => Promise<T>) => Promise<T> }

init picks the destination, caches the config on a module-level singleton, and is singleton-per-process — a second init call throws (rather than silently replacing the live recorder, which would orphan in-flight writes against the prior destination). Re-initialization in tests is supported via the internal __resetForTesting hook exported from the root barrel.

recordCall is the load-bearing entry. One call produces one ObserveRow written to the configured destination. The destination decides whether the write is synchronous (memory) or asynchronous (Postgres / Supabase / OTel batching).

getRecorder is an escape hatch for advanced wiring — returns the active recorder so you can pass it across module boundaries without re-importing recordCall. It throws if called before init (it never returns undefined).

subagent(name) is the per-sub-agent attribution helper, covered below.

ObserveConfig

interface ObserveConfig {
  readonly destination: ObserveDestination;
  readonly sampling?: { readonly rate: number };       // 0..1, FNV-1a per-turnId
  readonly redactor?: (row: ObserveRow) => ObserveRow; // pure, sync, pre-write
  readonly redact?: PIIRedactionConfig;                // substrate-side policy
  readonly throwOnDestinationError?: boolean;          // default false (silent-swallow)
}

• sampling.rate is hash-deterministic per turnId (FNV-1a) — all recordCall invocations inside one turn agree (ship-together or drop-together). rate === 0 drops everything; rate === 1 (default when omitted) keeps everything. Validated at init time; throws on NaN, non-finite, < 0, or > 1.
• redactor is the buyer-supplied PII hook. Applied to every row before destination.write. Pure function contract. If the callback throws, the SDK swallows, warns via the [Observe:redactor-threw] anchor, and drops the row (fail-closed — the raw pre-redaction row is never written).
• redact is the substrate-side policy (see PII redaction below) — when configured, the recorder applies it automatically per row.
• throwOnDestinationError opts out of the v0.x silent-swallow contract. When true, recordCall re-throws synchronous destination errors and leaves async rejections unhandled so process-level handlers can observe them. Default false.

The ObserveRow shape

The row is a strict subset of @pleach/core's AuditableCall v13 record plus the TokenCostRecord field set. Forward-compatibility is the point: any row the SDK writes is a valid AuditableCall row, and any consumer that already reads the v13 schema (@pleach/eval, @pleach/compliance, downstream dashboards) reads SDK rows without a code change.

interface ObserveRow {
  readonly turnId:        string;       // ULID
  readonly providerId:    string;       // "anthropic" | "openai" | "google" | "deepseek" | …
  readonly callClass:     CallClass;    // "synthesize" | "reasoning" | "utility" | "converse"
  readonly family:        ProviderFamily;
  readonly model:         string;
  readonly inputTokens:   number;
  readonly outputTokens:  number;
  readonly costUSD:       number;
  readonly startedAt:     number;       // epoch millis
  readonly completedAt:   number;       // epoch millis
  readonly subagent?:     string;       // set automatically by subagent(...)
  readonly subagentPath?: readonly string[]; // ALS-scope path when nested
  readonly fingerprint?:  string;       // `fp_v1:`-prefixed, opt-in column
  readonly tags?:         Record<string, string>;
  readonly byokFailureChain?: readonly {
    readonly provider:   ProviderFamily;
    readonly keyHint:    string;        // last 4 chars, never the full key
    readonly statusCode: number;        // -1 for transport-level failures
    readonly reason:     string;
  }[];
}

byokFailureChain is populated by the host when a byok-* transport call fails and the family-strict cascade pivots to the next provider — one entry per attempted BYOK provider. It carries a four-character keyHint and never the key itself, so the row stays safe to ship to a dashboard your key-holders don't control.

Fields the v13 record carries that ObserveRow does not populate (the full RoutingDecision, the cascade walk history, attestation hashes) stay undefined on the row. If a downstream consumer needs them, reach for @pleach/core — the runtime populates the full v13. The exception is the fingerprint: ObserveRow carries its own fp_v1: hash over {family, model, callClass, input}, which is a narrower value than v13's fingerprint payload and computed by this SDK rather than lifted from the record.

tags is the open extension hook. The SDK never inspects the map beyond the cost.* namespace (see Cost attribution below); it is passed through to the destination as-is. Conventional keys (env, release, user.tier) align with the existing audit-ledger conventions.

The four destinations

init({ destination }) accepts any of four factories from the ./destinations subpath. Each is a function that returns an ObserveDestination.

Postgres

import { Pool } from "pg";
import { init, recordCall } from "@pleach/observe";
import { postgres } from "@pleach/observe/destinations";

const pgClient = new Pool({ connectionString: process.env.DATABASE_URL });
init({
  destination: postgres({ pgClient, tableName: "pleach_observe_calls" }),
});

recordCall({
  turnId: "turn_001",
  providerId: "openai",
  callClass: "synthesize",
  family: "openai",
  model: "gpt-5",
  inputTokens: 850,
  outputTokens: 320,
  costUSD: 0.0091,
  startedAt: Date.now() - 900,
  completedAt: Date.now(),
});

Reuses a buyer-owned pg-shaped client. The SDK does not add pg as a dependency; the structural type is enough. The buyer-owned table must include the columns the parameterized INSERT targets — see the package's docs/postgres.md for the full CREATE TABLE and the optional ledger-injection mode (postgres({ ledger })) where you pass a ProviderDecisionLedger adapter from @pleach/core directly.

Postgres is the production default. Hosts already running Postgres for their application database get cost attribution joinable to the billing schema with one GROUP BY tenant_id.

Supabase

import { createClient } from "@supabase/supabase-js";
import { init, recordCall } from "@pleach/observe";
import { supabase } from "@pleach/observe/destinations";

const client = createClient(process.env.SUPABASE_URL!, process.env.SUPABASE_KEY!);
init({
  destination: supabase({ client, tableName: "pleach_observe_calls" }),
});

recordCall({
  turnId: "turn_supabase_001",
  providerId: "google",
  callClass: "reasoning",
  family: "google",
  model: "gemini-2.5-pro",
  inputTokens: 2100,
  outputTokens: 480,
  costUSD: 0.0185,
  startedAt: Date.now() - 1450,
  completedAt: Date.now(),
});

Accepts a buyer-constructed SupabaseClient. RLS, auth, and table schema stay under buyer control. An RLS template policy ships in the package's docs/supabase.md. The optional ledger-injection mode (supabase({ ledger })) is the same shape as Postgres above.

The trade-off vs the direct Postgres destination is the round trip through PostgREST. At >100 rows/s sustained, switch to the Postgres destination and keep Supabase Auth + Storage as separate concerns.

OpenTelemetry

import { init, recordCall } from "@pleach/observe";
import { otel } from "@pleach/observe/destinations";

init({
  destination: otel({
    serviceName: "my-agent",
    exportSpan: (envelope) => {
      // Hand `envelope` to your @opentelemetry/sdk-trace-base exporter.
      // SDK builds attributes per OTel GenAI semantic conventions.
      myExporter.export(envelope);
    },
  }),
});

recordCall({
  turnId: "turn_otel_001",
  providerId: "anthropic",
  callClass: "synthesize",
  family: "anthropic",
  model: "claude-sonnet-4-5",
  inputTokens: 1500,
  outputTokens: 600,
  costUSD: 0.0228,
  startedAt: Date.now() - 1700,
  completedAt: Date.now(),
});

Two callback modes ship:

• exportSpan(envelope) — the SDK builds the GenAI envelope and hands it to you (the example above).
• export(row) — pass-through; the SDK hands you the raw ObserveRow and you map it yourself.

This is the "I already have observability" destination. No new database to provision, no new dashboard to wire — the rows land in your existing backend through your own OTel configuration.

Memory

import { init, recordCall } from "@pleach/observe";
import { memory } from "@pleach/observe/destinations";

const sink = memory();
init({ destination: sink });

// later, in a test:
expect(sink.rows).toHaveLength(2);
expect(sink.rows[0].costUSD).toBeCloseTo(0.0143, 4);
sink.clear(); // explicit reset between tests

In-process buffer. Useful in unit tests and as a @pleach/eval fixture; not suitable for production. The buffer does not drain on init — your test owns the buffer's lifecycle, by design.

Sub-agent attribution

Attribute calls to named planner, critic, or tool-runner sub-agents in your UI. subagent(name) is arity-1: it returns a scoped recorder. Pick the mode that matches your loop.

import { init, subagent } from "@pleach/observe";
import { memory } from "@pleach/observe/destinations";

init({ destination: memory() });
const planner = subagent("planner");
const critic = subagent("critic");

planner.recordCall({
  turnId: "t1",
  providerId: "anthropic",
  callClass: "reasoning",
  family: "anthropic",
  model: "claude-opus-4-7",
  inputTokens: 1200,
  outputTokens: 320,
  costUSD: 0.014,
  startedAt: Date.now() - 900,
  completedAt: Date.now(),
});

critic.recordCall({
  turnId: "t1",
  providerId: "openai",
  callClass: "utility",
  family: "openai",
  model: "gpt-5",
  inputTokens: 450,
  outputTokens: 80,
  costUSD: 0.003,
  startedAt: Date.now() - 350,
  completedAt: Date.now(),
});

import { init, recordCall, subagent } from "@pleach/observe";
import { memory } from "@pleach/observe/destinations";

init({ destination: memory() });

await subagent("planner").run(async () => {
  await subagent("executor").run(async () => {
    // Row written here is tagged subagent: "executor",
    // subagentPath: ["planner", "executor"].
    recordCall({
      turnId: "t1",
      providerId: "anthropic",
      callClass: "synthesize",
      family: "anthropic",
      model: "claude-opus-4-7",
      inputTokens: 900,
      outputTokens: 220,
      costUSD: 0.011,
      startedAt: Date.now() - 800,
      completedAt: Date.now(),
    });
  });
});

Reading rows back

query(filter) is the read side. It ships on the Memory, Postgres, and Supabase destinations; OTel is write-only, on the reasoning that your OTel backend already owns the read path.

import { getRecorder } from "@pleach/observe"

const rows = await getRecorder().query({
  fingerprint: "fp_v1:…",          // exact match, prefix included
  since: new Date(Date.now() - 7 * 86_400_000),
  until: new Date(),
  limit: 100,
})

Every field is optional; a filter with no fields means all rows, and a destination may cap that case at its own default (Memory uses 1,000).

Two boundary rules worth reading before you aggregate on this:

• The window is half-open, [since, until) — since is inclusive, until is exclusive. Adjacent windows tile without double-counting a row on the boundary.
• Both bounds compare against startedAt, the row's nominal begin time, not completedAt — the same convention as @pleach/core's ProviderDecisionLedger.query. A long call is attributed to the window it started in.

Each accepts either epoch milliseconds or a Date.

A destination without a query implementation throws QueryNotSupportedError rather than returning []. That is deliberate: an empty array is indistinguishable from "no rows matched", so instanceof QueryNotSupportedError is how you branch to your own read path instead of silently reporting zero.

Persisting the fingerprint column

ObserveRow.fingerprint is a stable fp_v1:-prefixed hash of {family, model, callClass, input}, computed by computeObserveFingerprint(input) or recorder.fingerprint(input).

The Postgres and Supabase destinations only write it to a column when you opt in with persistFingerprint: true — it is off by default because the column needs an index to be worth querying, and adding one to an existing table is your migration to run, not the SDK's.

Knowing whether the recorder is live

init() installs a destination; before that, recordCall is a silent no-op. isRecorderInitialized() returns whether that has happened, for callers that would rather branch explicitly than rely on the no-op.

Cost attribution

@pleach/observe ships a tag-convention helper that populates the cost.* namespace on ObserveRow.tags for cross-tenant aggregation in dashboards. isCostAttributionKey(key) and extractCostAttributionTags(tags) read the same convention back out, so a dashboard doesn't re-derive the cost.* prefix rule by hand.

import { recordCall, costAttribution } from "@pleach/observe";

recordCall({
  turnId: "t1",
  providerId: "anthropic",
  callClass: "synthesize",
  family: "anthropic",
  model: "claude-opus-4-7",
  inputTokens: 900,
  outputTokens: 220,
  costUSD: 0.011,
  startedAt: Date.now() - 800,
  completedAt: Date.now(),
  tags: costAttribution({
    workflowId: "lead-enrichment",
    tenantId: "acme-corp",
    featureId: "draft-email",
  }),
});

Three canonical keys lock at v1.0.0: cost.workflow_id, cost.tenant_id, cost.feature_id. Other cost.* keys are reserved for future use; reach for tags directly for buyer-defined dimensions outside the namespace.

Optional PII redaction

@pleach/observe ships a built-in matcher catalog so you can opt in to substrate-side PII scrubbing without writing your own regexes.

import { init } from "@pleach/observe";
import { memory } from "@pleach/observe/destinations";
import { BUILT_IN_MATCHERS, DEFAULT_REPLACEMENT } from "@pleach/observe";

init({
  destination: memory(),
  redact: {
    fields: ["subagent", "tags.user_email", "tags.user_phone"],
    matchers: [
      ...BUILT_IN_MATCHERS, // email + phone + SSN
      // ...your custom matchers
    ],
    replacement: DEFAULT_REPLACEMENT, // "[REDACTED]"
  },
});

Public surface for the redaction config:

The ObserveDestination interface

Custom destinations are first-class. The interface is intentionally minimal:

interface ObserveDestination {
  write(row: ObserveRow): Promise<void> | void;
  flush?(): Promise<void>;
}

write accepts one row and is allowed to be sync (the memory destination is) or async (the Postgres destination is). flush is optional; if your destination buffers, drain on a process shutdown hook.

A buyer wiring a custom destination — say, a Kafka topic for a streaming pipeline, or a Cloudflare Queues binding — implements the two methods and passes the result to init. No registration, no service locator; the destination is a value, not a name.

API surface — what to import from where

Seven publishable subpaths (explicit, not glob):

MemoryDestination (with .rows getter + .clear()), PostgresDestination (with .mode), SupabaseDestination (with .mode), and OtelSpanEnvelope are exported from their respective destination modules; type re-export from ./destinations is deferred to a follow-up consolidation.

Writing your own destination

@pleach/observe/testing ships the conformance harness the built-in destinations are graded by, so a destination you write is held to the same contract rather than to your own reading of it.

import { assertObserveDestinationConformance } from "@pleach/observe/testing"

await assertObserveDestinationConformance(() => myDestination())

assert* throws on the first failed check; runObserveDestinationConformance returns a ConformanceReport instead, for a suite that wants to report every failure in one pass rather than stopping at the first.

Run it in your own test suite. A destination that satisfies the ObserveDestination type but not the harness type-checks and then misbehaves at runtime — the interface constrains shape, the harness constrains behaviour.

Wiring the integration emitters

The three integration subpaths each expose a factory or projector for a brownfield emit path. Typical wiring:

• withObserveEmit(inner, opts?) wraps an existing ProviderDecisionLedger so each recordCall(call) also writes an ObserveRow. The inner ledger resolves first; the observe write is fire-and-forget and skips silently until init() has run. Pass { adapter } to override the default AuditableCall → ObserveRow projection.
• createCostEmitter({ writer, familyDeriver?, callClassResolver? }) returns an emitter whose emit(input) derives a cost row from a raw provider response and hands it to your writer. emit returns void — do not await it; the writer's soft-fail keeps the response path unblocked. Emit is skipped when messageId is empty or family / callClass cannot be resolved.
• projectGatewayCostEventToObserveRow(event, opts?) maps a @pleach/gateway cost event (breach / warn / defer) onto an ObserveRow and returns it without writing. The tags["gateway.cost.kind"] discriminator carries the event kind; the caller forwards the row to a destination.

import { withObserveEmit } from "@pleach/observe/integrations/ledger-decorator";

// Fan every audit write out to an observe row too.
const ledger = withObserveEmit(existingLedger);

Sampling gate

shouldSample(turnId, rate) is the deterministic write gate — an FNV-1a hash of turnId compared against rate, so every recordCall within one turn agrees. rate >= 1 always samples; rate <= 0 never does. The recorder is the typical caller, but you can invoke it directly. validateSamplingConfig({ rate }) is the init()-time check — it throws on NaN, non-finite, or out-of-range rate.

What is hookable, what is not

The SDK exposes a deliberate subset of the substrate. Five primitives are hookable from a brownfield agent loop:

Five things are not hookable, by design — they are properties of the runtime, not of the row:

• Family-locked routing. The cascade-on-503 walk and the cross-family non-widening invariant live in @pleach/core's routing decision. The SDK observes calls; it does not make them.
• Reactive channels. Channel subscription, fan-out, and back-pressure are @pleach/core-internal.
• Replay determinism. Replaying a recorded turn against a fresh runtime requires the event log — the SDK writes a subset of fields, not the full log.
• Checkpoint / restore. @pleach/core/checkpointing is the surface; the SDK has no notion of session state.
• Time-travel. @pleach/replay's event-granular forking consumes the canonical event log, which the SDK does not produce.

The split is the load-bearing line between brownfield and greenfield: the row is the substrate; the runtime is the engine. A buyer who needs the engine adopts @pleach/core. A buyer who needs the row uses the SDK and writes to wherever they want.

Composing with @pleach/core

When a buyer ends up running both — the SDK for legacy code paths, the runtime for new ones — the two compose cleanly:

• The runtime can detect the SDK's init config at startup. If a destination is configured, runtime audit writes can route through the SDK's transport instead of opening a parallel sink.
• Fingerprint compute is shared. Both surfaces import from the same @pleach/core/fingerprint module; the SDK does not carry a second implementation.
• Sub-agent attribution survives the migration. Rows written through subagent("planner").recordCall(...) carry the same subagent field that runtime-written rows do.

The brownfield-to-greenfield migration is therefore monotonic — existing rows keep their ids and join cleanly to runtime-written rows. The adoption paths page walks the migration shape end-to-end.

License + versioning

Published under FSL-1.1-Apache-2.0 (Functional Source License with Apache 2.0 as the future license) — same license posture as the rest of the @pleach/* substrate set. Source- available, usable in production, free of charge during the FSL window; auto-transitions to permissive Apache 2.0 two years after first stable publish. See Versioning for the release-cadence policy.

The 0.x line is pre-1.0 — pin the exact 0.x.y version you adopt. The 0.0.x → 0.1.0 jump and the 0.1.x → 1.0.0 jump are both non-caretable; ^0.1.0 will not pick up later 0.x releases.

Roadmap

Highlights for 0.x → 1.0:

• Sampling configuration (init({ sampling: { rate: 0.1 } })) — shipped.
• Buyer-supplied redactor hook — shipped.
• Substrate-side PII redaction (init({ redact: { fields, matchers } })) — shipped, matcher catalog locked.
• First publish-rehearsal (0.2.0) — operator-attended.

v1+ candidates:

• Pleach-hosted Observe destination (one option among several; never the only option). A pleachHosted factory is already reserved on @pleach/observe/destinations and throws HostedBackendNotYetGA on first write until the receiver ships separately — wiring the call now is safe; the receiver flip will land as a patch release with no SDK-side change.
• Query surface (AuditableCallQuery → ObserveRow).
• Cost-attribution dashboard.
• Richer sub-agent attribution (deeper nesting via AsyncLocalStorage where supported).

Where to go next