@pleach/observe

@pleach/observe is the brownfield entry point to the Pleach audit ledger. It is 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) and writes one auditable row per LLM call to a backend you pick.

The row is a strict subset of @pleach/core's AuditableCall v13 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 tags?:         Record<string, string>;
}

Fields the v13 record carries that ObserveRow does not populate (the full RoutingDecision, the cascade walk history, the fingerprint payload, attestation hashes) stay undefined on the row. If a downstream consumer needs them, reach for @pleach/core — the runtime populates the full v13.

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(),
    });
  });
});

Cost attribution

@pleach/observe ships a tag-convention helper that populates the cost.* namespace on ObserveRow.tags for cross-tenant aggregation in dashboards.

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

Three 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.

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