Build a Streaming Usage Ledger for OpenAI-Compatible AI Gateways

Streaming chat completions make AI products feel fast. They also make billing and reliability harder to explain.

In a normal request, the client sends a prompt, the server waits, and one final JSON response carries the generated text plus usage metadata. A ledger can record one row after the response finishes.

Streaming changes the shape of the problem. The user may see 200 tokens before the model times out. A mobile client may disconnect after receiving a useful answer. A retry may start while the first request is still flushing chunks. Some upstream providers send final usage data in the last event; others require you to calculate or reconcile it later. If your gateway supports DeepSeek, GLM, Kimi, Qwen, MiniMax, and other OpenAI-compatible routes, the accounting problem becomes a distributed systems problem.

This article shows a practical pattern: a streaming usage ledger that records what was requested, what was emitted, what was acknowledged by the client, what was billed by the provider, and what should be shown to the user.

The goal is not to make streaming complicated. The goal is to make partial answers auditable.

Why streaming needs its own ledger

Most teams start with one usage table:

Field	Meaning
request_id	Application request identifier
user_id	Internal user or workspace
model	Selected model
input_tokens	Prompt tokens
output_tokens	Final completion tokens
cost_usd	Calculated request cost
status	success or failure

That table is useful for non-streaming calls, but it loses important facts when the response is streamed.

For example, imagine a coding agent that asks a model to rewrite a file. The model streams 900 output tokens. The browser receives 760 tokens, then the user closes the tab. The upstream provider continues for another second and reports 940 output tokens in the final usage frame. Your application retries the same task on a faster model because the client never received the final done event.

If you only store one final row, you cannot answer these questions:

Question	Why it matters
Did the user receive useful output?	Support and refund reviews need evidence.
Did the provider finish after the client disconnected?	Gateway margins depend on upstream work, not only browser delivery.
Was the retry necessary?	Agent frameworks often retry aggressively after stream interruptions.
Which model was actually charged?	Routing can switch from planning to execution models.
Which pricing version was used?	Rate cards and group multipliers change over time.

A streaming ledger keeps these facts separate instead of forcing them into one status field.

Pricing facts must be versioned

Before designing the ledger, capture the pricing snapshot used for the calculation.

For this article, I checked AIWave live pricing on 2026-08-30. The pricing API returned success=true, pricing_version=a42d372ccf0b5dd13ecf71203521f9d2, 63 model records, auto_groups=["default"], and group ratios of default=3 and vip=1. Target OpenAI-compatible rows included deepseek-v4-flash, deepseek-v4-pro, kimi-k3, glm-5, glm-5.1, Qwen 3.5/3.6/3.7 routes, and MiniMax M2/M3 routes.

That is the level of specificity a ledger needs. It does not need to quote every public provider rate in the hot path. It does need to store enough metadata to reproduce the calculation later:

Ledger field	Example
pricing_version	a42d372ccf0b5dd13ecf71203521f9d2
billing_group	default or vip
group_ratio	3 or 1
model_ratio	Model row ratio at request time
completion_ratio	Output multiplier at request time
cache_ratio	Cached-input multiplier when exposed
source_checked_at	2026-08-30T21:00:00+08:00

Without those fields, a support engineer reviewing a request next week may accidentally recalculate it with today's rates instead of the rates that were active when the stream ran.

Event model

Use one request row and many event rows.

The request row is stable:

create table ai_request (
  request_id text primary key,
  user_id text not null,
  route text not null,
  requested_model text not null,
  resolved_model text not null,
  billing_group text not null,
  pricing_version text not null,
  status text not null,
  started_at text not null,
  finished_at text
);

The event rows describe what happened:

create table stream_event (
  event_id text primary key,
  request_id text not null references ai_request(request_id),
  sequence integer not null,
  event_type text not null,
  output_chars integer not null default 0,
  output_tokens_estimate integer not null default 0,
  provider_input_tokens integer,
  provider_output_tokens integer,
  error_code text,
  created_at text not null
);

Keep event_type boring and explicit:

Event type	Meaning
request_started	Gateway accepted the request.
provider_connected	Upstream connection opened.
chunk_received	Gateway received streamed content.
chunk_delivered	Gateway wrote content to the client socket.
client_disconnected	Client closed the connection before done.
provider_usage	Provider reported final token usage.
retry_started	Application or gateway began a retry.
request_finished	Gateway closed the request lifecycle.

This structure avoids a common mistake: treating streamed text as the same thing as billable usage. Text delivery is a user-experience event. Provider usage is a billing event. They often line up, but they are not the same fact.

