Gateway · Cost events

Every successful gateway call emits one CostEvent to the configured destination. The cadence is per-call, not batched: accuracy and reproducibility for compliance attestation outweigh the batching savings. A destination that wants its own batching implements it internally — and the four reference adapters that follow do exactly that.

CostEmitter contract

interface CostEmitter {
  emit(event: CostEvent): Promise<void> | void
  flush?(): Promise<void>
}

CostEmitter widens the existing CostEventSink primitive used by the Phase A GatewayClient — it adds an optional flush() for buffered adapters. Every adapter is structurally a CostEventSink too: code that holds a CostEmitter can pass it anywhere a CostEventSink is expected.

Implementations MUST:

• Never throw synchronously from emit() — failures inside an adapter MUST NOT propagate up the gateway call. Buffer + log, or swallow with an internal counter.
• Tolerate being called from the host's async context. The middleware does not wrap the call in try/catch; adapters own their own failure surface.

Implementations MAY:

• Return a Promise from emit() to await a remote write.
• Return void from emit() if they buffer internally and flush asynchronously (Postgres pool, Honeycomb batch).
• Implement flush() if they buffer. Callers wire flush() into their shutdown hook to drain pending events before exit.

CostMiddleware factory

The wire-up. Takes a CostEmitter plus a computeCost mapper and returns a middleware function callable after each route response.

computeCost is generic over the observation shape you pass it. The output of runtime.routeChatCompletion(input) carries only { providerInvoked, modelInvoked, content, usage, costUsd, routingDecisionId } — it does not carry tenantId, family, callClass, byokActive, or the full routingDecision. Those come from the request input plus your own routing context, so merge them onto the observation at the call site and read them in computeCost:

import { createCostMiddleware, createOtelCostEmitter } from "@pleach/gateway/cost"
import { asTenantId } from "@pleach/gateway"
import type {
  RouteChatCompletionInput,
  RouteChatCompletionOutput,
} from "@pleach/gateway"

// The merged observation: route output + the fields the host supplies.
type CostObservation = RouteChatCompletionOutput & {
  tenantId: string
  family: RouteChatCompletionInput["family"]
  callClass: "synthesize" | "reasoning" | "converse" | "utility"
  byokActive: boolean
}

const middleware = createCostMiddleware<CostObservation>({
  emitter: createOtelCostEmitter({ meter }),
  computeCost: (route, response) => ({
    type: "domain.gateway.cost.recorded",
    tenantId: asTenantId(response.tenantId),   // non-empty: sourced from input.tenantId
    family: response.family,
    callClass: response.callClass,
    modelInvoked: response.modelInvoked,
    costUsd: response.costUsd,
    promptTokens: response.usage.promptTokens,
    completionTokens: response.usage.completionTokens,
    byokActive: response.byokActive,
    routingDecision: {
      family: response.family,
      call_class: response.callClass,
      selected_model: response.modelInvoked,
      transport: "openrouter",
      byok_active: response.byokActive,
      upgrade_path_invoked: false,
      fallback_chain: [],
      raw_provider_cost_usd: response.costUsd,
      markup_pct: 0,
    },
    timestamp: new Date().toISOString(),
  }),
})

const output = await runtime.routeChatCompletion(input)
void middleware({
  route: "chat.completion",
  response: {
    ...output,
    tenantId: input.tenantId,
    family: input.family,
    callClass: "converse",   // host routing context — not on the route output
    byokActive: false,       // host routing context — not on the route output
  },
})

routeChatCompletion is the Pack-221 slice-2 stub today: it validates input and identifies the target provider but returns costUsd: 0 and a stub content until the real transport lands. Wire the cost path now so it's ready; the cost numbers go live with the transport.

The route field is a free-form string identifier — typically "chat.completion", "embedding", or a tenant-scoped slug. The middleware does not interpret it; it forwards verbatim to computeCost.

