Recipes

// 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
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
"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 (
    <HarnessProvider runtime={runtime} apiBase="/api/chat">
      <Chat />
    </HarnessProvider>
  );
}

function Chat() {
  const { messages, sendMessage, isLoading } = useHarness();
  return (
    <div>
      {messages.map((m) => <Bubble key={m.id} message={m} />)}
      <Composer onSend={sendMessage} disabled={isLoading} />
    </div>
  );
}

// lib/tools/lookupWeather.ts
import { defineTool } from "@pleach/core";
import { z } from "zod";

const cache = new Map<string, { ts: number; value: WeatherResult }>();
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;
  },
});

// 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<void> {
    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<SessionState | null> {
    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<void> {
    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<void> {
    this.db.prepare("DELETE FROM sessions WHERE id = ?").run(sessionId);
  }

  async listSessions(filter: { userId: string; limit?: number }): Promise<SessionState[]> {
    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;
}

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

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

const runtime = new SessionRuntime({
  plugins: [moderationPlugin],
  enabledSafetyPolicies: [
    "moderation.input-scan",
    "moderation.output-scan",
  ],
  // ...
});

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

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

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);

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

{
  "dependencies": {
    "@pleach/core":       "^1.1.0",
    "@pleach/compliance": "^1.0.0"
  }
}

// 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],
};

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

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

// lib/projections/toolFailureCounts.ts
import type { GraphProjection, EventLogRow } from "@pleach/core/eventLog";

export interface ToolFailureCounts {
  readonly byTool: Readonly<Record<string, number>>;
}

export const toolFailureCountsProjection: GraphProjection<ToolFailureCounts> = {
  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,
      },
    };
  },
};

// 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 });
}

-- 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;

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

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

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

// 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)…
}

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