# @pleach/observe (/docs/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`](/docs/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 [#when-pleachobserve-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`](/docs/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](/docs/adoption-paths). `observe` is the brownfield SDK; `@pleach/core` is the greenfield substrate. ## Install [#install] ```bash 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 [#quickstart--paste-and-run] The Memory destination has no external dependencies; ideal for tests, local development, and first-look demos. ```typescript 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 [#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. ```typescript init(config: ObserveConfig): void recordCall(row: ObserveRow): void getRecorder(): ObserveRecorder // throws if called before init() subagent(name: string): ObserveRecorder & { run: (cb: () => Promise) => Promise } ``` `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` [#observeconfig] ```typescript 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](#optional-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-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. ```typescript 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; } ``` 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`](/docs/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](#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 [#the-four-destinations] `init({ destination })` accepts any of four factories from the `./destinations` subpath. Each is a function that returns an `ObserveDestination`. | Destination | Factory | Use case | | ------------- | --------------------------------------------------------------------------- | ------------------------------------------------------- | | Memory | `memory({ maxRows? })` | Tests, local dev, demos | | Postgres | `postgres({ pgClient, tableName?, schemaName? })` or `postgres({ ledger })` | Self-hosted production; existing PG infrastructure | | Supabase | `supabase({ client, tableName? })` or `supabase({ ledger })` | Managed PG; RLS-friendly; quick start | | OpenTelemetry | `otel({ exportSpan, serviceName? })` or `otel({ export, serviceName? })` | Existing OTel collector / Honeycomb / Datadog / Grafana | ### Postgres [#postgres] ```typescript 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 [#supabase] ```typescript 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 [#opentelemetry] ```typescript 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(), }); ``` The SDK builds the GenAI-semantic-convention envelope (span name `` `.` `` — e.g. `anthropic.synthesize`; OTel GenAI attributes `gen_ai.system`, `gen_ai.request.model`, `gen_ai.usage.input_tokens`, `gen_ai.usage.output_tokens`, `gen_ai.operation.name`, plus the SDK-additive `pleach.observe.turn_id`, `pleach.observe.provider_id`, `pleach.observe.cost_usd`, and `pleach.observe.tag.*`) and hands it to your callback. You own the OTel SDK on your side: construct a `BatchSpanProcessor` against an OTLP exporter pointed at Honeycomb, Datadog, Grafana Tempo, or any OTLP collector, and forward the envelope from `exportSpan`. 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 [#memory] ```typescript 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 [#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. Grab the recorder once, call `.recordCall(...)` directly. Zero `AsyncLocalStorage` overhead; ideal when you already thread per-agent identity through your code. ```typescript 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(), }); ``` Each row is automatically tagged `subagent: "planner"` or `subagent: "critic"`. Wrap your sub-agent invocation in `.run(async () => { ... })` and any `recordCall` issued inside the callback inherits a `subagentPath` automatically. Use this for nested planner/executor/critic loops where you do not want to thread the name through every call site. ```typescript 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(), }); }); }); ``` Backed by `AsyncLocalStorage`. The nested-scope shape falls back gracefully on runtimes that do not support `AsyncLocalStorage` — only closure-scope works there. ## Cost attribution [#cost-attribution] `@pleach/observe` ships a tag-convention helper that populates the `cost.*` namespace on `ObserveRow.tags` for cross-tenant aggregation in dashboards. ```typescript 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 [#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. ```typescript 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: | Export | Shape | When to use | | --------------------------- | -------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `PIIRedactionConfig` | type | Type the `redact:` arg to `init`. | | `BUILT_IN_MATCHERS` | `readonly Matcher[]` | Spread alongside custom matchers. | | `DEFAULT_REPLACEMENT` | `string` | Replacement token used by built-ins (`"[REDACTED]"`). Override per matcher. | | `APPLY_REDACTION_POLICY_ID` | `string` | Stable policy identifier (`"observe.applyRedaction.v1"`). `applyRedaction` stamps it on every row it scrubs under the `APPLY_REDACTION_POLICY_TAG_KEY` (`redactionPolicyId`) tag, so downstream queries can attribute redactions (`WHERE tags->>'redactionPolicyId' = 'observe.applyRedaction.v1'`; surfaces in OTel as `pleach.observe.tag.redactionPolicyId`). | For full GDPR / HIPAA-grade redaction policy management — DSAR flows, consent revocation, regional residency rules — pair `@pleach/observe` with [`@pleach/compliance`](/docs/compliance). The `@pleach/recipes` `compliantChatbot` factory wires both together. The `redactor` / `redact` callbacks above stay buyer-controlled; `@pleach/observe` does not auto-apply any compliance policy. ## The `ObserveDestination` interface [#the-observedestination-interface] Custom destinations are first-class. The interface is intentionally minimal: ```typescript interface ObserveDestination { write(row: ObserveRow): Promise | void; flush?(): Promise; } ``` `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 [#api-surface--what-to-import-from-where] Three publishable subpaths (explicit, not glob): | Subpath | Public exports | | ------------------------------ | -------------------------------------------------------------------------------------------------------------------- | | `@pleach/observe` | `init`, `recordCall`, `getRecorder`, `subagent`, `costAttribution`, sampling + redaction surface, fingerprint helper | | `@pleach/observe/destinations` | `memory()`, `postgres()`, `supabase()`, `otel()` + their option types | | `@pleach/observe/types` | `ObserveRow`, `ObserveDestination`, `ObserveConfig`, `ObserveRecorder`, `CallClass`, `ProviderFamily` | `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 [#what-is-hookable-what-is-not] The SDK exposes a deliberate subset of the substrate. Five primitives are hookable from a brownfield agent loop: | Primitive | Hookable from SDK | | -------------------------------- | -------------------------------------------------------- | | `AuditableCall` row construction | Yes — `recordCall(row)`. | | `turnId` propagation | Yes — set on the row directly. | | Sub-agent attribution | Yes — `subagent(name)` (closure + ALS-scope). | | Fingerprint compute | Yes — `computeObserveFingerprint` re-export. | | PII redaction | Yes — `redactor` callback or `redact:` substrate policy. | 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](/docs/family-lock). The SDK observes calls; it does not make them. * **Reactive channels.** Channel subscription, fan-out, and back-pressure are [`@pleach/core`](/docs/channels)-internal. * **Replay determinism.** Replaying a recorded turn against a fresh runtime requires the [event log](/docs/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`](/docs/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` [#composing-with-pleachcore] 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](/docs/adoption-paths) page walks the migration shape end-to-end. ## License + versioning [#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](/docs/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 [#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 [#where-to-go-next]