Durable Token Usage Sink (SQLite)

Track cumulative token spend per gateway identity in a restart-safe SQLite ledger — no new dependencies.

from praisonaiagents.telemetry.durable_sink import SqliteTokenUsageSink

sink = SqliteTokenUsageSink("~/.praisonai/usage.db")
sink.record(identity="tg:123", scope="telegram", cost_usd=0.02)
sink.spent(identity="tg:123", scope="telegram", since=0.0)   # -> 0.02

Quick Start

Ephemeral (in-memory) for tests

Use ":memory:" for a throwaway database that never touches disk.

from praisonaiagents.telemetry.durable_sink import SqliteTokenUsageSink

sink = SqliteTokenUsageSink(":memory:")
sink.record(identity="tg:123", scope="telegram", cost_usd=0.01)

On-disk for production gateways

Pass a file path — ~ is expanded and parent directories are created for you.

from praisonaiagents.telemetry.durable_sink import SqliteTokenUsageSink

sink = SqliteTokenUsageSink("~/.praisonai/usage.db")

Attribute a turn to an identity

Call persist from the telemetry path, passing identity, scope, and cost_usd via metadata.

from types import SimpleNamespace

metrics = SimpleNamespace(input_tokens=120, output_tokens=80, total_tokens=200)
sink.persist(
    task_id="t1",
    agent_name="Support Bot",
    model="gpt-4o",
    metrics=metrics,
    metadata={"identity": "tg:123", "scope": "telegram", "cost_usd": 0.02},
)

Query cumulative spend for a /usage command

Read aggregate spend and token totals for one identity.

report = sink.usage(identity="tg:123", scope="telegram")
print(f"${report['cost_usd']:.2f} over {report['records']} turns")

How It Works

Each turn’s metrics are persisted keyed by canonical identity + scope, then read back cumulatively for admission or a /usage reply. Lookups stay fast because rows are indexed on (identity, scope, ts) via idx_token_usage_identity_scope_ts.

Imports

from praisonaiagents.telemetry.durable_sink import SqliteTokenUsageSink

API Reference

`SqliteTokenUsageSink`

Implements the existing TokenUsageSinkProtocol using stdlib sqlite3. Thread-safe and context-manager friendly.

Option	Type	Default	Description
`db_path`	`str`	`":memory:"`	SQLite file path. `~` is expanded, parent directories are created. Use `":memory:"` for an ephemeral database.

Method	Signature	Description
`record`	`(*, identity, scope="", model="", input_tokens=0, output_tokens=0, total_tokens=None, cost_usd=0.0, ts=None) -> None`	Gateway-facing write. Appends a usage row keyed by canonical user id.
`persist`	`(task_id, agent_name, model, metrics, metadata=None) -> None`	`TokenUsageSinkProtocol` entry point. Reads `identity`, `scope`, `cost_usd` from `metadata`; falls back to `agent_name` as identity.
`spent`	`(*, identity, scope=None, since=0.0) -> float`	Cumulative USD spent. `scope=None` sums across all scopes. `since` is the window start.
`oldest_spend_ts`	`(*, identity, scope=None, since=0.0) -> float \| None`	Timestamp of the earliest non-zero-cost charge in the window. `None` when there is no spend.
`usage`	`(*, identity, scope=None, since=0.0) -> dict`	Aggregate for a `/usage` surface: `identity`, `scope`, `cost_usd`, `input_tokens`, `output_tokens`, `total_tokens`, `records`.
`close()`	`() -> None`	Close the connection.
`__enter__` / `__exit__`	context manager	Enables `with SqliteTokenUsageSink(db) as sink: ...`.

Schema

