Add CHRONICLE_ENABLED env flag to disable LIVE recording

CHANGELOG.md

@@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **`CHRONICLE_ENABLED`**: set to `0` / `false` / `off` / `no` to turn off LIVE
+  recording. `@boundary`, `wrap`, `wrap_llm`, `record()`, and `EnvelopeRecorder`
+  become passthrough so an agent can be run with and without Chronicle. Replay is
+  unaffected. Check with `chronicle.is_enabled()`.
+- **`BufferedStore`**: in-memory buffer with batched flush over any inner store
+  (`JsonlStore.append_many` for one open/write). Also
+  `open_store("buffered:32:runs.jsonl")`.
+- **Recording hot-path speedups**: cache `inspect.signature` per boundary,
+  dataclass-aware `_json_safe`, `Envelope.model_construct` on LIVE record,
+  `JsonlStore(keep_open=True)`, and `retain_envelopes=` on `record()` /
+  session (skip in-memory list when only the store write is needed).
+- **BufferedStore durability**: `record()` flushes the store on context exit;
+  failed flushes restore the in-memory batch instead of dropping it.
+- **Message capture**: `_json_safe` keeps full dataclass / duck-typed message
+  fields (not just `role`/`content`); every messages entry is coerced to a dict.
+
 ## [0.3.0] - 2026-07-24
 
 ### Added

README.md

@@ -464,6 +464,7 @@ See `examples/langgraph_demo/agent.py`.
 
 | Variable | Purpose |
 |---|---|
+| `CHRONICLE_ENABLED` | Set to `0` / `false` / `off` / `no` to disable LIVE recording (`@boundary`, `wrap`, `record()`, `EnvelopeRecorder` become passthrough). Default on. Replay is unaffected. |
 | `CHRONICLE_BUILD_ID` | Pin runtime build ID in envelope metadata |
 | `CHRONICLE_STORE` | Default envelope store path |
 | `PHOENIX_COLLECTOR_ENDPOINT` | Phoenix OTLP endpoint (default `http://localhost:4317`) |
@@ -578,7 +579,7 @@ during the live run, so tests are deterministic, free, and fast. Only the option
 `@boundary` and `wrap()` are **transparent**: they never change what your function
 returns or raises. Recording adds one wrapper call and one JSON append per boundary
 crossing, so the cost scales with how many boundaries you mark, not with anything in a
-hot loop. Enable it where you want a record; leave it off elsewhere.
+hot loop. Set `CHRONICLE_ENABLED=0` to make LIVE recording a no-op for A/B runs.
 </details>
 
 <details>

chronicle/__init__.py

@@ -2,6 +2,7 @@
 
 from chronicle.api import record, replay_trace
 from chronicle.boundary import boundary, wrap_llm
