From 6659f99d21dee2a2fbae3585f83f396a76dd3a69 Mon Sep 17 00:00:00 2001 From: Tisha Chawla Date: Tue, 28 Jul 2026 21:20:18 +0530 Subject: [PATCH] feat: pluggable storage backends (SQLite, remote control plane) Recording wrote only to a local JSONL file. Deployed agents need durable and shared storage, so introduce a Store protocol with three backends. The default stays local JSONL (zero config); nothing downstream changes. - Store protocol in chronicle/envelope/backends.py. EnvelopeStore (JSONL) already satisfies it; exposed as JsonlStore too. - SqliteStore: durable, queryable, zero dependency (stdlib sqlite3). Thread-safe appends for concurrent async requests. Good for a single deployed instance. - RemoteStore: ships envelopes to a Chronicle control plane over HTTP (stdlib urllib). Never raises into the agent (a failed append is warned and dropped); reads return [] on failure. - open_store(target) picks a backend from a string: a path -> JsonlStore, sqlite:/// or *.db/*.sqlite -> SqliteStore, http(s):// -> RemoteStore. - chronicle.record(store=...) now accepts any Store, or a path/URL string routed through open_store, so store="sqlite:///runs.db" and control-plane URLs work. - examples/control_plane/server.py: a minimal reference control plane (stdlib http.server + SQLite) that RemoteStore talks to and TokenOps can share. Exports: Store, JsonlStore, SqliteStore, RemoteStore, open_store, EnvelopeStore. Tests: tests/test_stores.py (SQLite round-trip, open_store dispatch, record into SQLite and via a sqlite:// URL, RemoteStore end-to-end against the reference server, and that a dead control plane never crashes the agent). ruff 0.15.22 clean. Co-Authored-By: Claude Opus 4.8 Signed-off-by: Tisha Chawla --- README.md | 25 +++++ chronicle/__init__.py | 14 +++ chronicle/api.py | 9 +- chronicle/envelope/__init__.py | 18 ++- chronicle/envelope/backends.py | 172 +++++++++++++++++++++++++++++ examples/control_plane/__init__.py | 6 + examples/control_plane/server.py | 104 +++++++++++++++++ tests/test_stores.py | 114 +++++++++++++++++++ 8 files changed, 458 insertions(+), 4 deletions(-) create mode 100644 chronicle/envelope/backends.py create mode 100644 examples/control_plane/__init__.py create mode 100644 examples/control_plane/server.py create mode 100644 tests/test_stores.py diff --git a/README.md b/README.md index 0343e2d..07a1ebf 100644 --- a/README.md +++ b/README.md @@ -371,6 +371,31 @@ chronicle list-fixtures # list committed envelope fi +
+Storage backends (local file, SQLite, or a shared control plane) + +
+ +Recording writes to a **store**. The default is a local JSONL file, which is all you +need for development and CI. For deployed agents, point at SQLite or a shared control +plane instead. All backends implement the same `Store` protocol, so nothing else changes. + +```python +import chronicle + +with chronicle.record("run-1", store="runs.jsonl"): ... # local file (default) +with chronicle.record("run-1", store="sqlite:///runs.db"): ... # deployed instance +with chronicle.record("run-1", store="https://chronicle.internal"): ... # control plane +``` + +`open_store(target)` picks the backend from the string, or pass a `SqliteStore` / +`RemoteStore` instance directly. `RemoteStore` ships each envelope to a shared HTTP +service and never raises into the agent (a failed send is warned and dropped). A minimal +reference control plane (stdlib, SQLite-backed) is in `examples/control_plane/server.py`; +TokenOps can point at the same service to co-locate its cost ledger with trace storage. + +
+
Cost and governance observers (on_crossing) diff --git a/chronicle/__init__.py b/chronicle/__init__.py index c599bcd..f44b2fd 100644 --- a/chronicle/__init__.py +++ b/chronicle/__init__.py @@ -12,6 +12,14 @@ ToolCall, ToolSchema, ) +from chronicle.envelope.backends import ( + JsonlStore, + RemoteStore, + SqliteStore, + Store, + open_store, +) +from chronicle.envelope.store import EnvelopeStore from chronicle.execution_graph import ExecutionGraph from chronicle.redaction import apply_redactors, default_redactors, redact_secrets from chronicle.replay.plan import BoundaryMode, ReplayPlan @@ -26,12 +34,17 @@ "ChronicleSession", "ContextMetadata", "Envelope", + "EnvelopeStore", "ExecutionGraph", "InputState", + "JsonlStore", "RagChunk", + "RemoteStore", "ReplayPlan", "SamplingParams", "SessionMode", + "SqliteStore", + "Store", "ToolCall", "ToolSchema", "apply_redactors", @@ -39,6 +52,7 @@ "default_redactors", "get_session", "instrument_langgraph", + "open_store", "record", "redact_secrets", "replay_trace", diff --git a/chronicle/api.py b/chronicle/api.py index 4b57cfe..3a3b35d 100644 --- a/chronicle/api.py +++ b/chronicle/api.py @@ -12,7 +12,7 @@ from contextlib import contextmanager from pathlib import Path -from chronicle.envelope.store import EnvelopeStore +from chronicle.envelope.backends import Store, open_store from chronicle.replay.plan import ReplayPlan from chronicle.session import ChronicleSession, reset_session @@ -21,7 +21,7 @@ def record( trace_id: str | None = None, *, - store: EnvelopeStore | str | Path | None = None, + store: Store | str | Path | None = None, model_version: str | None = None, build_id: str | None = None, redactors: list[Callable[[str], str]] | None = None, @@ -42,7 +42,10 @@ def record( """ session = reset_session() if store is not None: - session.store = store if isinstance(store, EnvelopeStore) else EnvelopeStore(store) + # A Store instance (has append) is used directly; a path/URL string is routed + # through open_store, so store="sqlite:///runs.db" or an http control-plane URL + # both work as well as a plain ".jsonl" path. + session.store = store if hasattr(store, "append") else open_store(store) if model_version is not None: session.model_version = model_version if build_id is not None: diff --git a/chronicle/envelope/__init__.py b/chronicle/envelope/__init__.py index 68df1f6..89db763 100644 --- a/chronicle/envelope/__init__.py +++ b/chronicle/envelope/__init__.py @@ -1,5 +1,21 @@ +from chronicle.envelope.backends import ( + JsonlStore, + RemoteStore, + SqliteStore, + Store, + open_store, +) from chronicle.envelope.capture import EnvelopeRecorder from chronicle.envelope.schema import Envelope from chronicle.envelope.store import EnvelopeStore -__all__ = ["Envelope", "EnvelopeRecorder", "EnvelopeStore"] +__all__ = [ + "Envelope", + "EnvelopeRecorder", + "EnvelopeStore", + "JsonlStore", + "RemoteStore", + "SqliteStore", + "Store", + "open_store", +] diff --git a/chronicle/envelope/backends.py b/chronicle/envelope/backends.py new file mode 100644 index 0000000..75046f4 --- /dev/null +++ b/chronicle/envelope/backends.py @@ -0,0 +1,172 @@ +"""Storage backends for envelopes. + +Every backend satisfies the ``Store`` protocol, so `chronicle.record(store=...)` and +`session.store` accept any of them and nothing downstream changes. + +- ``JsonlStore`` (the default ``EnvelopeStore``): append-only JSONL on local disk. + Zero config, perfect for local development and CI fixtures. +- ``SqliteStore``: durable, queryable SQLite. Zero dependency (stdlib ``sqlite3``). + A good fit for a single deployed agent instance. +- ``RemoteStore``: ships envelopes to a Chronicle control plane over HTTP. Point many + deployed agents at one shared service (which TokenOps can also write to). Uses stdlib + ``urllib`` and never raises into the agent: a failed append is warned and dropped, so + recording can never break production. + +Pick one with ``open_store(target)``: + + open_store("runs.jsonl") # JsonlStore (local file) + open_store("sqlite:///runs.db") # SqliteStore (deployed instance) + open_store("https://chronicle.internal") # RemoteStore (shared control plane) +""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +import urllib.error +import urllib.request +import warnings +from pathlib import Path +from typing import Protocol, runtime_checkable + +from chronicle.envelope.schema import Envelope +from chronicle.envelope.store import EnvelopeStore + +# The local JSONL store is the default backend; expose it under the backend name too. +JsonlStore = EnvelopeStore + + +@runtime_checkable +class Store(Protocol): + """What a recording backend must provide. ``EnvelopeStore`` already satisfies it.""" + + def append(self, envelope: Envelope) -> None: ... + def read_all(self) -> list[Envelope]: ... + def find_by_trace_id(self, trace_id: str) -> list[Envelope]: ... + def find_by_envelope_id(self, envelope_id: str) -> Envelope | None: ... + + +class SqliteStore: + """Append-only envelope store backed by SQLite (stdlib, zero dependency). + + Durable and queryable with no files to hand-manage. Safe for concurrent appends + from multiple threads (async requests share one store); writes are serialized with a + lock and the connection allows cross-thread use. + """ + + def __init__(self, path: str | Path) -> None: + self.path = str(path) + if self.path != ":memory:": + Path(self.path).parent.mkdir(parents=True, exist_ok=True) + self._lock = threading.Lock() + self._conn = sqlite3.connect(self.path, check_same_thread=False) + self._conn.execute( + "CREATE TABLE IF NOT EXISTS envelopes (" + "envelope_id TEXT PRIMARY KEY, trace_id TEXT, sequence INTEGER, data TEXT)" + ) + self._conn.execute( + "CREATE INDEX IF NOT EXISTS idx_trace ON envelopes (trace_id, sequence)" + ) + self._conn.commit() + + def append(self, envelope: Envelope) -> None: + with self._lock: + self._conn.execute( + "INSERT OR REPLACE INTO envelopes (envelope_id, trace_id, sequence, data) " + "VALUES (?, ?, ?, ?)", + (envelope.envelope_id, envelope.trace_id, envelope.sequence, + envelope.model_dump_json()), + ) + self._conn.commit() + + def read_all(self) -> list[Envelope]: + rows = self._conn.execute( + "SELECT data FROM envelopes ORDER BY sequence, envelope_id" + ).fetchall() + return [Envelope.from_json(r[0]) for r in rows] + + def find_by_trace_id(self, trace_id: str) -> list[Envelope]: + rows = self._conn.execute( + "SELECT data FROM envelopes WHERE trace_id = ? ORDER BY sequence, envelope_id", + (trace_id,), + ).fetchall() + return [Envelope.from_json(r[0]) for r in rows] + + def find_by_envelope_id(self, envelope_id: str) -> Envelope | None: + row = self._conn.execute( + "SELECT data FROM envelopes WHERE envelope_id = ?", (envelope_id,) + ).fetchone() + return Envelope.from_json(row[0]) if row else None + + def close(self) -> None: + self._conn.close() + + +class RemoteStore: + """Ships envelopes to a Chronicle control plane over HTTP (stdlib ``urllib``). + + Point deployed agents at one shared service. Recording must never break the agent, + so a failed append is warned and dropped rather than raised. Reads return an empty + list on failure. See ``examples/control_plane/server.py`` for a reference service. + """ + + def __init__(self, base_url: str, *, api_key: str | None = None, timeout: float = 5.0) -> None: + self.base_url = base_url.rstrip("/") + self.api_key = api_key + self.timeout = timeout + + def _headers(self) -> dict[str, str]: + headers = {"Content-Type": "application/json"} + if self.api_key: + headers["Authorization"] = f"Bearer {self.api_key}" + return headers + + def append(self, envelope: Envelope) -> None: + request = urllib.request.Request( + f"{self.base_url}/envelopes", + data=envelope.model_dump_json().encode("utf-8"), + headers=self._headers(), + method="POST", + ) + try: + urllib.request.urlopen(request, timeout=self.timeout).read() + except (urllib.error.URLError, OSError) as exc: + warnings.warn(f"chronicle RemoteStore append dropped: {exc}", stacklevel=2) + + def read_all(self) -> list[Envelope]: + return self._get("/envelopes") + + def find_by_trace_id(self, trace_id: str) -> list[Envelope]: + return self._get(f"/traces/{trace_id}/envelopes") + + def find_by_envelope_id(self, envelope_id: str) -> Envelope | None: + found = self._get(f"/envelopes/{envelope_id}") + return found[0] if found else None + + def _get(self, path: str) -> list[Envelope]: + request = urllib.request.Request(f"{self.base_url}{path}", headers=self._headers()) + try: + body = urllib.request.urlopen(request, timeout=self.timeout).read() + except (urllib.error.URLError, OSError): + return [] + payload = json.loads(body) + items = payload if isinstance(payload, list) else [payload] + return [Envelope.from_json(json.dumps(item)) for item in items] + + +def open_store(target: str | Path, **kwargs) -> Store: + """Build a store from a target string. See module docstring for the forms. + + - ``http(s)://...`` -> RemoteStore (control plane), accepts api_key/timeout + - ``sqlite:///path`` or ``*.db`` / ``*.sqlite`` -> SqliteStore + - anything else -> JsonlStore (local file, the default) + """ + text = str(target) + if text.startswith(("http://", "https://")): + return RemoteStore(text, **kwargs) + if text.startswith("sqlite:///"): + return SqliteStore(text[len("sqlite:///"):], **kwargs) + if text.endswith((".db", ".sqlite")): + return SqliteStore(text, **kwargs) + return JsonlStore(text) diff --git a/examples/control_plane/__init__.py b/examples/control_plane/__init__.py new file mode 100644 index 0000000..f8dfcf9 --- /dev/null +++ b/examples/control_plane/__init__.py @@ -0,0 +1,6 @@ +"""A minimal reference Chronicle control plane. + +A shared HTTP service that deployed agents ship envelopes to via ``RemoteStore``, backed +by SQLite. Zero dependency (stdlib ``http.server`` + ``sqlite3``). TokenOps can point at +the same service to co-locate cost ledger and trace storage. See ``server.py``. +""" diff --git a/examples/control_plane/server.py b/examples/control_plane/server.py new file mode 100644 index 0000000..3b57bf7 --- /dev/null +++ b/examples/control_plane/server.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Reference Chronicle control plane: a shared HTTP store for deployed agents. + +Deployed agents record to it with ``chronicle.record(store=RemoteStore(url))``. It keeps +envelopes in SQLite and serves them back for inspection. Zero dependency. + +Run: + + python -m examples.control_plane.server --port 8900 --db control_plane.db + +Endpoints: + + POST /envelopes store one envelope (JSON body) + GET /envelopes list all envelopes + GET /traces//envelopes list a trace's envelopes + GET /envelopes/ one envelope (as a list) + GET /health {"status": "ok"} + +This is a reference, not a hardened service: add auth, retention, and TLS before real +use. TokenOps can point at the same host to co-locate its cost ledger with trace storage. +""" + +from __future__ import annotations + +import argparse +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +from chronicle.envelope.backends import SqliteStore +from chronicle.envelope.schema import Envelope + + +def make_server(host: str = "127.0.0.1", port: int = 8900, db: str = ":memory:") -> ThreadingHTTPServer: + """Build (but do not start) a control-plane server backed by a SqliteStore at ``db``.""" + store = SqliteStore(db) + + class Handler(BaseHTTPRequestHandler): + def _send(self, code: int, payload: object | None = None) -> None: + body = json.dumps(payload).encode("utf-8") if payload is not None else b"" + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if body: + self.wfile.write(body) + + def _dump(self, envelopes) -> list[dict]: + return [json.loads(e.model_dump_json()) for e in envelopes] + + def do_POST(self) -> None: + if self.path.rstrip("/") == "/envelopes": + length = int(self.headers.get("Content-Length", 0)) + raw = self.rfile.read(length) + try: + envelope = Envelope.from_json(raw) + except Exception as exc: # noqa: BLE001 - report bad input, do not crash + self._send(400, {"error": f"invalid envelope: {exc}"}) + return + store.append(envelope) + self._send(201, {"status": "stored", "envelope_id": envelope.envelope_id}) + else: + self._send(404, {"error": "not found"}) + + def do_GET(self) -> None: + path = self.path.rstrip("/") or "/" + if path == "/health": + self._send(200, {"status": "ok"}) + elif path == "/envelopes": + self._send(200, self._dump(store.read_all())) + elif path.startswith("/traces/") and path.endswith("/envelopes"): + trace_id = path[len("/traces/"):-len("/envelopes")] + self._send(200, self._dump(store.find_by_trace_id(trace_id))) + elif path.startswith("/envelopes/"): + envelope = store.find_by_envelope_id(path[len("/envelopes/"):]) + self._send(200, self._dump([envelope] if envelope else [])) + else: + self._send(404, {"error": "not found"}) + + def log_message(self, *args) -> None: # quiet by default + return + + server = ThreadingHTTPServer((host, port), Handler) + server.store = store # type: ignore[attr-defined] + return server + + +def main() -> None: + parser = argparse.ArgumentParser(description="Chronicle control plane (reference)") + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8900) + parser.add_argument("--db", default="control_plane.db") + args = parser.parse_args() + + server = make_server(args.host, args.port, args.db) + print(f"Chronicle control plane on http://{args.host}:{args.port} (db={args.db})") + print("POST /envelopes | GET /envelopes | GET /traces//envelopes | GET /health") + try: + server.serve_forever() + except KeyboardInterrupt: + server.shutdown() + + +if __name__ == "__main__": + main() diff --git a/tests/test_stores.py b/tests/test_stores.py new file mode 100644 index 0000000..71f4efb --- /dev/null +++ b/tests/test_stores.py @@ -0,0 +1,114 @@ +"""Storage backends: SQLite and Remote satisfy the Store protocol, round-trip +envelopes, and drive chronicle.record. open_store dispatches by target string, and a +RemoteStore talks to the reference control-plane server end to end. +""" + +from __future__ import annotations + +import threading +import warnings + +import chronicle +from chronicle import ( + EnvelopeStore, + JsonlStore, + RemoteStore, + SqliteStore, + Store, + boundary, + open_store, +) +from chronicle.envelope.schema import ActionResult, ContextMetadata, Envelope, InputState +from examples.control_plane.server import make_server + + +def _env(trace_id: str, seq: int, node: str = "agent") -> Envelope: + return Envelope( + node_id=node, + boundary_kind="tool", + trace_id=trace_id, + sequence=seq, + metadata=ContextMetadata(model_version="m", build_id="b"), + input_state=InputState(messages=[]), + action_result=ActionResult(completion=f"ok-{seq}"), + ) + + +def test_sqlite_roundtrip(tmp_path): + store = SqliteStore(tmp_path / "runs.db") + a, b, c = _env("t1", 1), _env("t1", 2), _env("t2", 1) + for e in (a, b, c): + store.append(e) + assert len(store.read_all()) == 3 + assert [e.sequence for e in store.find_by_trace_id("t1")] == [1, 2] + assert store.find_by_envelope_id(a.envelope_id).action_result.completion == "ok-1" + assert store.find_by_envelope_id("missing") is None + store.close() + + +def test_backends_satisfy_store_protocol(tmp_path): + assert isinstance(SqliteStore(":memory:"), Store) + assert isinstance(JsonlStore(tmp_path / "r.jsonl"), Store) + assert isinstance(RemoteStore("http://localhost:1"), Store) + + +def test_open_store_dispatch(tmp_path): + assert isinstance(open_store(tmp_path / "r.jsonl"), EnvelopeStore) + assert isinstance(open_store(str(tmp_path / "r.db")), SqliteStore) + assert isinstance(open_store("sqlite:///" + str(tmp_path / "x.db")), SqliteStore) + assert isinstance(open_store("https://cp.example"), RemoteStore) + + +def test_record_into_sqlite(tmp_path): + store = SqliteStore(tmp_path / "runs.db") + + with chronicle.record("t-rec", store=store): + + @boundary("agent", kind="tool") + def do(x): + return {"ok": x} + + do(1) + + assert len(store.find_by_trace_id("t-rec")) == 1 + store.close() + + +def test_record_with_sqlite_url_string(tmp_path): + url = "sqlite:///" + str(tmp_path / "u.db") + with chronicle.record("t-url", store=url): + + @boundary("agent", kind="tool") + def do(x): + return {"ok": x} + + do(1) + + assert len(open_store(url).find_by_trace_id("t-url")) == 1 + + +def test_remote_store_end_to_end(): + server = make_server("127.0.0.1", 0) # port 0 -> ephemeral + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + store = RemoteStore(f"http://127.0.0.1:{port}") + a, b = _env("rt", 1), _env("rt", 2) + store.append(a) + store.append(b) + assert [e.sequence for e in store.find_by_trace_id("rt")] == [1, 2] + assert len(store.read_all()) == 2 + one = store.find_by_envelope_id(a.envelope_id) + assert one is not None and one.envelope_id == a.envelope_id + assert store.find_by_envelope_id("missing") is None + finally: + server.shutdown() + + +def test_remote_store_append_never_raises_the_agent(): + store = RemoteStore("http://127.0.0.1:1", timeout=0.2) # nothing listening + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + store.append(_env("x", 1)) # must not raise + assert store.read_all() == [] # read failure returns empty, not an error