Metadata-Version: 2.4
Name: nosql-trace
Version: 0.1.0
Summary: Durable, pluggable NoSQL log sink for agentic workflows (MongoDB, DynamoDB, and Mongo-API-compatible flavors)
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.12
Requires-Dist: boto3>=1.34
Requires-Dist: pymongo>=4.9
Description-Content-Type: text/markdown

# 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.

## 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,

    # mongo driver (only used when backend="mongodb")
    write_concern=1,
    insert_ordered=False,
    connect_timeout_ms=5000,
    server_selection_timeout_ms=5000,

    # 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
)
```

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 flush(timeout: float = 5.0) -> None: ...
    # blocking manual flush of everything currently buffered
```

`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) and
[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).

## 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()` p99 latency | < 1 ms @ 5,000 events/sec | ~0.08 ms |
| Sustained throughput (8 threads) | 5,000 events/sec | ~19,000 events/sec |

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(), flush()
├── config.py           # Config dataclass, backend selection, configure()
├── models.py            # LogEvent, EventKind
├── context.py            # contextvars for trace_id/span stack
├── 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)

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

## License

MIT
