# Recipes (/docs/recipes) Patterns the gardener follows. Ten complete-enough-to-crib implementations covering the consumer-facing patterns most products land on: Next.js streaming chat, custom [tools](/docs/tools), custom [storage](/docs/storage), BYOK provider routing, moderation, multi-tenant, compliance, custom projections, hash-chain verification, and a regulated-host end-to-end. Each recipe links back to the reference page for the primitives it builds on. For platform-team patterns (long-running jobs, multi-step [interrupts](/docs/interrupts), per-call cost reporting, OTel), see [Platform & operations recipes](/docs/platform-recipes). | Recipe | Centers on | | -------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------- | | [1. Next.js App Router chat with streaming](#1-nextjs-app-router-chat-with-streaming) | `@pleach/core/react` + `HarnessProvider` | | [2. Custom tool: weather lookup with caching](#2-custom-tool-weather-lookup-with-caching) | `defineTool` + `ctx.signal` | | [3. Custom storage adapter: SQLite](#3-custom-storage-adapter-sqlite) | `StorageAdapter` + optimistic concurrency | | [4. BYOK: per-tenant provider credentials](#4-byok-per-tenant-provider-credentials) | Per-request `AnthropicSdkProvider` | | [5. Moderation pipeline with safety policies](#5-moderation-pipeline-with-safety-policies) | `contributeSafetyPolicies` + stream observers | | [6. Multi-tenant runtime with the runtime.tenant facet](#6-multi-tenant-runtime-with-the-runtimetenant-facet) | `tenantId` facet + RLS | | [7. Adopting `@pleach/compliance` — scrubbers + audit gates](#7-adopting-pleachcompliance--scrubbers--audit-gates) | `complianceProfilePlugin` + CI gates | | [8. Building a custom event-log projection](#8-building-a-custom-event-log-projection) | `GraphProjection` fold | | [9. Verifying the hash chain (manual SQL pass)](#9-verifying-the-hash-chain-manual-sql-pass) | `harness_event_log` recursive CTE | | [10. Regulated host: attestation + hash chain + scrubbers + audit ledger end-to-end](#10-regulated-host-attestation--hash-chain--scrubbers--audit-ledger-end-to-end) | `withAttestation` + `c9PhaseBEnabled` | ## 1. Next.js App Router chat with streaming [#1-nextjs-app-router-chat-with-streaming] The canonical pattern. SSE-streamed chat with persisted sessions, running on Fluid Compute. ### `lib/runtime.ts` [#libruntimets] ```typescript // lib/runtime.ts import { SessionRuntime, AiSdkProvider } from "@pleach/core"; import { SupabaseAdapter } from "@pleach/core/sessions"; import { SupabaseSaver } from "@pleach/core/checkpointing"; import { createOpenRouter } from "@openrouter/ai-sdk-provider"; const openrouter = createOpenRouter({ apiKey: process.env.OPENROUTER_API_KEY! }); import { createClient } from "@supabase/supabase-js"; export function buildRuntime(userId: string, orgId?: string) { const supabase = createClient( process.env.SUPABASE_URL!, process.env.SUPABASE_SERVICE_ROLE_KEY!, ); return new SessionRuntime({ storage: new SupabaseAdapter({ client: supabase }), checkpointer: new SupabaseSaver({ client: supabase }), provider: new AiSdkProvider({ model: openrouter("anthropic/claude-sonnet-4-5"), maxSteps: 5, }), userId, organizationId: orgId, metaToolNames: new Set(["set_step_complete", "wait_for_jobs"]), }); } ``` ### `app/api/chat/route.ts` [#appapichatroutets] ```typescript // app/api/chat/route.ts import { auth } from "@/lib/auth"; import { buildRuntime } from "@/lib/runtime"; export async function POST(req: Request) { const session = await auth.verifyRequest(req); if (!session) return new Response("Unauthorized", { status: 401 }); const { sessionId, content } = await req.json(); const runtime = buildRuntime(session.userId, session.orgId); const stream = new ReadableStream({ async start(controller) { try { for await (const event of runtime.executeMessage(sessionId, content)) { controller.enqueue(`data: ${JSON.stringify(event)}\n\n`); } } finally { controller.close(); } }, }); return new Response(stream, { headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-transform", "Connection": "keep-alive", "X-Accel-Buffering": "no", }, }); } ``` ### `app/(chat)/page.tsx` [#appchatpagetsx] ```tsx // app/(chat)/page.tsx "use client"; import { useMemo } from "react"; import { SessionRuntime } from "@pleach/core"; import { HarnessProvider, useHarness } from "@pleach/core/react"; export default function ChatPage({ userId }: { userId: string }) { const runtime = useMemo( () => new SessionRuntime({ // Client-side runtime is for streaming consumption; server // owns the actual storage. This client doesn't need a real // adapter — it reads via the API route. userId, }), [userId], ); return ( ); } function Chat() { const { messages, sendMessage, isLoading } = useHarness(); return (
{messages.map((m) => )}
); } ``` See [API routes](/docs/api-routes), [React](/docs/react), and [Deployment](/docs/deployment) for the underlying contracts. ## 2. Custom tool: weather lookup with caching [#2-custom-tool-weather-lookup-with-caching] A tool that hits an external API, honors the abort signal, and returns a typed result. ```typescript // lib/tools/lookupWeather.ts import { defineTool } from "@pleach/core"; import { z } from "zod"; const cache = new Map(); const TTL_MS = 5 * 60_000; export const lookupWeather = defineTool({ name: "lookup_weather", description: "Current weather conditions for a city.", inputSchema: z.object({ city: z.string().min(1), country: z.string().length(2).optional(), }), outputSchema: z.object({ tempC: z.number(), conditions: z.string(), humidity: z.number(), }), async execute(input, ctx) { const key = `${input.city}|${input.country ?? ""}`.toLowerCase(); const cached = cache.get(key); if (cached && Date.now() - cached.ts < TTL_MS) { return cached.value; } const url = new URL("https://api.example.com/weather"); url.searchParams.set("q", input.city); if (input.country) url.searchParams.set("country", input.country); const res = await fetch(url, { headers: { "X-API-Key": process.env.WEATHER_API_KEY! }, signal: ctx.signal, }); if (!res.ok) { throw new Error(`Weather API ${res.status}: ${res.statusText}`); } const data = await res.json(); const value = { tempC: data.temperature, conditions: data.description, humidity: data.humidity, }; cache.set(key, { ts: Date.now(), value }); return value; }, }); ``` Three things worth noting: 1. The cache is a process-local Map — fine for a single-instance deployment; for horizontal scale, swap in Redis. 2. `ctx.signal` threads into the `fetch` call. The user pressing stop aborts the HTTP request. 3. The output schema validates the return shape. A misbehaving upstream that returns malformed data trips this validator, surfacing as `tool.failed` instead of poisoning the conversation. See [Tools](/docs/tools) for the full `defineTool` contract. ## 3. Custom storage adapter: SQLite [#3-custom-storage-adapter-sqlite] A storage adapter that persists to SQLite via `better-sqlite3`. Useful for single-instance deployments and desktop apps (Electron, Tauri). ```typescript // lib/storage/sqliteAdapter.ts import Database from "better-sqlite3"; import type { StorageAdapter, SessionState } from "@pleach/core"; export class SqliteAdapter implements StorageAdapter { private db: Database.Database; constructor(opts: { path: string }) { this.db = new Database(opts.path); this.db.exec(` CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, organization_id TEXT, state TEXT NOT NULL, version INTEGER NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id, updated_at DESC); `); } async createSession(state: SessionState): Promise { this.db.prepare(` INSERT INTO sessions (id, user_id, organization_id, state, version, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) `).run( state.id, state.userId, state.organizationId ?? null, JSON.stringify(state), state.version, state.createdAt.getTime(), state.updatedAt.getTime(), ); } async getSession(sessionId: string): Promise { const row = this.db .prepare("SELECT state FROM sessions WHERE id = ?") .get(sessionId) as { state: string } | undefined; if (!row) return null; return reviveDates(JSON.parse(row.state)); } async updateSession(sessionId: string, state: SessionState): Promise { const result = this.db.prepare(` UPDATE sessions SET state = ?, version = ?, updated_at = ? WHERE id = ? AND version = ? `).run( JSON.stringify(state), state.version, state.updatedAt.getTime(), sessionId, state.version - 1, // optimistic-concurrency check ); if (result.changes === 0) { const err = new Error("Version conflict") as Error & { code: string }; err.code = "3001"; throw err; } } async deleteSession(sessionId: string): Promise { this.db.prepare("DELETE FROM sessions WHERE id = ?").run(sessionId); } async listSessions(filter: { userId: string; limit?: number }): Promise { const rows = this.db.prepare(` SELECT state FROM sessions WHERE user_id = ? ORDER BY updated_at DESC LIMIT ? `).all(filter.userId, filter.limit ?? 50) as Array<{ state: string }>; return rows.map((r) => reviveDates(JSON.parse(r.state))); } // ... event log + checkpoint methods follow the same pattern } function reviveDates(obj: any): SessionState { obj.createdAt = new Date(obj.createdAt); obj.updatedAt = new Date(obj.updatedAt); obj.lastActiveAt = new Date(obj.lastActiveAt); return obj; } ``` Three load-bearing details: 1. The `version` check on `updateSession` enforces optimistic concurrency — the same protocol the Supabase adapter uses. 2. Version conflicts throw with `code: "3001"` so the runtime's [sync](/docs/sync) layer surfaces them through the standard error path. 3. `reviveDates` is necessary because JSON doesn't preserve Date objects. Every storage adapter has this concern. See [Storage](/docs/storage) for the full interface. ## 4. BYOK: per-tenant provider credentials [#4-byok-per-tenant-provider-credentials] A runtime build function that resolves the provider per-request from the tenant's stored credentials. ```typescript // lib/runtime.ts import { SessionRuntime, AnthropicSdkProvider, AiSdkProvider } from "@pleach/core"; interface TenantConfig { providerType: "anthropic" | "openai" | "default"; apiKey?: string; // tenant-supplied (BYOK) model?: string; } export async function buildRuntimeForTenant(req: AuthedRequest) { const tenant = await loadTenant(req.user.orgId); const provider = pickProvider(tenant.config); return new SessionRuntime({ storage: sharedStorage, checkpointer: sharedCheckpointer, provider, userId: req.user.id, organizationId: req.user.orgId, }); } function pickProvider(config: TenantConfig) { if (config.providerType === "anthropic" && config.apiKey) { return new AnthropicSdkProvider({ apiKey: config.apiKey, // tenant's key model: config.model ?? "claude-sonnet-4-5", }); } if (config.providerType === "openai" && config.apiKey) { return new AiSdkProvider({ model: openai("gpt-4o", { apiKey: config.apiKey }), }); } // Fallback to the platform's shared credentials. return new AiSdkProvider({ model: openrouter("anthropic/claude-sonnet-4-5"), maxSteps: 5, }); } ``` The tenant's API key never appears in the [audit ledger](/docs/audit-ledger) (which records `family` / `modelId` / `transport`, not the credential) and never reaches the browser bundle (runtime construction is server-side). For more sophisticated routing — per-call key selection, rate limit management — `@pleach/gateway@0.1.0` ships the Phase A surface (`GatewayClient.route`, BYOK fingerprint, family-locked failover); other transports are deferred to Phase B per [Gateway](/docs/gateway). DIY routing or a third-party gateway remain viable for unsupported families. See [Multi-tenant](/docs/multi-tenant) for the full pattern. ## 5. Moderation pipeline with safety policies [#5-moderation-pipeline-with-safety-policies] A plugin that ships two safety policies (input scan and output scan) plus a stream observer that halts on policy violation. ```typescript // lib/plugins/moderation.ts import type { HarnessPlugin } from "@pleach/core"; const FORBIDDEN_PATTERNS = [ /\bcredit\s*card\s*number\b/i, /\bssn\b/i, // ... your patterns ]; export const moderationPlugin: HarnessPlugin = { name: "moderation", contributeSafetyPolicies: () => [ { id: "moderation.input-scan", version: "1.0.0", framework: "internal", enforcementLevel: "hard-gate", content: "Refuse requests asking to expose payment or identifier data.", preDispatchCheck: (msg) => { const matched = FORBIDDEN_PATTERNS.find((p) => p.test(msg.content)); return matched ? { verdict: "refuse", reason: `Pattern ${matched} detected in input` } : { verdict: "ok" }; }, }, { id: "moderation.output-scan", version: "1.0.0", framework: "internal", enforcementLevel: "hard-gate", content: "Never include credit card numbers or SSNs in responses.", postCompletionCheck: (msg) => { const matched = FORBIDDEN_PATTERNS.find((p) => p.test(msg.content)); return matched ? { verdict: "refuse", reason: `Pattern ${matched} detected in output` } : { verdict: "ok" }; }, }, ], contributeStreamObservers: () => [{ when: { callClass: "*" }, factory: () => ({ observerId: "redact-streaming", onChunk(chunk) { // SeamStreamEvent is { kind, payload } — the text rides in payload. if (chunk.kind !== "content_delta") return { kind: "continue" }; const { text } = chunk.payload as { text: string }; const redacted = FORBIDDEN_PATTERNS.reduce( (t, pattern) => t.replace(pattern, "[REDACTED]"), text, ); return redacted === text ? { kind: "continue" } : { kind: "amend", chunk: { kind: "content_delta", payload: { text: redacted } } }; }, }), }], }; ``` Register and enable: ```typescript const runtime = new SessionRuntime({ plugins: [moderationPlugin], enabledSafetyPolicies: [ "moderation.input-scan", "moderation.output-scan", ], // ... }); ``` The two policies fire at different lifecycle points — `preDispatchCheck` before the LLM sees the input, `postCompletionCheck` after the LLM produces output. The stream observer redacts in-flight chunks so the user never sees the offending text even before the post-completion gate fires. See [Safety policies](/docs/safety) and [Plugin contract](/docs/plugin-contract). ## 6. Multi-tenant runtime with the runtime.tenant facet [#6-multi-tenant-runtime-with-the-runtimetenant-facet] A Next.js API route that constructs a `SessionRuntime` per-tenant. `tenantId` flows to the [event log](/docs/event-log), audit ledger, OTel spans, and outbound HTTP without re-threading. RLS at the database layer is the enforcement; the facet provides the value. ### `lib/runtime.ts` [#libruntimets-1] ```typescript // lib/runtime.ts import { SessionRuntime } from "@pleach/core"; export function buildRuntime( userId: string, tenantId: string, subTenantId?: string, ) { return new SessionRuntime({ storage: pgAdapter, userId, tenantId, subTenantId, }); } ``` `tenantId` rejects the empty string at construction with `TenantIdEmptyError` — an unset env var interpolated as `""` throws at init rather than silently leaking writes across tenants months later. ### `app/api/chat/route.ts` [#appapichatroutets-1] ```typescript // app/api/chat/route.ts import { auth } from "@/lib/auth"; import { buildRuntime } from "@/lib/runtime"; export async function POST(req: Request) { const session = await auth.verifyRequest(req); if (!session) return new Response("Unauthorized", { status: 401 }); const tenantId = session.tenantId; const subTenantId = session.teamId; const runtime = buildRuntime(session.userId, tenantId, subTenantId); const { sessionId, content } = await req.json(); const stream = new ReadableStream({ async start(controller) { try { for await (const event of runtime.executeMessage(sessionId, content)) { controller.enqueue(`data: ${JSON.stringify(event)}\n\n`); } } finally { controller.close(); } }, }); return new Response(stream, { headers: { "Content-Type": "text/event-stream" }, }); } ``` ### `db/policies.sql` [#dbpoliciessql] ```sql ALTER TABLE harness_event_log ENABLE ROW LEVEL SECURITY; CREATE POLICY tenant_isolation ON harness_event_log USING (tenant_id = current_setting('app.tenant_id', true)::text); ``` The `true` flag on `current_setting` returns `NULL` instead of erroring when the setting is unset — the policy then refuses every row, which is the correct failure mode for an unauthenticated connection. Set `app.tenant_id` on session start from the JWT claim. ### `lib/outbound.ts` [#liboutboundts] ```typescript // lib/outbound.ts import { withTenantHeader } from "@pleach/core/tenant"; export function tenantFetch(tenantId: string) { return withTenantHeader(fetch, { header: "x-tenant-id", tenantId, }); } // Upstream gateway calls inherit the supplied tenant. await tenantFetch(req.tenantId)("https://gateway.internal/v1/chat", { method: "POST", body: JSON.stringify(payload), }); ``` The wrapper writes the configured header on every request the returned `fetch` makes; per-call overrides win when the caller sets the same header explicitly. The host passes `req.tenantId` once at construction — typically inside a tool handler that already has the request-scoped runtime context in scope, not at wrap time — a runtime whose tenant is set late still produces a correctly-stamped header. See [Tenant facet](/docs/tenant-facet), [Multi-tenant](/docs/multi-tenant), [Facets](/docs/facets), and [OTel observability](/docs/otel-observability). ## 7. Adopting `@pleach/compliance` — scrubbers + audit gates [#7-adopting-pleachcompliance--scrubbers--audit-gates] A regulated-domain agent that adopts `@pleach/compliance`, wires the four bundled scrubbers, registers a custom `KeyedRegex` for an MRN pattern, and inherits the two CI gates that fail the build when tenant scoping drifts. ### `package.json` [#packagejson] ```json { "dependencies": { "@pleach/core": "^1.1.0", "@pleach/compliance": "^1.0.0" } } ``` ### `lib/compliance.ts` [#libcompliancets] ```typescript // lib/compliance.ts import { KeyedRegexScrubber, complianceProfilePlugin, } from "@pleach/compliance"; import type { HarnessPlugin } from "@pleach/core"; // The HIPAA profile bundle (SSN_US + credit card + US_DL) plus an // MRN scrubber layered on top — composition is additive at construction. export const hipaaPlugin = complianceProfilePlugin("hipaa", { extraScrubbers: [ new KeyedRegexScrubber({ id: "mrn", entries: [ { name: "MRN", pattern: /\b\d{7,10}\b/g, replacement: "[MRN-REDACTED]" }, ], }), ], }); // A second freestanding plugin for site-specific patterns. The runtime // stacks both at registration time; per-plugin facets compose without // either side knowing about the other. const customMrn = new KeyedRegexScrubber({ id: "site-mrn", entries: [ { name: "site-MRN", pattern: /\bMRN-[A-Z]{2}\d{6}\b/g, replacement: "[MRN-REDACTED]" }, ], }); export const customScrubberPlugin: HarnessPlugin = { name: "site-mrn", contributeScrubbers: () => [customMrn], }; ``` `complianceProfilePlugin("hipaa")` returns a single plugin that contributes the profile's standard scrubber bundle (`SSN_US`, credit-card, `US_DL`) against the standard event-type allowlist. `extraScrubbers` layers additional `KeyedRegexScrubber` instances on top in the same plugin; a separately-registered second plugin demonstrates the stack-multiple-plugins pattern. ### `lib/runtime.ts` [#libruntimets-2] ```typescript // lib/runtime.ts import { SessionRuntime } from "@pleach/core"; import { hipaaPlugin, customScrubberPlugin } from "./compliance"; export function buildRuntime(userId: string, tenantId: string) { return new SessionRuntime({ storage, checkpointer, userId, tenantId, plugins: [hipaaPlugin, customScrubberPlugin], }); } ``` ### `.github/workflows/audit.yml` [#githubworkflowsaudityml] ```yaml name: audit on: [pull_request] jobs: audit: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-node@v4 with: { node-version: 22 } - run: npm ci - run: npm run audit:tenant-scoping - run: npm run audit:harness-event-log-tenant-id-required ``` The first gate scans source for write sites that bypass the facet; the second checks the applied schema for the `NOT NULL` constraint on `harness_event_log.tenant_id`. Together they close the loop: source-level catches code that would write `NULL`, schema-level catches schemas that would accept it. See [Compliance](/docs/compliance), [Scrubbers](/docs/scrubbers), [Plugin contract](/docs/plugin-contract), and [Tenant facet](/docs/tenant-facet). ## 8. Building a custom event-log projection [#8-building-a-custom-event-log-projection] A custom `GraphProjection` that counts tool failures grouped by `toolName` across a session. Useful for dashboards that highlight tools degrading mid-conversation. ### `lib/projections/toolFailureCounts.ts` [#libprojectionstoolfailurecountsts] ```typescript // lib/projections/toolFailureCounts.ts import type { GraphProjection, EventLogRow } from "@pleach/core/eventLog"; export interface ToolFailureCounts { readonly byTool: Readonly>; } export const toolFailureCountsProjection: GraphProjection = { name: "tool-failure-counts", initial: () => ({ byTool: {} }), reduce(state, row) { if (row.event_type !== "tool.failed") return state; const toolName = (row.payload as { toolName?: string }).toolName ?? "unknown"; return { byTool: { ...state.byTool, [toolName]: (state.byTool[toolName] ?? 0) + 1, }, }; }, }; ``` The early `event_type` guard keeps the fold tight — only `tool.failed` rows mutate state (`reduce` returns the same `state` reference otherwise) — so the projection cost scales with failures, not with total event volume. ### `app/api/sessions/[id]/health/route.ts` [#appapisessionsidhealthroutets] ```typescript // app/api/sessions/[id]/health/route.ts import { buildRuntime } from "@/lib/runtime"; import { toolFailureCountsProjection } from "@/lib/projections/toolFailureCounts"; export async function GET(req: Request, { params }: { params: { id: string } }) { const runtime = buildRuntime(/* … */); const counts = await runtime.events.fold(toolFailureCountsProjection, { sessionId: params.id, }); return Response.json({ byTool: counts.byTool }); } ``` Projections compute on demand from the event log. The substrate already persists every event — a materialized projection table would duplicate that storage and drift the moment a backfill or correction landed only in one place. Re-deriving from the row stream keeps the answer consistent with what actually happened. See [Event-log projections](/docs/event-log-projections), [Event log](/docs/event-log), and [Stream events](/docs/stream-events). ## 9. Verifying the hash chain (manual SQL pass) [#9-verifying-the-hash-chain-manual-sql-pass] A manual verification of the `harness_event_log` hash chain for a single tenant. Writer-side stamping is in soak behind the `c9PhaseBEnabled` flag. Programmatic verification ships in `@pleach/replay@0.1.0` as `verifyChainForChat(chatId)` and `generateProof(chatId, range)` — see [Hash chain](/docs/hash-chain). The SQL pattern below is the manual fallback for ad-hoc audits or hosts that haven't adopted the SKU. ### `db/verify-chain.sql` [#dbverify-chainsql] ```sql -- Walk a tenant's stamped slice in (tenant_id, created_at) order and -- report the first index where prev_hash diverges from the previous -- row's row_hash. compute_canonical_hash is a placeholder for the -- writer's canonical encoding helper — the canonical encoding is -- defined in @pleach/core's src/event-log/ and is the authoritative -- source. This SQL is illustrative, not a drop-in. WITH RECURSIVE chain AS ( SELECT record_id, tenant_id, created_at, prev_hash, row_hash, compute_canonical_hash(harness_event_log.*) AS recomputed_hash, LAG(row_hash) OVER ( PARTITION BY tenant_id ORDER BY created_at, record_id ) AS expected_prev_hash, ROW_NUMBER() OVER ( PARTITION BY tenant_id ORDER BY created_at, record_id ) AS idx FROM harness_event_log WHERE tenant_id = $1 AND row_hash IS NOT NULL ) SELECT idx, record_id, created_at, prev_hash, expected_prev_hash, row_hash, recomputed_hash FROM chain WHERE prev_hash IS DISTINCT FROM expected_prev_hash OR row_hash IS DISTINCT FROM recomputed_hash ORDER BY idx LIMIT 1; ``` The query returns zero rows when the chain verifies clean. When a row comes back, `idx` is the first break — either `prev_hash` doesn't match the previous row's `row_hash` (a reorder, a backfill, or a deletion upstream) or `row_hash` doesn't match the recomputed canonical hash (the row's persisted bytes were mutated after write). A verification break flags the tenant for manual review against backups. Don't auto-rewrite history — the chain's job is to make tampering visible, and a self-healing writer that quietly restamps every break defeats the proof. Open a ticket, pull the last clean snapshot, and reconcile the divergence by hand. Pre-hash rows — those written before the writer-side stamping rolled out for the tenant — carry `NULL` in both columns. The `WHERE row_hash IS NOT NULL` filter skips them; chain verification resumes at the first stamped row. See [Hash chain](/docs/hash-chain), [Audit ledger](/docs/audit-ledger), and [Eval and replay](/docs/eval-and-replay). ## 10. Regulated host: attestation + hash chain + scrubbers + audit ledger end-to-end [#10-regulated-host-attestation--hash-chain--scrubbers--audit-ledger-end-to-end] Regulated domains — healthcare, finance, government — need four properties at once: row-level signing so any post-hoc mutation is detectable, tamper-evident chaining so deletion or reordering is detectable, PII redaction so the event log can be stored without violating data-handling policies, and a portable proof an auditor can verify offline. Each has a dedicated subsystem in `@pleach/core` plus sibling SKUs; this recipe wires all four together. ### `lib/attestation-setup.ts` [#libattestation-setupts] ```typescript // lib/attestation-setup.ts import { AwsKmsAttestationKeyStore, type AttestationKeyStore, } from "@pleach/core/attestation"; import { FileBackedAttestationKeyStore } from "@pleach/core/attestation/testing"; export function buildKeyStore(): AttestationKeyStore { if (process.env.NODE_ENV === "production") { return new AwsKmsAttestationKeyStore({ region: process.env.AWS_REGION!, keyArn: process.env.ATTESTATION_KEY_ARN!, }); } // Dev / test — 32-byte ed25519 seed on disk. return new FileBackedAttestationKeyStore({ privateKeyPath: process.env.ATTESTATION_PRIVATE_KEY_PATH!, keyId: process.env.ATTESTATION_KEY_ID ?? "dev-key-1", }); } ``` The production constructor is a stub at v0.9 (interface conformance only; `sign()` throws `NotImplementedError`); the `FileBackedAttestationKeyStore` from the `/testing` subpath is the real implementation that wraps `@noble/curves/ed25519`. ### `lib/compliance-plugin.ts` [#libcompliance-plugints] ```typescript // lib/compliance-plugin.ts import type { HarnessPlugin } from "@pleach/core"; import { SsnUsScrubber, CreditCardScrubber, UsDriverLicenseScrubber, KeyedRegexScrubber, } from "@pleach/compliance"; export const compliancePlugin: HarnessPlugin = { name: "compliance", contributeScrubbers: () => [ new SsnUsScrubber(), new CreditCardScrubber(), new UsDriverLicenseScrubber(), new KeyedRegexScrubber({ entries: [ // Custom MRN pattern for the healthcare slice. { key: "MRN", pattern: /\b\d{7,10}\b/g }, ], }), ], }; ``` `@pleach/compliance` does not ship a pre-bundled `compliancePlugin` export at v0.9 — the host assembles the plugin from the four concrete scrubber classes. Composition is additive: stack additional `KeyedRegexScrubber` instances for site-specific patterns, or layer a second plugin alongside this one. ### `lib/attesting-ledger.ts` [#libattesting-ledgerts] ```typescript // lib/attesting-ledger.ts import { MemoryProviderDecisionLedger, type AuditableCall, type ProviderDecisionLedger, } from "@pleach/core/audit"; import { signAttestation, computePayloadHash } from "@pleach/core/attestation"; import { canonicalizeAttestationPayload } from "@pleach/core/attestation"; import type { AttestationKeyStore } from "@pleach/core/attestation"; export function withAttestation( inner: ProviderDecisionLedger, keyStore: AttestationKeyStore, keyId: string, ): ProviderDecisionLedger { return { async recordCall(call: AuditableCall) { // Build the unsigned payload from the call's canonical fields, // sign it, and attach the signature to the call before // forwarding to the underlying ledger. Payload shape and field // list are defined by AttestationPayloadV1 in @pleach/core. const payload = buildUnsignedPayload(call, keyId); const signed = await signAttestation(payload, keyStore); await inner.recordCall({ ...call, attestation: signed }); }, }; } ``` The wrapper composes around any `ProviderDecisionLedger` — the illustrative `MemoryProviderDecisionLedger` here, a Supabase-backed adapter in production, or an S3-backed sink. The signature lands in the persisted row alongside the canonical payload; verification re-canonicalizes the same fields and checks the signature against the public key resolved from `signaturePublicKeyId`. ### `app/api/harness/route.ts` [#appapiharnessroutets] ```typescript // app/api/harness/route.ts import { SessionRuntime, EventLogWriter } from "@pleach/core"; import { MemoryProviderDecisionLedger } from "@pleach/core/audit"; import { buildKeyStore } from "@/lib/attestation-setup"; import { compliancePlugin } from "@/lib/compliance-plugin"; import { withAttestation } from "@/lib/attesting-ledger"; export async function POST(req: Request) { const { userId, tenantId, chatId, content } = (await req.json()) as { userId: string; tenantId: string; chatId: string; content: string; }; const keyStore = buildKeyStore(); const ledger = withAttestation( new MemoryProviderDecisionLedger(), keyStore, process.env.ATTESTATION_KEY_ID ?? "dev-key-1", ); const eventLog = new EventLogWriter(supabase, { c9PhaseBEnabled: true, // hash-chain stamping on }); const runtime = new SessionRuntime({ storage, eventLogWriter: eventLog, // hash-chain-stamping writer userId, tenantId, // threads to chain + ledger plugins: [compliancePlugin], }); // The attesting `ledger` composed above is wired at the audit adapter / // route layer — audit capture is route-scoped, not a SessionRuntime config // field. // …executeMessage(chatId, content)… } ``` `c9PhaseBEnabled` must be on at write time, or the chain is sparse — pre-existing rows carry `NULL` in chain columns and verification will skip them. `tenantId` flows from the auth layer through to the event log row, the chain partition, the attestation payload's identity fields, and the audit ledger row. ### Verifier flow [#verifier-flow] The auditor's offline check fans out across the three layers: ```typescript import { gatherChainEntries, generateProof } from "@pleach/core/eventLog"; import { verifyAttestation } from "@pleach/core/attestation"; // 1. Walk the chat's chain once — gatherChainEntries returns BOTH the // (sequenceNumber, rowHash) entries a proof needs AND the verify verdict. // A clean verdict proves no rows were reordered, deleted, or mutated. const { entries, verifyResult } = await gatherChainEntries({ reader, tenantId, chatId, }); if (verifyResult.ok === false) { // 2. Chain broke. `failedIndex` is on the failure branch — pull the affected // window and re-verify each row's attestation in turn to localize. for await (const row of reader.iterateChatEvents({ tenantId, chatId, fromSequenceNumber: verifyResult.failedIndex, })) { const ok = await verifyAttestation(row.attestation, publicKey); if (!ok) console.error("attestation break at seq", row.sequenceNumber); } } // 3. For a clean range, generate a portable proof an external auditor can // verify without DB access. The head is the last gathered entry's sequence. const head = entries[entries.length - 1].sequenceNumber; const proof = generateProof({ entries, tenantId, chatId, headSequenceNumber: head, leafSequenceNumber: head, }); ``` `gatherChainEntries` walks the chain once — `verifyResult` is the broad-stroke "did anything change?" verdict and `entries` feed the proof; per-row `verifyAttestation` localizes which specific row(s) diverged; `generateProof` produces a `ChainProofV1` the auditor verifies against the published public key + the canonical encoding, no database needed. ### Caveats [#caveats] * `c9PhaseBEnabled` must be on at write time. Rows written with the flag off carry `NULL` in `prev_hash` and `row_hash`; the verifier skips them as the legacy prefix, and the chain only begins at the first stamped row. * Pre-existing rows carry `NULL` in the chain columns and read as the legacy prefix — `verifyChainForChat` accepts this and starts checking at `firstChainedSequenceNumber`. * Attestation signs the canonical payload hash. Rotating the keystore invalidates old signatures unless the rotation policy preserves the previous key for verification — keep the `getPublicKey(keyId)` resolver covering every keyId that's ever signed a row in the retention window. * Scrubbing happens **before** the event-log row is stamped — the chain hashes the redacted view. This is intentional: verifying the chain doesn't need access to the original PII, and storing the unredacted bytes anywhere would defeat the scrubber. * The `MemoryProviderDecisionLedger` here is illustrative. In production wrap a persistent ledger (Supabase, S3-backed); the `withAttestation` wrapper composes around either. * Tombstoning via the `GDPRSoftDelete` plug-point doesn't break the chain — `recordId` stays in place; identifying fields clear under the subject key. See [The AuditableCall row](/docs/auditable-call-row) for the soft-delete shape. See [Attestation](/docs/attestation), [Hash chain](/docs/hash-chain), [Scrubbers](/docs/scrubbers), [Audit ledger](/docs/audit-ledger), and [Compliance](/docs/compliance). ## Where to go next [#where-to-go-next]