+from chronicle.config import is_enabled
 from chronicle.envelope.schema import (
     ActionResult,
     ContextMetadata,
@@ -13,6 +14,7 @@
     ToolSchema,
 )
 from chronicle.envelope.backends import (
+    BufferedStore,
     JsonlStore,
     RemoteStore,
     SqliteStore,
@@ -42,6 +44,7 @@ def __getattr__(name: str):
 __all__ = [
     "ActionResult",
     "BoundaryMode",
+    "BufferedStore",
     "ChronicleSession",
     "ContextMetadata",
     "Envelope",
@@ -65,6 +68,7 @@ def __getattr__(name: str):
     "get_session",
     "instrument_langgraph",
     "instrument_otel",
+    "is_enabled",
     "open_store",
     "record",
     "redact_secrets",

chronicle/api.py

@@ -12,6 +12,7 @@
 from contextlib import contextmanager
 from pathlib import Path
 
+from chronicle.config import is_enabled
 from chronicle.envelope.backends import Store, open_store
 from chronicle.replay.plan import ReplayPlan
 from chronicle.session import ChronicleSession, reset_session
@@ -26,6 +27,7 @@ def record(
     build_id: str | None = None,
     redactors: list[Callable[[str], str]] | None = None,
     export: str | Path | None = None,
+    retain_envelopes: bool = True,
 ) -> Iterator[ChronicleSession]:
     """Record a run in one block.
 
@@ -39,8 +41,18 @@ def record(
             export="fixtures/traces/incident-001/",
         ) as session:
             run_agent(...)
+
+    When ``CHRONICLE_ENABLED`` is off, this is a no-op: yields a fresh session
+    with no store and does not export. Boundaries inside the block also skip
+    LIVE recording.
+
+    Set ``retain_envelopes=False`` when you only need the store write (skips the
+    in-session list; ``export_trace`` will be empty).
     """
     session = reset_session()
+    if not is_enabled():
+        yield session
+        return
     if store is not None:
         # 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
@@ -52,11 +64,21 @@ def record(
         session.build_id = build_id
     if redactors is not None:
         session.redactors = redactors
+    session.retain_envelopes = retain_envelopes
     session.begin_trace(trace_id)
-    yield session
+    try:
+        yield session
+    finally:
+        # Buffered (and keep-open) stores must flush so a short run or remainder
+        # batch is not left only in memory when the context exits.
+        store_obj = session.store
+        if store_obj is not None:
+            flush = getattr(store_obj, "flush", None)
+            if callable(flush):
+                flush()
     # Export only on a clean exit, so a crash mid-run doesn't overwrite a fixture
     # with a partial trace. Call session.export_trace(...) yourself if you need it.
-    if export is not None:
+    if export is not None and retain_envelopes:
         session.export_trace(export)
 
 

chronicle/boundary.py

@@ -16,16 +16,19 @@
 
 from __future__ import annotations
 
+import dataclasses
 import functools
 import inspect
 from collections.abc import Callable, Mapping
 from typing import Any, TypeVar
 
+from chronicle.config import is_enabled
 from chronicle.envelope.schema import ActionResult, InputState
 from chronicle.session import (
     SessionMode,
     get_session,
     model_version_from,
+    peek_session,
     result_to_action_result,
     sampling_params_from,
 )
@@ -119,40 +122,58 @@ def _bind_boundary(
 ) -> Callable[..., Any]:
     """Shared LIVE / replay wrapper for ``@boundary`` and ``wrap_llm`` (sync + async)."""
 
+    # Resolve once — inspect.signature dominates per-crossing cost otherwise.
+    try:
+        cached_sig: inspect.Signature | None = inspect.signature(fn)
+    except (TypeError, ValueError):
+        cached_sig = None
+
     if inspect.iscoroutinefunction(fn):
 
         @functools.wraps(fn)
         async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
-            session = get_session()
+            if not is_enabled():
+                # Passthrough for LIVE; still honor an active REPLAY session.
+                session = peek_session()
+                if session is None or session.mode == SessionMode.LIVE:
+                    return await fn(*args, **kwargs)
+            else:
+                session = get_session()
             if session.mode == SessionMode.LIVE:
                 return await _record_call_async(
                     session, fn, boundary_id, kind, args, kwargs,
-                    extract_input, extract_result, extract_metadata,
+                    extract_input, extract_result, extract_metadata, cached_sig,
                 )
             invocation_index = session._replay_cursor.get(boundary_id, 0) + 1
             if session.replay_plan.should_stub(boundary_id, invocation_index):
                 return session.stub_result(boundary_id, kind)
             return await _live_cutpoint_call_async(
                 session, fn, boundary_id, kind, args, kwargs,
-                extract_input, invocation_index,
+                extract_input, invocation_index, cached_sig,
             )
 
         return async_wrapper
 
     @functools.wraps(fn)
     def wrapper(*args: Any, **kwargs: Any) -> Any:
-        session = get_session()
+        if not is_enabled():
+            # Passthrough for LIVE; still honor an active REPLAY session.
+            session = peek_session()
+            if session is None or session.mode == SessionMode.LIVE:
+                return fn(*args, **kwargs)
+        else:
+            session = get_session()
         if session.mode == SessionMode.LIVE:
             return _record_call(
                 session, fn, boundary_id, kind, args, kwargs,
-                extract_input, extract_result, extract_metadata,
+                extract_input, extract_result, extract_metadata, cached_sig,
             )
         invocation_index = session._replay_cursor.get(boundary_id, 0) + 1
         if session.replay_plan.should_stub(boundary_id, invocation_index):
             return session.stub_result(boundary_id, kind)
         return _live_cutpoint_call(
             session, fn, boundary_id, kind, args, kwargs,
-            extract_input, invocation_index,
+            extract_input, invocation_index, cached_sig,
         )
 
     return wrapper
@@ -183,8 +204,11 @@ def _run_on_leave(session, boundary_id, kind, input_state, entered: bool) -> Non
         session.on_leave(boundary_id, kind, input_state)
 
 
-def _record_call(session, fn, boundary_id, kind, args, kwargs, extract_input, extract_result, extract_metadata):
-    input_state = _capture_input(fn, args, kwargs, extract_input)
+def _record_call(
+    session, fn, boundary_id, kind, args, kwargs,
+    extract_input, extract_result, extract_metadata, cached_sig=None,
+):
+    input_state = _capture_input(fn, args, kwargs, extract_input, cached_sig)
     call_kwargs, entered = _apply_on_enter(session, boundary_id, kind, input_state, kwargs)
     try:
         try:
@@ -198,8 +222,11 @@ def _record_call(session, fn, boundary_id, kind, args, kwargs, extract_input, ex
         _run_on_leave(session, boundary_id, kind, input_state, entered)
 
 
-async def _record_call_async(session, fn, boundary_id, kind, args, kwargs, extract_input, extract_result, extract_metadata):
-    input_state = _capture_input(fn, args, kwargs, extract_input)
+async def _record_call_async(
+    session, fn, boundary_id, kind, args, kwargs,
+    extract_input, extract_result, extract_metadata, cached_sig=None,
+):
+    input_state = _capture_input(fn, args, kwargs, extract_input, cached_sig)
     call_kwargs, entered = _apply_on_enter(session, boundary_id, kind, input_state, kwargs)
     try:
         try:
@@ -256,8 +283,10 @@ def _call_metadata(result, kind, extract_metadata):
 # Cut-point (REPLAY mode, live boundary). No envelope; capture for assertions.
 # --------------------------------------------------------------------------- #
 
-def _live_cutpoint_call(session, fn, boundary_id, kind, args, kwargs, extract_input, invocation_index):
-    input_state = _capture_input(fn, args, kwargs, extract_input)
+def _live_cutpoint_call(
+    session, fn, boundary_id, kind, args, kwargs, extract_input, invocation_index, cached_sig=None,
+):
+    input_state = _capture_input(fn, args, kwargs, extract_input, cached_sig)
     session.capture_live_input(boundary_id, invocation_index, input_state)
     call_kwargs, entered = _apply_on_enter(session, boundary_id, kind, input_state, kwargs)
     try:
@@ -272,8 +301,10 @@ def _live_cutpoint_call(session, fn, boundary_id, kind, args, kwargs, extract_in
         _run_on_leave(session, boundary_id, kind, input_state, entered)
 
 
-async def _live_cutpoint_call_async(session, fn, boundary_id, kind, args, kwargs, extract_input, invocation_index):
-    input_state = _capture_input(fn, args, kwargs, extract_input)
+async def _live_cutpoint_call_async(
+    session, fn, boundary_id, kind, args, kwargs, extract_input, invocation_index, cached_sig=None,
+):
+    input_state = _capture_input(fn, args, kwargs, extract_input, cached_sig)
     session.capture_live_input(boundary_id, invocation_index, input_state)
     call_kwargs, entered = _apply_on_enter(session, boundary_id, kind, input_state, kwargs)
     try:
@@ -307,38 +338,43 @@ def _advance_cutpoint(session, boundary_id, invocation_index):
 _IO_KEYS = ("messages", "system_prompt", "rag_chunks")
 
 
-def _capture_input(fn, args, kwargs, extract_input) -> InputState:
+def _capture_input(fn, args, kwargs, extract_input, cached_sig=None) -> InputState:
     if extract_input is not None:
         return extract_input(*args, **kwargs)
     # The default capture must never break the wrapped call.
     try:
-        return _bind_input_state(fn, args, kwargs)
+        return _bind_input_state(fn, args, kwargs, cached_sig)
     except Exception:
         return InputState(messages=[], graph_state={"args": _json_safe(list(args)), "kwargs": _json_safe(dict(kwargs))})
 
 
-def _bind_input_state(fn, args, kwargs) -> InputState:
-    graph_state = _bound_arguments(fn, args, kwargs)
+def _bind_input_state(fn, args, kwargs, cached_sig=None) -> InputState:
+    graph_state = _bound_arguments(fn, args, kwargs, cached_sig)
     source = _io_source(graph_state)
     messages = source.get("messages") or []
     if not messages and "user_message" in source:
         messages = [{"role": "user", "content": source["user_message"]}]
-    return InputState(
+    # Messages must be dicts for the envelope schema; coerce each non-mapping row.
+    if messages:
+        messages = [m if isinstance(m, Mapping) else _json_safe(m) for m in messages]
+    return InputState.model_construct(
         messages=messages,
         system_prompt=source.get("system_prompt"),
         rag_chunks=_coerce_rag_chunks(source.get("rag_chunks")),
         graph_state=graph_state,
     )
 
 
-def _bound_arguments(fn, args, kwargs) -> dict[str, Any]:
+def _bound_arguments(fn, args, kwargs, cached_sig=None) -> dict[str, Any]:
     """Record the call by real parameter names. Skip self/cls, flatten **kwargs.
 
     Falls back to positional capture when the callable has no introspectable
     signature (some builtins / C callables).
     """
+    sig = cached_sig
     try:
-        sig = inspect.signature(fn)
+        if sig is None:
+            sig = inspect.signature(fn)
         bound = sig.bind_partial(*args, **kwargs)
     except (TypeError, ValueError):
         return {"args": _json_safe(list(args)), "kwargs": _json_safe(dict(kwargs))}
@@ -395,9 +431,27 @@ def _json_safe(value: Any, _depth: int = 0) -> Any:
         return {str(k): _json_safe(v, _depth + 1) for k, v in value.items()}
     if isinstance(value, (list, tuple, set)):
         return [_json_safe(v, _depth + 1) for v in value]
+    # Prefer full structured dumps so tool_calls / name / id are not stripped.
+    if dataclasses.is_dataclass(value) and not isinstance(value, type):
+        return {
+            f.name: _json_safe(getattr(value, f.name), _depth + 1)
+            for f in dataclasses.fields(value)
+        }
     if hasattr(value, "model_dump"):
         try:
             return value.model_dump()
         except Exception:
             return repr(value)
+    # Duck-typed chat message: keep every public attribute, not just role/content.
+    role = getattr(value, "role", None)
+    content = getattr(value, "content", None)
+    if isinstance(role, str) and isinstance(content, str):
+        data = getattr(value, "__dict__", None)
+        if isinstance(data, dict) and data:
+            return {
+                str(k): _json_safe(v, _depth + 1)
+                for k, v in data.items()
+                if not str(k).startswith("_")
+            }
+        return {"role": role, "content": content}
     return repr(value)

chronicle/config.py

@@ -0,0 +1,23 @@
+"""Runtime configuration from the environment."""
+
+from __future__ import annotations
+
+import os
+
+# Explicit falsy tokens. Unset means enabled (backward compatible).
+_DISABLED = frozenset({"0", "false", "no", "off"})
+
+
+def is_enabled() -> bool:
+    """Whether Chronicle LIVE recording / instrumentation is active.
+
+    Controlled by ``CHRONICLE_ENABLED`` (default on). Set to ``0``, ``false``,
+    ``off``, or ``no`` to make ``@boundary``, ``wrap``, ``wrap_llm``,
+    ``record()``, and ``EnvelopeRecorder`` no-ops for live runs so an agent can
+    be timed with and without Chronicle. Replay is unaffected so cut-point
+    fixtures keep working.
+    """
+    raw = os.environ.get("CHRONICLE_ENABLED")
+    if raw is None:
+        return True
+    return raw.strip().lower() not in _DISABLED

chronicle/envelope/__init__.py

@@ -1,4 +1,5 @@
 from chronicle.envelope.backends import (
+    BufferedStore,
     JsonlStore,
     RemoteStore,
     SqliteStore,
@@ -10,6 +11,7 @@
 from chronicle.envelope.store import EnvelopeStore
 
 __all__ = [
+    "BufferedStore",
     "Envelope",
     "EnvelopeRecorder",
     "EnvelopeStore",