Storage Adapters

Outcome. Persist decision records in a backend you can still query months later.

Prerequisites. The storage extra; the SQLite backend uses the runtime shipped with the package.

Packages. briefcase-ai

Languages. Python

A storage adapter is the backend that holds your audit trail — the durable home for every decision Briefcase AI captures, and the surface you query when someone asks why a decision was made.

storage For: platform & governance

When you’d reach for this

Your support-triage agent has been recording classify_ticket decisions for a month, and a reviewer now wants to pull every support-queue decision from last week. Those records have to live somewhere durable and queryable. The backend you pick decides whether that review is a one-line query against a file on disk or a lost cause because the records were only ever in memory.

How it works

1. Init — call briefcase.init() once and construct a backend.

2. Create — build a DecisionSnapshot for each classify_ticket call.

3. Save — save_decision() persists the record and returns its id.

4. Query — pull records back with a SnapshotQuery when you need to review them.

flowchart LR
    A[classify_ticket] --> B[DecisionSnapshot]
    B --> C[SqliteBackend]
    C --> D[(Audit trail)]
    E[Reviewer query] --> C

Diagram description

A decision becomes a DecisionSnapshot, which is saved into the configured storage backend, building a durable audit trail. A reviewer’s SnapshotQuery reads back from the same backend.

Install

pip install briefcase-ai[storage]

Note

The storage extra installs no third-party dependencies; the SQLite backend uses the runtime that ships with the package.

Which backend should I use?

Backend	Persistence	Reach for it when
SqliteBackend.in_memory()	No — gone on exit	Tests and local experiments, where no durable audit trail is needed
SqliteBackend("path.db")	Yes — a file on disk	A single node where you want a real, queryable audit trail
BufferedBackend	Yes — wraps another backend, after a flush	High write volume, where one transaction per batch beats one per decision

BufferedBackend is not a separate store. It wraps a durable backend (such as SqliteBackend) and holds decisions in memory until buffer_size of them accumulate, then writes the whole batch in one transaction. save_decision returns the decision’s id straight away, and load_decision finds a decision that is still in the buffer, so a returned id is usable immediately.

Init -> create -> save -> query

1. Init the runtime and backend

   import briefcase
   from briefcase.storage import SqliteBackend
   
   briefcase.init()  # start the native runtime once per process
   backend = SqliteBackend("decisions.db")

2. Create a decision for each classify_ticket call.

   from briefcase import DecisionSnapshot, Input, Output
   
   decision = DecisionSnapshot("classify_ticket")
   decision.add_input(Input("text", "reset my password", "string"))
   output = Output("category", "account_access", "string")
   output.with_confidence(0.92)
   decision.add_output(output)
   decision.add_tag("queue", "support")

3. Save the record — save_decision() returns its id.

   decision_id = backend.save_decision(decision)
   loaded = backend.load_decision(decision_id)
   print(loaded.function_name)  # classify_ticket

4. Query the audit trail with a SnapshotQuery.

   from briefcase import SnapshotQuery
   
   results = backend.query(
       SnapshotQuery()
       .with_function_name("classify_ticket")
       .with_tag("queue", "support")
   )
   print(len(results))

briefcase.init() must be called once before using a backend to start the native runtime.

Backends in detail

In-memory (for tests)

SqliteBackend.in_memory() keeps data in memory — fast and ephemeral, the right choice for tests where you do not need records to survive the process.

import briefcase
from briefcase.storage import SqliteBackend

briefcase.init()
backend = SqliteBackend.in_memory()

File on disk (for a real audit trail)

SqliteBackend(path) writes to a file — a durable, queryable audit trail in one place, the workhorse for single-node deployments.

import briefcase
from briefcase.storage import SqliteBackend

briefcase.init()
backend = SqliteBackend("decisions.db")
print(backend.health_check())  # True

Buffered (for high volume)

BufferedBackend wraps a durable backend and commits decisions in batches. Use it as a context manager so a partial batch cannot be left behind:

import briefcase
from briefcase.storage import SqliteBackend, BufferedBackend
from briefcase import DecisionSnapshot, Input

briefcase.init()
store = SqliteBackend("decisions.db")

with BufferedBackend(store, buffer_size=100) as backend:
    decision = DecisionSnapshot("classify_ticket")
    decision.add_input(Input("text", "update my address", "string"))
    decision_id = backend.save_decision(decision)

    print(backend.pending())                              # 1 — not written yet
    print(backend.load_decision(decision_id).function_name)  # readable anyway