Minimal Python implementation

Here is a small SQLite-backed ledger. It is intentionally plain so you can drop it into a FastAPI, Flask, or background-worker codebase.

import os
import sqlite3
import time
import uuid
from contextlib import contextmanager

DB_PATH = os.environ.get("LEDGER_DB_PATH", "stream-ledger.sqlite3")


@contextmanager
def db():
    con = sqlite3.connect(DB_PATH)
    con.row_factory = sqlite3.Row
    try:
        yield con
        con.commit()
    finally:
        con.close()


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def init_schema():
    with db() as con:
        con.executescript("""
        create table if not exists ai_request (
          request_id text primary key,
          user_id text not null,
          route text not null,
          requested_model text not null,
          resolved_model text not null,
          billing_group text not null,
          pricing_version text not null,
          status text not null,
          started_at text not null,
          finished_at text
        );

        create table if not exists stream_event (
          event_id text primary key,
          request_id text not null references ai_request(request_id),
          sequence integer not null,
          event_type text not null,
          output_chars integer not null default 0,
          output_tokens_estimate integer not null default 0,
          provider_input_tokens integer,
          provider_output_tokens integer,
          error_code text,
          created_at text not null
        );
        """)


def start_request(user_id, route, requested_model, resolved_model, group, pricing_version):
    request_id = str(uuid.uuid4())
    with db() as con:
        con.execute(
            """
            insert into ai_request
            (request_id, user_id, route, requested_model, resolved_model,
             billing_group, pricing_version, status, started_at)
            values (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                request_id,
                user_id,
                route,
                requested_model,
                resolved_model,
                group,
                pricing_version,
                "streaming",
                now_iso(),
            ),
        )
    add_event(request_id, 0, "request_started")
    return request_id


def add_event(request_id, sequence, event_type, **fields):
    with db() as con:
        con.execute(
            """
            insert into stream_event
            (event_id, request_id, sequence, event_type, output_chars,
             output_tokens_estimate, provider_input_tokens, provider_output_tokens,
             error_code, created_at)
            values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                str(uuid.uuid4()),
                request_id,
                sequence,
                event_type,
                fields.get("output_chars", 0),
                fields.get("output_tokens_estimate", 0),
                fields.get("provider_input_tokens"),
                fields.get("provider_output_tokens"),
                fields.get("error_code"),
                now_iso(),
            ),
        )


def finish_request(request_id, status):
    with db() as con:
        con.execute(
            "update ai_request set status = ?, finished_at = ? where request_id = ?",
            (status, now_iso(), request_id),
        )

Capturing a stream

The gateway should log both provider chunks and client delivery. That distinction matters when a socket write fails.