CREATE TABLE IF NOT EXISTS token_usage (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    identity      TEXT    NOT NULL,
    scope         TEXT    NOT NULL DEFAULT '',
    model         TEXT    NOT NULL DEFAULT '',
    input_tokens  INTEGER NOT NULL DEFAULT 0,
    output_tokens INTEGER NOT NULL DEFAULT 0,
    total_tokens  INTEGER NOT NULL DEFAULT 0,
    cost_usd      REAL    NOT NULL DEFAULT 0.0,
    ts            REAL    NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_token_usage_identity_scope_ts
    ON token_usage (identity, scope, ts);

The idx_token_usage_identity_scope_ts index is why per-identity spend lookups stay fast even as the ledger grows.

SDK Reference

Full auto-generated API surface for the telemetry package.

Configuration Options

Three ways to construct the sink.

from praisonaiagents.telemetry.durable_sink import SqliteTokenUsageSink

# Level 1: String path — most common for production
sink = SqliteTokenUsageSink("~/.praisonai/usage.db")

# Level 2: ":memory:" — ephemeral testing
sink = SqliteTokenUsageSink(":memory:")

# Level 3: Context manager — auto-close
with SqliteTokenUsageSink("~/.praisonai/usage.db") as sink:
    sink.record(identity="tg:123", scope="telegram", cost_usd=0.02)

Common Patterns

Attribute a turn to a gateway identity

from types import SimpleNamespace

metrics = SimpleNamespace(input_tokens=120, output_tokens=80, total_tokens=200)
sink.persist(
    task_id="t1",
    agent_name="Support Bot",
    model="gpt-4o",
    metrics=metrics,
    metadata={"identity": "tg:123", "scope": "telegram", "cost_usd": 0.02},
)

Fall back to agent name

# No identity in metadata → spend attributed to agent_name, nothing dropped
sink.persist("t2", "Support Bot", "gpt-4o", metrics, metadata=None)
sink.spent(identity="Support Bot")   # -> accumulated spend

Rolling-window spend for admission

import time
from praisonaiagents.gateway import WindowedSpendBudgetPolicy

policy = WindowedSpendBudgetPolicy(limit_usd=2.00, window_seconds=86_400)
now = time.time()
spent = sink.spent(identity="tg:123", scope="telegram", since=policy.window_start(now))
decision = policy.check(identity="tg:123", scope="telegram", spent_usd=spent, now=now)

/usage command reply

report = sink.usage(identity="tg:123", scope="telegram")
reply = f"Used ${report['cost_usd']:.2f} across {report['records']} messages."

Accurate retry hints

import time

now = time.time()
window_start = policy.window_start(now)
spent = sink.spent(identity="tg:123", scope="telegram", since=window_start)
oldest = sink.oldest_spend_ts(identity="tg:123", scope="telegram", since=window_start)
decision = policy.check(
    identity="tg:123", scope="telegram",
    spent_usd=spent, now=now, oldest_spend_ts=oldest,
)

Best Practices

Use an absolute path for production

Pass a full path like ~/.praisonai/usage.db. ~ is expanded and parent directories are created automatically, so the ledger persists across restarts.

Zero-cost rows never distort retry hints

oldest_spend_ts ignores rows with cost_usd == 0, so free events don’t skew the retry hint fed to the spend-budget policy.

Prefer stdlib sqlite3 here

This sink uses stdlib sqlite3 — no extra dependency. For Postgres or Redis, implement TokenUsageSinkProtocol as a separate sink instead.

Token Usage Protocol

The TokenUsageSinkProtocol this sink implements

Spend Budget

Cap per-identity spend using this sink’s readings

Token Tracking

Measure token consumption per turn

Cost Tracking

Estimate and report LLM costs

Getting Started

Install

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Telemetry & Analytics

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Durable Token Usage Sink (SQLite)

Quick Start

How It Works

Imports

API Reference

`SqliteTokenUsageSink`

Schema

SDK Reference

Configuration Options

Common Patterns

Attribute a turn to a gateway identity

Fall back to agent name

Rolling-window spend for admission

/usage command reply

Accurate retry hints

Best Practices

Token Usage Protocol

Spend Budget

Token Tracking

Cost Tracking

​Quick Start

​How It Works

​Imports

​API Reference

​SqliteTokenUsageSink

​Schema

SDK Reference

​Configuration Options

​Common Patterns

​Attribute a turn to a gateway identity

​Fall back to agent name

​Rolling-window spend for admission

​/usage command reply

​Accurate retry hints

​Best Practices

​Related

Token Usage Protocol

Spend Budget

Token Tracking

Cost Tracking

Quick Start

How It Works

Imports

API Reference

`SqliteTokenUsageSink`

Schema

Configuration Options

Common Patterns

Attribute a turn to a gateway identity

Fall back to agent name

Rolling-window spend for admission

/usage command reply

Accurate retry hints

Best Practices

Related