# leaving the block flushes; the batch is now durable

flush() writes the batch immediately and returns how many decisions it wrote. pending() is how many are waiting.

A governance query: load decisions for review

The point of a durable backend is the review it enables. When a reviewer asks for last week’s support-queue decisions, you answer with a tagged SnapshotQuery against the same store that captured them.

import briefcase
from briefcase.storage import SqliteBackend
from briefcase import SnapshotQuery

briefcase.init()
backend = SqliteBackend("decisions.db")

# Pull every support-queue triage decision for review
query = (
    SnapshotQuery()
    .with_function_name("classify_ticket")
    .with_tag("queue", "support")
    .with_limit(50)
    .with_offset(0)
)
for decision in backend.query(query):
    # hand each record to a reviewer, or replay it to verify
    ...

SnapshotQuery supports with_function_name, with_module_name, with_tag, with_limit, and with_offset. There is no time filter, so tag with what you will want to select on later. query() returns Snapshot objects; the records are on .decisions:

for snapshot in backend.query(query):
    for decision in snapshot.decisions:
        print(decision.function_name, decision.outputs[0].value)

From here a reviewer can audit a decision or replay it.

Snapshots: grouping multiple decisions

A Snapshot groups several decisions; save() returns the snapshot id and load() returns it.

import briefcase
from briefcase.storage import SqliteBackend
from briefcase import DecisionSnapshot, Input, Snapshot

briefcase.init()
backend = SqliteBackend.in_memory()

decision = DecisionSnapshot("classify_ticket")
decision.add_input(Input("text", "where is my order", "string"))

session = Snapshot("session")
session.add_decision(decision)

snapshot_id = backend.save(session)
restored = backend.load(snapshot_id)
print(len(restored.decisions))  # 1

The persistence interface

SqliteBackend exposes the full interface (BufferedBackend only buffers save_decision calls before flushing them to the backend it wraps):

backend.save(snapshot)             # store a Snapshot, returns its id
backend.load(snapshot_id)          # load a Snapshot
backend.save_decision(decision)    # store a DecisionSnapshot, returns its id
backend.load_decision(decision_id)
backend.query(snapshot_query)      # run a SnapshotQuery
backend.delete(snapshot_id)
backend.health_check()

Available backends

Backend	Class	Description
SQLite	SqliteBackend	Local SQLite database (file or in-memory)
Buffered	BufferedBackend	Wraps a backend and batches writes

Scaling beyond a single node

The open-source package ships these two backends, and SqliteBackend covers single-node deployments well. When you outgrow it — multiple writers, central retention, or shared query access — the path is a server-grade backend: S3, GCS, Azure Blob, and PostgreSQL backends are available in the enterprise build. Moving to one is a backend swap, not a change to your decision code.

Limits

Queries filter on identity and tags, never on time. SnapshotQuery offers function name, module name, tag, limit, and offset. “Every decision from last week” is not expressible against the store; tag the window you will want (add_tag("day", "2026-08-13")) at write time, or filter in Python after loading.

query() returns Snapshot objects, not decisions. Each result wraps its records in .decisions. Iterating the query result directly gives you snapshots.

Buffered decisions live in memory until they are flushed. A process that dies with a partial batch loses it, and query() cannot see pending decisions (though load_decision can). Use the context manager, call flush() before exit, or set buffer_size=1 to write through. Dropping a backend with unflushed decisions prints a warning to stderr rather than losing them quietly.

A missing id raises rather than returning None. load_decision and load raise KeyError. delete is the exception: it returns False for an id that was not there.

One process per file. Opening a .db that another process already holds fails at construction, not at write time: RuntimeError: Failed to create SQLite backend: Connection error: Failed to set WAL mode: database is locked. Give each process its own file, or put the shared store behind one service.

Nothing expires. There is no retention policy, TTL, or vacuum. The file grows until you prune it yourself, and records you delete are gone with no tombstone. If you need history that survives correction, that is Bitemporal Storage.

API reference

briefcase.storage has the full backend and SnapshotQuery surface.

Where this fits

Storage is the Store & Query act: the durable home for everything Capture produced, and the surface the later acts read from.

Decision Recording Where decision snapshots come from — the Capture act that fills this backend.

Deterministic Replay Next: load a record back out and reproduce exactly what happened.