Platform & operations recipes

End-to-end implementations for the platform-team patterns — long-running jobs, multi-step interrupts, per-call cost reporting, and OpenTelemetry wiring.

Four complete-enough-to-crib implementations covering the patterns a platform or ops team reaches for: long-running async jobs, multi-step approval interrupts, per-call cost reporting against the audit ledger, and OpenTelemetry wiring with in-process span access. Each links back to the reference page for the primitives it builds on.

For consumer-facing wiring (streaming chat, custom tools, custom storage, BYOK, moderation, multi-tenant, compliance), see Recipes.

1. Long-running async job tool

A tool that dispatches work to a queue, emits progress, and resolves when the job completes.

// lib/tools/runBenchmark.ts
import { defineTool } from "@pleach/core";

export const runBenchmark = defineTool({
  name: "run_benchmark",
  description: "Run a multi-hour benchmark; returns results when complete.",
  inputSchema: z.object({
    suite:  z.string(),
    config: z.record(z.unknown()),
  }),
  async execute(input, ctx) {
    const job = await queue.enqueue({
      type:   "benchmark",
      suite:  input.suite,
      config: input.config,
    });

    // Wait for terminal status. Honor abort.
    return new Promise((resolve, reject) => {
      const onComplete = (result) => resolve(result);
      const onFail     = (err) => reject(new Error(err.message));
      const onAbort    = () => {
        queue.cancel(job.id);
        reject(new Error("Aborted"));
      };

      ctx.signal?.addEventListener("abort", onAbort);
      queue.on(job.id, "complete", onComplete);
      queue.on(job.id, "fail", onFail);
    });
  },
});

Two patterns at play:

The tool returns a Promise that settles only on the queue's terminal complete/fail event — the runtime awaits the tool for the full job duration and surfaces the result when it lands.
The abort signal (ctx.signal) cancels both the runtime-side wait and the underlying queue job, so the user pressing stop reclaims the queue slot.

See Stream events for the job.* event shapes.

2. Multi-step interrupt flow (approve → edit → execute)

A tool that requests human approval for arguments before executing, with the option to edit.

const runtime = new SessionRuntime({
  interrupt: {
    enabled: true,
    perToolApproval: {
      requireApprovalFor: ["delete_resource", "send_email"],
      defaultDecision:    "pending",
    },
  },
  // ...
});

Drive the turn and resolve each approval request as it arrives. The stream pauses on interrupt.requested; call runtime.interrupts.resolve with an ApprovalDecision to resume:

import type { ApprovalDecision } from "@pleach/core";

for await (const event of runtime.executeMessage(sessionId, prompt)) {
  if (event.type === "interrupt.requested") {
    const { toolCall } = event.interrupt;

    // Show your modal for `toolCall.name` / `toolCall.arguments`,
    // then file the user's decision:
    const decision: ApprovalDecision = await showApprovalUI(toolCall);
    //   approve as-is:            { approved: true }
    //   approve with edited args: { approved: true, modifiedArguments: {...} }
    //   reject (tool won't fire): { approved: false }

    runtime.interrupts.resolve(event.interrupt.id, decision);
  }
}

resolve resumes the turn from where it paused. A tool approved as-is fires with its original args; modifiedArguments dispatch it with edited args; { approved: false } skips it. For a React component surface that tracks the pending interrupts as state, use useInterruptUI from @pleach/core/react (see Interrupts).

See Interrupts.

3. Per-call cost reporter

A ProviderDecisionLedger decorator that emits a per-call cost event for a billing pipeline.

import type { ProviderDecisionLedger, AuditableCall } from "@pleach/core/audit";
import { lookupModelCost } from "@pleach/core/query";

class CostEmittingLedger implements ProviderDecisionLedger {
  constructor(
    private primary: ProviderDecisionLedger,
    private emit:    (event: CostEvent) => void,
  ) {}

  async recordCall(call: AuditableCall): Promise<void> {
    const cost = lookupModelCost(call.call.model);
    if (call.tokenCost) {
      const usage = call.tokenCost;
      // `cost.input` / `cost.output` are USD per 1K tokens, so
      // cents = tokens * rate / 1000 * 100 = tokens * rate / 10.
      const cents = Math.ceil(
        (usage.inputTokens  * cost.input +
         usage.outputTokens * cost.output) / 10,
      );
      this.emit({
        sessionId: call.sessionId,
        turnId:    call.turnId,
        modelId:   call.call.model,
        cents,
        at:        call.createdAt,
      });
    }
    return this.primary.recordCall(call);
  }
}

Wire it in:

appRegistries.setProviderDecisionLedgerFactory(
  () => new CostEmittingLedger(
    new SupabaseProviderDecisionLedger(supabase),
    (event) => billing.recordCost(event),
  ),
);

Every LLM call produces a cost event before the row lands in the ledger. Idempotent on (sessionId, turnId, stageId, seqWithinTurn) — retries don't double-charge.

See Audit ledger.

4. OpenTelemetry wiring + reading spans in-process

A production OTel setup that exports the four substrate-emitted spans — session.turn, llm.invocation, graph.stage, tool.execution — plus the in-process runtime.spans path so tests and DevTools can read the trace tree without an external collector.

`lib/otel.ts`

// lib/otel.ts
import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
import { Resource } from "@opentelemetry/resources";
import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";

export function startOtel() {
  const sdk = new NodeSDK({
    resource: new Resource({
      [SemanticResourceAttributes.SERVICE_NAME]:
        process.env.OTEL_SERVICE_NAME ?? "agent-api",
      [SemanticResourceAttributes.SERVICE_VERSION]:
        process.env.GIT_SHA ?? "dev",
    }),
    traceExporter: new OTLPTraceExporter({
      url:     process.env.OTLP_ENDPOINT,
      headers: { "x-honeycomb-team": process.env.HONEYCOMB_API_KEY ?? "" },
    }),
  });

  sdk.start();
  return sdk;
}

Start the SDK before constructing any SessionRuntime. The pleach.tenant_id attribute is set on every span automatically when runtime.tenant is configured — no per-call wiring.

`lib/runtime.ts`

// lib/runtime.ts
import { SessionRuntime } from "@pleach/core";

export function buildRuntime(userId: string, tenantId: string) {
  return new SessionRuntime({
    storage:     pgAdapter,
    userId,
    tenantId,
    otelFlushIntervalTurns: 1,   // flush at the end of every turn
  });
}

otelFlushIntervalTurns: 1 drains the exporter buffer per turn — lowest tail latency for dev. Raise it (5, 10) under high-throughput production where per-turn flush cost shows up.

`app/api/dev/spans/route.ts`

// app/api/dev/spans/route.ts
import { buildRuntime } from "@/lib/runtime";

export async function GET(req: Request) {
  const runtime = buildRuntime("dev-user", "dev-tenant");
  const spans   = runtime.spans.snapshot();
  return Response.json({ spans });
}

runtime.spans.snapshot() returns the in-process buffer (last 1,000 spans by default) without round-tripping through an exporter. Useful for DevTools panels and integration tests that assert on the trace tree directly.

See OTel observability and Facets.

Platform & operations recipes

1. Long-running async job tool

2. Multi-step interrupt flow (approve → edit → execute)

3. Per-call cost reporter

4. OpenTelemetry wiring + reading spans in-process

`lib/otel.ts`

`lib/runtime.ts`

`app/api/dev/spans/route.ts`

Where to go next

Recipes

Observability

Audit ledger

Interrupts

Deployment

On this page