Metadata-Version: 2.4
Name: nosql-trace
Version: 0.2.0
Summary: Durable, pluggable NoSQL log sink for agentic workflows (MongoDB, DynamoDB, and Mongo-API-compatible flavors), with identity/cost tracking and opt-in SDK auto-instrumentation
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.12
Requires-Dist: boto3>=1.34
Requires-Dist: pymongo>=4.9
Provides-Extra: agno
Requires-Dist: agno>=1.0; extra == 'agno'
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.25; extra == 'anthropic'
Provides-Extra: crewai
Requires-Dist: crewai>=0.30; extra == 'crewai'
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.2; extra == 'langchain'
Provides-Extra: llama-index
Requires-Dist: llama-index-core>=0.10; extra == 'llama-index'
Provides-Extra: openai
Requires-Dist: openai>=1.0; extra == 'openai'
Provides-Extra: pydantic-ai
Requires-Dist: pydantic-ai>=0.0.30; extra == 'pydantic-ai'
Description-Content-Type: text/markdown

# nosql_trace

[![PyPI](https://img.shields.io/pypi/v/nosql-trace)](https://pypi.org/project/nosql-trace/)

**Durable, pluggable NoSQL log sink for agentic workflows.** Embed it in an
agent process to record LLM calls, tool calls, and agent steps that survive
process crashes, never block your agent on network I/O, and never raise
into your application code — regardless of which backend you send them to.

Supported backends:

- **MongoDB** (and MongoDB-API-compatible flavors: Amazon DocumentDB, Azure
  Cosmos DB's Mongo API, FerretDB — via `mongo_uri` alone, no extra config)
- **AWS DynamoDB**

Runtime dependencies: **`pymongo` + `boto3`** — one official driver per
backend. Everything else (validation, serialization, ULID generation,
buffering, retry/backoff) is stdlib — no pydantic, no orjson, no retry library.

- [Why](#why)
- [Install](#install)
- [Quickstart](#quickstart)
- [Choosing a backend](#choosing-a-backend)
- [Core concepts](#core-concepts)
- [Configuration](#configuration)
- [Failure modes & guarantees](#failure-modes--guarantees)
- [Public API reference](#public-api-reference)
- [Performance](#performance)
- [Development](#development)
- [Design docs](#design-docs)

## Why

Agent runs produce a lot of ad-hoc logging: prompts, tool outputs,
intermediate steps, errors — often megabytes of it, often right before a
crash. `nosql_trace` gives you a single place to send that data that:

- **Never blocks the agent.** `log_event()` does dataclass-based validation
  and one local SQLite insert — no network round-trip, ever, on the calling
  thread.
- **Never loses data to a crash.** Every event is durably written to a local
  WAL-mode SQLite buffer before delivery is attempted, regardless of which
  backend it's headed to. If the process dies before the background flush
  worker gets to it, the next process that starts against the same buffer
  directory recovers and delivers it.
- **Never blows up on a huge payload.** Oversized `input`/`output` fields are
  truncated or dropped according to per-backend byte limits (DynamoDB's
  400KB item cap is much smaller than Mongo's 16MB), with the fact recorded
  on the stored event — never an exception in your code.
- **Never raises by default.** Any failure in the logging/delivery path
  (invalid configuration aside, which is validated at `configure()` time) is
  swallowed unless you explicitly opt out via `fail_silently=False`.
- **Reconstructs the call tree**, not just a flat log. `span()`/`traced()`
  automatically link nested events via `trace_id`/`parent_span_id` using
  `contextvars`, so a single agent run's nested steps can be replayed from
  the stored records alone.
- **Switches backends without touching instrumentation code.** The same
  `log_event()`/`span()`/`traced()` calls work unchanged whether you point
  `configure()` at MongoDB or DynamoDB.

It's a library, not a service — no UI, no analytics, no tracing
visualization. Just get the data into your NoSQL store of choice, safely.

## Install

```bash
uv add nosql_trace
# or
pip install nosql_trace
```

Requires Python 3.12+ and a reachable MongoDB or DynamoDB instance.

## Quickstart

```python
import nosql_trace

nosql_trace.configure(
    mongo_uri="mongodb://localhost:27017",
    db_name="nosql_trace_demo",
)

with nosql_trace.span("run-agent", kind=nosql_trace.EventKind.AGENT_STEP):
    nosql_trace.log_event("tool_call", "search_web", input={"q": "weather"})

nosql_trace.flush(timeout=5.0)
```

After this runs, the `agent_logs` collection in `nosql_trace_demo` has two
documents: the `search_web` tool call and the `run-agent` span, with the
tool call's `parent_span_id` equal to the span's `span_id`.

`configure()` starts a background thread that batches and delivers events;
`flush()` is a manual blocking flush, useful right before process exit in
runtimes (e.g. serverless) where the built-in `atexit`/signal hooks might
not fire reliably.

### Wrapping existing functions

```python
@nosql_trace.traced()
def call_llm(prompt: str) -> str:
    ...

@nosql_trace.traced("fetch_docs")
def search_tool(query: str) -> list[dict]:
    ...
```

`traced()` is a decorator form of `span()` — it defaults the event name to
the wrapped function's `__qualname__` if you don't pass one.

### Recording an event without a span

```python
nosql_trace.log_event(
    nosql_trace.EventKind.LLM_CALL,
    "chat_completion",
    input={"messages": [...]},
    output={"text": "..."},
    duration_ms=842.3,
    status="ok",
)
```

If a `span()` is active on the current thread/task, `trace_id` and
`parent_span_id` are filled in automatically — pass them explicitly to
override.

### Identity-scoped tracing

```python
with nosql_trace.trace(tenant_id="acme", user_id="u-42", session_id="s-1", workflow_id="research-agent"):
    nosql_trace.log_event("agent_step", "plan")
```

`trace()` sets `session_id`/`user_id`/`tenant_id`/`workflow_id` (and
`trace_id`, generating one if you don't supply it) for every
`log_event()`/`span()`/`log_llm()` call made inside the `with` block, the
same way `trace_id` already propagates via `contextvars`. All four fields
are optional, stored as top-level (not `metadata`-nested) fields, and
indexed per backend — a MongoDB compound index on `(tenant_id, session_id,
ts)`, a DynamoDB GSI on `(tenant_id, session_id)` — so filtering by
tenant/user/session/workflow doesn't require a full scan.

### LLM token & cost tracking

```python
nosql_trace.log_llm(
    model="gpt-5", provider="openai",
    prompt_tokens=300, completion_tokens=400, total_tokens=700,
    latency=812.3, cost={"input": 0.001, "output": 0.004},
    input={"messages": [...]}, output={"text": "..."},
)
```

`log_llm()` is a thin wrapper over `log_event()` that stores model,
provider, token counts, latency, and cost as a structured `llm_usage`
sub-schema. Unlike `input`/`output`, `llm_usage` is **never truncated or
dropped** by the size guardrails — cost data survives even when the
surrounding payload is large enough to be truncated.

### Operational health & dead-letter replay

```python
nosql_trace.trace_health()
# -> {"pending": 12, "failed": 2, "retries": 2, "dead_lettered": 1,
#     "buffer_size_bytes": 12, "oldest_event_age_s": 4.2}

nosql_trace.export_dead_letters()  # inspect without removing
nosql_trace.replay()               # redeliver every dead-lettered event
nosql_trace.replay_errors()        # redeliver only permanent-classified failures
```

These wrap the dead-letter table and metrics counters that already back
the crash-recovery/retry machinery — no new storage, just a public API
surface. `replay()`/`replay_errors()` reuse the same dedup-by-event-id
delivery path as normal flushing, so already-delivered events are never
duplicated.

### Sampling, redaction, and reading traces back

```python
nosql_trace.configure(
    mongo_uri="mongodb://localhost:27017",
    db_name="my_app",
    sample_rate=0.1,                         # keep 10% of traces, decided once per trace_id
    redact_callback=lambda event: event,      # mutate/scrub an event before it's buffered; return None to drop it
)

nosql_trace.get_trace(trace_id)                            # all events for a trace, ts-ascending
nosql_trace.list_traces(tenant_id="acme", limit=50)         # recent trace summaries for a tenant
```

`sample_rate` (default `1.0`) makes an all-or-nothing decision per
`trace_id` — every event in a sampled-out trace is skipped before it ever
reaches the buffer, so partial traces never happen. `redact_callback` runs
before the size guardrails, on every event, regardless of sampling.
`get_trace()`/`list_traces()` return `[]` if unconfigured or if the active
sink doesn't implement the read path (a WARNING is logged in that case, not
raised).

### SDK auto-instrumentation (opt-in extras)

```bash
uv add "nosql_trace[openai]"   # or [anthropic], [langchain], [crewai], [agno], [llama-index], [pydantic-ai]
```

```python
from nosql_trace.integrations.openai import trace_openai

client = trace_openai()(OpenAI())
client.chat.completions.create(model="gpt-5", messages=[...])  # automatically traced via log_llm()
```

Each of the seven decorators (`trace_openai`, `trace_anthropic`,
`trace_langchain`, `trace_crewai`, `trace_agno`, `trace_llamaindex`,
`trace_pydanticai`) lives in its own module under
`nosql_trace.integrations` and is never imported by the base package —
installing `nosql_trace` with no extras adds zero new runtime
dependencies. Using a decorator whose extra isn't installed raises a
`ConfigurationError` naming the `pip install` command needed, instead of
an opaque `ImportError`. Wrapped calls always return the original value
and propagate the original exception unchanged.

## Choosing a backend

```python
# MongoDB (default) — also covers DocumentDB, Cosmos DB Mongo API, FerretDB
nosql_trace.configure(
    backend="mongodb",           # default, can be omitted
    mongo_uri="mongodb://localhost:27017",
    db_name="my_app",
)

# AWS DynamoDB
nosql_trace.configure(
    backend="dynamodb",
    dynamo_table_name="agent_logs",
    dynamo_region="us-east-1",   # optional; falls back to boto3's normal region resolution
)
```

`log_event()`/`span()`/`traced()`/`flush()` are identical regardless of
`backend` — nothing in your instrumentation code needs to change if you
switch. An invalid `backend` value or a missing backend-required setting
(e.g. `dynamo_table_name` when `backend="dynamodb"`) raises immediately from
`configure()`, never discovered later via a silent delivery failure.

Note: size guardrails apply the **active backend's own limits** — an event
that fits comfortably under MongoDB's defaults may still be truncated when
delivered to DynamoDB, because DynamoDB's item-size limit (400KB) is much
smaller than MongoDB's (16MB). This is expected, not a bug.

### MongoDB-API-compatible flavors

Amazon DocumentDB, Azure Cosmos DB's Mongo API, and FerretDB all speak the
MongoDB wire protocol, so they work with `backend="mongodb"` and no other
code changes — just point `mongo_uri` at them. If a specific flavor doesn't
support one of the library's optional index setups, `configure()` still
succeeds (the gap is recorded locally, not raised).

## Core concepts

| Concept | What it is |
|---|---|
| **Event** | One recorded occurrence: an LLM call, tool call, agent step, or custom event. Carries a ULID `id`, timestamp, `kind`, optional `trace_id`/`span_id`/`parent_span_id`, `input`/`output`/`metadata` dicts, `duration_ms`, and `status` (`"ok"`/`"error"`). |
| **Trace** | All events sharing a `trace_id` — one end-to-end agent run. |
| **Span** | A named unit of work (`span()`/`traced()`) that produces exactly one event on completion and sets `parent_span_id` for anything logged inside it. |
| **Buffer** | Local durable queue (SQLite, WAL mode) that every event passes through before delivery, regardless of destination — this is what makes crash recovery possible. |
| **Sink** | The destination backend — `MongoSink` (batched `insert_many`, dedup-by-`_id`) or `DynamoSink` (batched `batch_writer`, dedup via idempotent overwrite keyed on the event id). |
| **Flush worker** | Background daemon thread that pulls batches off the buffer and writes them to the active sink, with retry/backoff and a circuit breaker for sustained outages; caps its batch size to whatever the sink advertises (DynamoDB's hard 25-item `batch_write_item` limit vs Mongo's configurable `batch_size`). |

Both the buffer and the sink are defined behind small `Protocol` interfaces
(`BufferBackend`, `SinkBackend`), which is what makes adding a destination
(like DynamoDB) additive rather than a rewrite — the public API, buffer,
worker, and lifecycle hooks are untouched by backend choice.

## Configuration

```python
nosql_trace.configure(
    backend="mongodb",                  # "mongodb" (default) or "dynamodb"

    # mongodb (also covers Mongo-API-compatible flavors)
    mongo_uri="mongodb://localhost:27017",
    db_name="my_app",
    collection_name="agent_logs",       # default

    # dynamodb (used when backend="dynamodb")
    dynamo_table_name="agent_logs",
    dynamo_region="us-east-1",
    dynamo_endpoint_url=None,           # override for local DynamoDB / testing

    # local durable buffer
    buffer_path="./.nosql_trace/buffer_{pid}.db",  # {pid} is filled in per process
    max_buffer_bytes=512 * 1024 * 1024, # disk cap; oldest events evicted first beyond this
    max_field_bytes=None,               # per input/output field; defaults per backend if unset
    max_doc_bytes=None,                 # whole-record cap; defaults per backend if unset
    #   mongodb defaults: max_field_bytes=64KB, max_doc_bytes=512KB (safely under Mongo's 16MB hard limit)
    #   dynamodb defaults: max_field_bytes=32KB, max_doc_bytes=380KB (safely under Dynamo's 400KB hard item limit)

    # background flush worker
    batch_size=500,                     # capped to 25 automatically when backend="dynamodb"
    flush_interval_s=2.0,
    max_retries=8,                      # attempts before a row moves to the dead-letter table
    backoff_base_s=0.5,
    backoff_max_s=60.0,

    # write coalescing (hot-path latency; see "Performance" below)
    durable_push=False,                 # False (default): log_event() never commits to SQLite on the
                                         #   caller thread; True restores the old per-event commit
    commit_batch_size=200,              # staged events committed together once this many accumulate
    commit_interval_s=0.2,              # ...or once this long has passed, whichever comes first

    # mongo driver (only used when backend="mongodb")
    write_concern=1,
    insert_ordered=False,
    connect_timeout_ms=5000,
    server_selection_timeout_ms=5000,
    ttl_seconds=None,                   # sets a TTL index (mongo) / expires_at + table TTL (dynamo)

    # sampling & redaction
    sample_rate=1.0,                    # 0.0-1.0; per-trace_id, deterministic, all-or-nothing
    redact_callback=None,               # (LogEvent) -> LogEvent | None; None return drops the event

    # behavior
    fail_silently=True,                 # False makes log_event()/span() raise on internal errors
    on_drop_callback=lambda event, reason: alert(event, reason),  # called when an event is evicted/dropped
)
```

> **Release note:** `durable_push` defaults to `False`, trading a bounded
> crash window (at most `commit_batch_size` events staged in memory, never
> committed to the local SQLite buffer) for a `log_event()` call that never
> blocks on disk I/O. If your workload needs the old zero-loss-on-crash
> guarantee more than it needs hot-path latency, set `durable_push=True`.

Calling `configure()` again is safe and idempotent — it drains and restarts
the background worker under the new configuration rather than leaving two
delivery paths running. Events already sitting in the local buffer from
before a backend switch are delivered to whichever backend is active when
they flush (no retroactive re-routing).

## Failure modes & guarantees

| Scenario | Behavior |
|---|---|
| Backend temporarily unreachable or throttling | Events accumulate in the local SQLite buffer (bounded by `max_buffer_bytes`), retried with exponential backoff — no data loss up to the cap. Applies equally to Mongo network errors and DynamoDB throttling. |
| Process crash after `log_event()`, before flush | Recovered automatically the next time a process starts pointed at the same `buffer_path` directory. |
| Process crash mid-flush (after backend write, before local ack) | The retried write hits the same event id — MongoDB rejects it as a duplicate key (no double-count); DynamoDB overwrites the item with identical content (same net effect, no duplicate item). |
| Oversized or malformed `input`/`output` | Truncated (with `metadata["_truncated"]`/`_original_bytes`) or, if the whole record is still too big, dropped from the record entirely (`metadata["_dropped"]`) — using the active backend's own size limits — never raised, never blocks. |
| Sustained outage beyond disk cap | Oldest buffered events are evicted first (a documented, bounded data-loss mode); `on_drop_callback` fires so you can alert on it. |
| Malformed/permanently-rejected record (validation error, bad table/collection, access denied) | Classified as non-retryable and moved to a local dead-letter table instead of retried forever. |
| Invalid `backend` or missing backend-required setting | Raised immediately from `configure()` — never discovered later via a silent delivery failure. |
| Calling thread performance | Bounded to validation + one SQLite insert — no network I/O, no backend round-trip, ever, on the hot path. |

## Public API reference

```python
def configure(
    *,
    backend: Literal["mongodb", "dynamodb"] = "mongodb",
    mongo_uri: str | None = None,
    db_name: str | None = None,
    dynamo_table_name: str | None = None,
    **kwargs,
) -> None: ...

def log_event(
    kind: EventKind | str,
    name: str,
    *,
    trace_id: str | None = None,
    input: dict | None = None,
    output: dict | None = None,
    metadata: dict | None = None,
    duration_ms: float | None = None,
    status: str = "ok",
    error: str | None = None,
) -> str: ...  # returns the generated event id

def span(name: str, kind: EventKind = EventKind.AGENT_STEP, trace_id: str | None = None, **metadata): ...
    # context manager; logs one event on exit with duration_ms/status/error derived automatically

def traced(name: str | None = None): ...
    # decorator form of span()

def trace(
    *, session_id=None, user_id=None, tenant_id=None, workflow_id=None, trace_id=None,
): ...  # context manager; sets identity fields (+ trace_id) for everything logged inside it

def log_llm(
    model: str, provider: str, *,
    prompt_tokens=None, completion_tokens=None, total_tokens=None,
    latency=None, cost=None, input=None, output=None, **kwargs,
) -> str: ...  # returns the generated event id; kwargs accept the same trace_id/session_id/etc. as log_event()

def trace_health() -> dict: ...
    # {"pending", "failed", "retries", "dead_lettered", "buffer_size_bytes", "oldest_event_age_s"}

def replay() -> int: ...          # redeliver all dead-lettered events; returns count requeued
def replay_errors() -> int: ...   # redeliver only permanent-classified dead-lettered events
def export_dead_letters() -> list[dict]: ...  # inspect dead-lettered records without removing them

def get_trace(trace_id: str) -> list[dict]: ...
    # all events for a trace, ts-ascending; [] if unconfigured or sink lacks read support

def list_traces(*, tenant_id: str | None = None, session_id: str | None = None, limit: int = 100) -> list[dict]: ...
    # recent trace summaries filtered by identity fields; [] if unconfigured or sink lacks read support

def flush(timeout: float = 5.0) -> None: ...
    # blocking manual flush of everything currently buffered; never sleeps a backoff on the caller thread
```

`EventKind` values: `llm_call`, `tool_call`, `agent_step`, `custom`. These
and `log_event`/`span`/`traced`/`flush` are identical regardless of backend.

Full behavioral contracts:
[specs/001-agentlog-mongo-sink/contracts/public-api.md](specs/001-agentlog-mongo-sink/contracts/public-api.md)
(original Mongo-only contract),
[specs/002-multi-backend-sinks/contracts/public-api.md](specs/002-multi-backend-sinks/contracts/public-api.md)
(multi-backend additions: `backend` selector, per-backend guarantees), and
[specs/003-production-essentials/contracts/public-api.md](specs/003-production-essentials/contracts/public-api.md)
(identity fields, `log_llm()`, health/replay, SDK auto-instrumentation), and
[specs/004-production-hardening/contracts/api.md](specs/004-production-hardening/contracts/api.md)
(concurrency/lifecycle fixes, write coalescing, sampling, redaction, OTel
bridge, read API).

## Performance

Measured via `benchmarks/bench_latency.py` and `bench_throughput.py`
(mongomock backend, isolating call-site cost from real network delivery —
backend selection happens once at `configure()` time, not per call, so
these numbers are backend-independent):

| Metric | Target | Measured |
|---|---|---|
| `log_event()` p50/mean latency (coalesced push, `durable_push=False`) | < 100 µs | ~0.05 ms |
| `log_event()` p99 latency | < 100 µs (best-effort; see note) | ~0.15-0.18 ms |
| Sustained throughput (8 threads) | 5,000 events/sec | ~12,000 events/sec |

`durable_push=False` (the default) keeps `log_event()` entirely in memory —
it appends to a staging deque under a dedicated lock and returns, never
touching the SQLite connection the background worker uses to commit. p50
and mean latency reliably clear the 100µs hot-path budget; the p99 tail
includes occasional Windows/CPython scheduler and GC-adjacent noise from a
5,000-iteration microbenchmark, not lock contention on the hot path — see
`specs/004-production-hardening/research.md` for the measured breakdown.

Run them yourself:

```bash
uv run python benchmarks/bench_latency.py
uv run python benchmarks/bench_throughput.py
```

## Development

```bash
uv sync --dev

# fast tests, no I/O beyond SQLite tmp files
uv run pytest tests/unit tests/contract

# SQLite + mongomock/moto (in-process fakes) by default
uv run pytest tests/integration

# opt in to a real MongoDB instance
MONGO_TEST_URI="mongodb://localhost:27017" uv run pytest tests/integration

# opt in to real AWS DynamoDB
DYNAMO_TEST_TABLE="agentlogs-it" AWS_REGION="us-east-1" uv run pytest tests/integration/test_dynamo_sink.py
```

Project layout:

```text
src/nosql_trace/
├── __init__.py       # public API re-exports
├── api.py             # log_event(), span(), traced(), trace(), log_llm(), flush(),
│                       #   trace_health(), replay(), replay_errors(), export_dead_letters()
├── config.py           # Config dataclass, backend selection, configure()
├── models.py            # LogEvent, EventKind, LLMUsage
├── context.py            # contextvars for trace_id/span stack/identity fields
├── serialization.py       # ULID generation, JSON encode/decode, size guardrails
├── errors.py               # exception types
├── buffer/                  # BufferBackend protocol + SQLite (WAL) implementation
├── sink/                     # SinkBackend protocol + MongoDB + DynamoDB implementations
├── worker/                    # background flush thread + lifecycle hooks
├── _internal/metrics.py        # local counters (queued/flushed/failed/dropped)
└── integrations/                 # opt-in SDK auto-instrumentation (openai, anthropic,
                                    #   langchain, crewai, agno, llamaindex, pydanticai);
                                    #   never imported by __init__.py

tests/{unit,integration,contract}/
benchmarks/{bench_latency,bench_throughput}.py
```

## Design docs

- [specs/001-agentlog-mongo-sink/](specs/001-agentlog-mongo-sink/) — original durable MongoDB sink (spec, plan, tasks)
- [specs/002-multi-backend-sinks/](specs/002-multi-backend-sinks/) — DynamoDB + Mongo-flavor compatibility extension
- [specs/003-production-essentials/](specs/003-production-essentials/) — identity fields, cost tracking, health/replay, SDK auto-instrumentation
- [specs/004-production-hardening/](specs/004-production-hardening/) — concurrency/lifecycle fixes, write coalescing, sampling, redaction, OTel bridge, read API

## License

MIT