computeCost MAY return null to skip emission for a given observation (e.g. cached responses where no provider cost was incurred). Returning null is structurally identical to a buffered no-op; do not throw to skip — throw indicates a real error.

onError is optional but recommended. Default behavior when computeCost throws OR emitter.emit rejects is to swallow — gateway calls MUST NOT fail because the cost destination is unavailable. Set onError to forward to your observability layer.

The factory returns middleware & { flush }, so middleware.flush() on shutdown drains buffered adapters before exit.

Reference adapters

Four adapters cover the common cost destinations. All are structurally duck-typed against the host's existing client (pg.Pool, @opentelemetry/api Meter, global fetch) so no vendor SDK is a hard dependency.

OTel

import { metrics } from "@opentelemetry/api"
import { createOtelCostEmitter } from "@pleach/gateway/cost"

const meter = metrics.getMeter("@pleach/gateway")
const emitter = createOtelCostEmitter({ meter })

Emits two counters per CostEvent:

• pleach_gateway_cost_usd (Counter, double) — incremented by event.costUsd.
• pleach_gateway_tokens_total (Counter, long) — incremented by event.promptTokens + event.completionTokens, with attribute token.kind = "prompt" | "completion" recorded once per token-class so the dimension is preserved downstream.

Attributes on every measurement: tenant.id, provider.family, call.class, model.invoked, byok.active, cost.event.type.

The OTel SDK is an optional peer. The factory accepts a meter argument duck-typed against @opentelemetry/api's Meter shape; the adapter never imports the SDK at module-eval time. Override counter names via costCounterName / tokensCounterName.

Postgres

import { Pool } from "pg"
import { createPostgresCostEmitter } from "@pleach/gateway/cost"

const pool = new Pool({ connectionString: process.env.DATABASE_URL })
const emitter = createPostgresCostEmitter({
  pool,
  table: "pleach_gateway_cost_events",
  batchSize: 100,
  flushIntervalMs: 5000,
})

Buffered. emit() pushes into an in-memory buffer and returns immediately; flush is either timer-driven (default 5s), threshold- driven (default 100 events), or manual via flush().

Suggested schema (the byok_key_hash column is nullable; everything else is NOT NULL):

CREATE TABLE pleach_gateway_cost_events (
  id                BIGSERIAL PRIMARY KEY,
  event_type        TEXT        NOT NULL,
  tenant_id         TEXT        NOT NULL,
  family            TEXT        NOT NULL,
  call_class        TEXT        NOT NULL,
  model_invoked     TEXT        NOT NULL,
  cost_usd          NUMERIC(18, 8) NOT NULL,
  prompt_tokens     INTEGER     NOT NULL,
  completion_tokens INTEGER     NOT NULL,
  byok_active       BOOLEAN     NOT NULL,
  byok_key_hash     TEXT,
  routing_decision  JSONB       NOT NULL,
  ts                TIMESTAMPTZ NOT NULL
);
CREATE INDEX ON pleach_gateway_cost_events (tenant_id, ts DESC);
CREATE INDEX ON pleach_gateway_cost_events (family, call_class);

tenant_id is the canonical multi-tenant scoping column. Hosts with RLS MAY add a policy referencing current_tenant(); the adapter does NOT set per-connection session variables — that is the consumer's pool configuration.

The table option is validated against /^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?$/ (bare identifier or schema.table form) at construction. Postgres parameterized identifiers are not supported by the protocol; the allowlist is the safe floor.

Datadog

import { createDatadogCostEmitter } from "@pleach/gateway/cost"

const emitter = createDatadogCostEmitter({
  apiKey: process.env.DD_API_KEY!,
  site: "datadoghq.com",
  flushIntervalMs: 10_000,
  batchSize: 100,
})

POSTs to the Datadog /api/v2/series metrics submission endpoint using global fetch — no SDK dependency. EU customers use datadoghq.eu; US3 customers use us3.datadoghq.com; etc. The factory does NOT validate site against an allowlist (Datadog adds sites over time).

