Metadata-Version: 2.4
Name: axio-context-sqlite
Version: 0.8.0
Summary: SQLite-backed context store for Axio
Project-URL: Homepage, https://github.com/mosquito/axio-agent
Project-URL: Repository, https://github.com/mosquito/axio-agent
License: MIT
Keywords: agent,ai,context,llm,sqlite,storage
Requires-Python: >=3.12
Requires-Dist: aiosqlite>=0.20
Requires-Dist: axio
Description-Content-Type: text/markdown

# axio-context-sqlite

[![PyPI](https://img.shields.io/pypi/v/axio-context-sqlite)](https://pypi.org/project/axio-context-sqlite/)
[![Python](https://img.shields.io/pypi/pyversions/axio-context-sqlite)](https://pypi.org/project/axio-context-sqlite/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)

SQLite-backed persistent context store for [axio](https://github.com/mosquito/axio-agent).

## Installation

```bash
pip install axio-context-sqlite
```

## Usage

### `connect(db_path)`

Open (or create) a SQLite database at `db_path` and initialise the schema.
Returns an `aiosqlite.Connection`. The caller is responsible for closing it.

```python
async def main():
    conn = await connect("~/.axio/chat.db")
    # ... use the connection ...
    await conn.close()
```

The helper enables WAL journal mode and a 5-second busy timeout so concurrent
readers and a single writer can safely share the same database file.

### `SQLiteContextStore(conn, session_id, project=None, db_name="axio_context")`

Create a context store bound to one session.

| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `conn` | `aiosqlite.Connection` | — | Open database connection (from `connect()`) |
| `session_id` | `str` | — | Unique identifier for this conversation session |
| `project` | `str \| None` | `str(Path.cwd().resolve())` | Logical project scope used to group and list sessions. Defaults to the current working directory. |
| `db_name` | `str` | `"axio_context"` | Prefix used for the database table names (`axio_context_messages`, `axio_context_tokens`). |

Open a connection with `connect()`, then create a `SQLiteContextStore` bound to a
session. The caller owns the connection and is responsible for closing it.

<!-- name: test_readme_usage -->
```python
import asyncio
import tempfile
import pathlib
from axio_context_sqlite import connect, SQLiteContextStore
from axio.messages import Message
from axio.blocks import TextBlock

async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        store = SQLiteContextStore(conn, session_id="my-session")
        await store.append(Message(role="user", content=[TextBlock(text="Hello")]))
        history = await store.get_history()
        assert len(history) == 1
    finally:
        await conn.close()

asyncio.run(main())
```

`SQLiteContextStore` implements the `axio.context.ContextStore` ABC and persists
conversation history across process restarts. Multiple sessions can coexist in
the same database file, isolated by `session_id` and `project`.

#### Storage and compression

Message content is stored as serialized JSON. Payloads larger than 512 bytes
are automatically compressed with gzip (compresslevel 6) and stored
base64-encoded with a `gzip:` prefix. Smaller payloads are stored as-is with a
`plain:` prefix. Decompression happens transparently on read — callers never
see the encoded form.

#### SQLite performance settings

Every connection opened by `connect()` is configured with:

- `PRAGMA journal_mode=WAL` — enables concurrent readers alongside one writer
- `PRAGMA busy_timeout=5000` — waits up to 5 seconds before raising a lock error
- `PRAGMA synchronous=NORMAL` — balances durability and write throughput

### Agent integration

<!--
name: test_readme_agent
```python
from axio.testing import StubTransport, make_text_response
transport = StubTransport([make_text_response("Hi!")])
```
-->
<!-- name: test_readme_agent -->
```python
import asyncio
import tempfile
import pathlib
from axio.agent import Agent
from axio_context_sqlite import connect, SQLiteContextStore

async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        ctx = SQLiteContextStore(conn, session_id="main")
        agent = Agent(system="You are helpful.", tools=[], transport=transport)
        result = await agent.run("Hello!", ctx)
        assert result == "Hi!"
    finally:
        await conn.close()

asyncio.run(main())
```

### Listing sessions

<!-- name: test_readme_list_sessions -->
```python
import asyncio
import tempfile
import pathlib
from axio_context_sqlite import connect, SQLiteContextStore
from axio.messages import Message
from axio.blocks import TextBlock

async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        store = SQLiteContextStore(conn, session_id="main", project="/myproject")
        await store.append(Message(role="user", content=[TextBlock(text="hi")]))
        sessions = await store.list_sessions()
        for s in sessions:
            print(s.session_id, s.preview, s.message_count)
        assert len(sessions) == 1
    finally:
        await conn.close()

asyncio.run(main())
```

### Token accounting

`add_context_tokens(input_tokens, output_tokens)` atomically increments the
stored token counts for the current session and project using a SQL UPSERT
(`INSERT ... ON CONFLICT DO UPDATE SET ... = ... + excluded....`). This is safe
to call concurrently from multiple coroutines without an application-level lock.

`set_context_tokens()` replaces the counts unconditionally, and
`get_context_tokens()` returns a `(input_tokens, output_tokens)` tuple.

### Forking

`fork()` copies the current session's messages into a new session — useful for
branching conversations without affecting the original:

<!-- name: test_readme_fork -->
```python
import asyncio
import tempfile
import pathlib
from axio_context_sqlite import connect, SQLiteContextStore
from axio.messages import Message
from axio.blocks import TextBlock

async def main() -> None:
    conn = await connect(pathlib.Path(tempfile.mkdtemp()) / "chat.db")
    try:
        store = SQLiteContextStore(conn, session_id="main")
        await store.append(Message(role="user", content=[TextBlock(text="original")]))
        branch = await store.fork()
        assert branch.session_id != store.session_id
        assert len(await branch.get_history()) == 1
    finally:
        await conn.close()

asyncio.run(main())
```

## License

MIT