def estimate_tokens(text):
    # Replace with your tokenizer for production. This estimate is for interim
    # progress only; final provider usage should override it when available.
    return max(1, len(text) // 4)


def stream_to_client(request_id, provider_events, client_writer):
    delivered_chars = 0

    for sequence, event in enumerate(provider_events, start=1):
        if event["type"] == "content":
            text = event["text"]
            add_event(
                request_id,
                sequence,
                "chunk_received",
                output_chars=len(text),
                output_tokens_estimate=estimate_tokens(text),
            )

            try:
                client_writer.write(text)
                delivered_chars += len(text)
                add_event(
                    request_id,
                    sequence,
                    "chunk_delivered",
                    output_chars=len(text),
                    output_tokens_estimate=estimate_tokens(text),
                )
            except BrokenPipeError:
                add_event(request_id, sequence, "client_disconnected")
                finish_request(request_id, "client_disconnected")
                return

        if event["type"] == "usage":
            add_event(
                request_id,
                sequence,
                "provider_usage",
                provider_input_tokens=event["input_tokens"],
                provider_output_tokens=event["output_tokens"],
            )

    finish_request(request_id, "completed")

For a real gateway, also record the upstream request id when the provider exposes one. It gives your finance, support, and engineering teams a shared join key.

Reconciliation query

At the end of the request, build a reconciled view. This query compares delivered output with provider-reported usage.

select
  r.request_id,
  r.user_id,
  r.resolved_model,
  r.billing_group,
  r.pricing_version,
  r.status,
  sum(case when e.event_type = 'chunk_delivered' then e.output_chars else 0 end)
    as delivered_chars,
  sum(case when e.event_type = 'chunk_delivered' then e.output_tokens_estimate else 0 end)
    as delivered_tokens_estimate,
  max(e.provider_input_tokens) as provider_input_tokens,
  max(e.provider_output_tokens) as provider_output_tokens,
  sum(case when e.event_type = 'client_disconnected' then 1 else 0 end)
    as client_disconnects
from ai_request r
join stream_event e on e.request_id = r.request_id
where r.started_at >= datetime('now', '-1 day')
group by r.request_id;

The dashboard should display both numbers:

Metric	Source	Use
Delivered tokens estimate	Gateway chunk events	User-visible progress and UX debugging
Provider output tokens	Final provider usage	Billing and margin calculation
Client disconnect count	Gateway socket events	Retry and support analysis
Pricing version	Pricing snapshot	Audit and recalculation

Do not hide the mismatch. The mismatch is the evidence.

Retry rules

Retries are where streaming ledgers become valuable. A retry after zero delivered tokens is different from a retry after 900 delivered tokens.

Use policy bands:

Delivered output	Retry policy
0 tokens	Safe to retry automatically within the user budget.
Small prefix only	Retry with explicit previous_attempt_id.
Useful partial answer	Ask the product layer whether to continue, summarize, or stop.
Final answer delivered but done event missing	Do not blindly retry; reconcile first.

The retry row should reference the previous attempt:

create table request_retry (
  retry_id text primary key,
  previous_request_id text not null,
  next_request_id text not null,
  reason text not null,
  created_at text not null
);

This lets you answer a critical question: "How much of our monthly model spend comes from retries after partial streamed responses?"

If the answer is high, you may not have a pricing problem. You may have an SDK timeout problem, a proxy buffering problem, or an agent framework that interprets missing done events too aggressively.

What to show users

A user-facing usage page should not expose every internal event, but it should show enough detail to build trust.

For each streamed request, show:

User-facing field	Example
Model	deepseek-v4-pro
Status	Completed, interrupted, or retried
Input tokens	Provider-reported count when available
Output tokens	Provider-reported count when available
Delivered output	Present only for interrupted streams
Billing group	Default, VIP, or contract group
Rate snapshot	Pricing version and date

For AIWave-style gateways, this is especially important because teams may switch between Chinese AI model families behind one OpenAI-compatible client. A single USD invoice is much easier to trust when each row can explain the route, model, group, and pricing snapshot used at the time.

Operational checks

Add these alerts before traffic grows:

Alert	Why
Provider usage missing on completed streams	Billing will fall back to estimates.
High disconnect rate by region or client version	SDK or network issue.
Retry spend above threshold	Agent loops may be wasting budget.
Pricing version missing	Request cannot be audited later.
Delivered tokens much lower than provider tokens	Client is not receiving what upstream produced.

None of these alerts require customer-identifying data in the log. Store internal ids, request ids, and aggregated counters. Keep API keys out of events entirely.

Data retention boundaries

The ledger should be useful without becoming a transcript archive.

For most teams, the event table should store counts, states, ids, and route metadata. It should not store full prompts, full streamed text, secrets, payment details, or customer contact information. If support needs to inspect an individual incident, link the ledger row to a short-lived support case with explicit access controls instead of putting sensitive content into the accounting database.

That boundary also helps with procurement. A Tier 1 buyer may ask how a gateway handles data retention, billing evidence, and incident review. The stronger answer is not "we store everything." The stronger answer is: "we store the minimum evidence needed to explain usage, we separate billing events from content, and we can delete or aggregate old event rows while preserving invoice totals."

A simple retention policy works well:

Data class	Suggested treatment
Request and usage totals	Keep for finance and tax records.
Stream event metadata	Keep for a shorter operational window.
Provider request ids	Keep while support disputes are possible.
Prompt or completion text	Do not store in the ledger.

This keeps the ledger focused on accounting, not surveillance.

Implementation checklist

If you are adding streaming to an OpenAI-compatible gateway, ship the ledger with the stream, not after the first billing dispute.

1. Create a request row before opening the upstream stream.
2. Record the resolved model, billing group, and pricing version at request start.
3. Log provider chunk receipt separately from client socket delivery.
4. Treat final provider usage as the billing source when it is available.
5. Keep interim token estimates labeled as estimates.
6. Link retries to previous attempts.
7. Make interrupted streams visible in support tooling.
8. Alert when completed streams do not contain provider usage.

Streaming is a product feature, but partial streaming is an accounting state. Once you model it that way, retries become explainable, support reviews become faster, and users can see why a request was billed the way it was.

That is the difference between "the model streamed something" and "the gateway can prove what happened."