Two series per event:

• pleach.gateway.cost.usd (count, event.costUsd)
• pleach.gateway.tokens.total (count, prompt + completion)

Both series carry the tags tenant_id, family, call_class, model, byok. The token series additionally records prompt / completion as two separate points tagged token_kind:prompt / token_kind:completion.

Honeycomb

import { createHoneycombCostEmitter } from "@pleach/gateway/cost"

const emitter = createHoneycombCostEmitter({
  apiKey: process.env.HONEYCOMB_API_KEY!,
  dataset: "pleach-gateway",
  apiHost: "api.honeycomb.io",
  flushIntervalMs: 5_000,
  batchSize: 50,
})

POSTs to https://<apiHost>/1/batch/<dataset> using global fetch — no SDK dependency. EU customers may use api.eu1.honeycomb.io.

Each CostEvent becomes one Honeycomb event with flat top-level fields (tenant_id, family, call_class, model_invoked, cost_usd, prompt_tokens, completion_tokens, tokens_total, byok_active, byok_key_hash, event_type) — Honeycomb auto-indexes them all. The full routingDecision is serialized as routing_decision_json so Honeycomb users can derive query fields without changing the adapter.

Production-readiness notes

The three buffered adapters (Postgres, Datadog, Honeycomb) share a small set of production hardening tactics worth understanding before operating them at scale:

• unref() on interval timers. Each adapter calls .unref() on its periodic-flush setInterval handle. The host process exits cleanly when the cost adapter is the only outstanding handle — important for CLI tools and short-lived workers. Non-Node runtimes (Workers, Deno) don't expose unref(); the adapter type-guards the call so this is harmless cross-runtime.

• Failed batches re-buffer at the head. When a flush fails (HTTP 5xx, pool exhausted, network reset), the failed batch is unshift()-ed back to the front of the buffer. Ordering is preserved across the retry — the next flush replays the same batch in the same sequence. onError is invoked on every failure for observability.

• 1000-iter flush() safety cap. On graceful shutdown, flush() clears the interval timer and drains repeatedly until the buffer is empty. A safety counter caps the drain loop at 1000 iterations so a pathological re-buffer (every batch fails forever) cannot wedge shutdown. Hosts whose drain budget exceeds 1000 × batchSize events should log and force-exit.

• Emit-side never throws. emit() is void for the buffered adapters and never propagates a synchronous failure. The middleware's void middleware(...) call is fire-and-forget by design; cost emission is structurally decoupled from the gateway's hot path.

Enforcing a per-root-turn cost ceiling

To cap swarm spend, compose RootTurnCostAggregator with CostCeilingMiddleware from @pleach/gateway/swarm at your gateway integration seam. The aggregator rolls per-sub-agent cost events up under the originating rootTurnId; the middleware runs a pre- and post-call check, fires a once-per-rootTurnId warning at a configured soft threshold, and throws CostCeilingExceededError on a hard breach.

Per-root-turn aggregation is the right granularity for swarm deployments: a single user message may fan out to N sub-agents, each emitting M cost events. The aggregator rolls per-sub-agent events up under the originating rootTurnId so the ceiling applies to the user-visible turn cost, not the per-sub-agent slice.

Cited source

• packages/gateway/src/cost/CostEmitter.ts — contract.
• packages/gateway/src/cost/CostMiddleware.ts — middleware factory.
• packages/gateway/src/cost/adapters/otel.ts — OTel adapter.
• packages/gateway/src/cost/adapters/postgres.ts — Postgres adapter.
• packages/gateway/src/cost/adapters/datadog.ts — Datadog adapter.
• packages/gateway/src/cost/adapters/honeycomb.ts — Honeycomb adapter.
• packages/gateway/test/costEmitterSurface.smoke.test.mjs — surface regression-lock.

Where to go next