Perf: speed up LIVE recording hot path

CHANGELOG.md

@@ -15,6 +15,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **`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
 

chronicle/api.py

@@ -27,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.
 
@@ -44,6 +45,9 @@ def record(
     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():
@@ -60,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,6 +16,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import functools
 import inspect
 from collections.abc import Callable, Mapping
@@ -121,6 +122,12 @@ 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)
@@ -135,14 +142,14 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
             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
@@ -159,14 +166,14 @@ def wrapper(*args: Any, **kwargs: Any) -> Any:
         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
@@ -197,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:
@@ -212,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:
@@ -270,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:
@@ -286,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:
@@ -321,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))}
@@ -409,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/envelope/backends.py

@@ -88,10 +88,20 @@ def _flush_unlocked(self) -> None:
         self._buf = []
         append_many = getattr(self.inner, "append_many", None)
         if callable(append_many):
-            append_many(batch)
-        else:
-            for envelope in batch:
+            try:
+                append_many(batch)
+            except Exception:
+                # Nothing committed — put the batch back ahead of any newer appends.
+                self._buf = batch + self._buf
+                raise
+            return
+        for i, envelope in enumerate(batch):
+            try:
                 self.inner.append(envelope)
+            except Exception:
+                # Keep the failed envelope and everything after it.
+                self._buf = batch[i:] + self._buf
+                raise
 
     def read_all(self) -> list[Envelope]:
         self.flush()
@@ -247,11 +257,19 @@ def open_store(target: str | Path, **kwargs) -> Store:
         batch_kwargs = {k: v for k, v in kwargs.items() if k == "batch_size"}
         inner_kwargs = {k: v for k, v in kwargs.items() if k != "batch_size"}
         batch_size = int(batch_kwargs.get("batch_size", size_str))
+        # Prefer a kept-open JSONL handle under the buffer — one fd for the run.
+        if "keep_open" not in inner_kwargs and not str(inner).startswith(
+            ("http://", "https://", "sqlite:///", "buffered:")
+        ) and not str(inner).endswith((".db", ".sqlite")):
+            inner_kwargs["keep_open"] = True
         return BufferedStore(open_store(inner, **inner_kwargs), batch_size=batch_size)
     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)
+    keep_open = bool(kwargs.pop("keep_open", False))
+    if kwargs:
+        raise TypeError(f"unexpected open_store kwargs for JsonlStore: {sorted(kwargs)}")
+    return JsonlStore(text, keep_open=keep_open)

chronicle/envelope/store.py

@@ -3,33 +3,49 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import IO, Text
 
 from chronicle.envelope.schema import Envelope
 
 
 class EnvelopeStore:
-    """Append-only JSONL store for immutable envelope records."""
+    """Append-only JSONL store for immutable envelope records.
 
-    def __init__(self, path: str | Path) -> None:
+    Set ``keep_open=True`` to hold one append file handle across writes (and
+    ``append_many`` flushes). Call ``close()`` when done — ``BufferedStore`` does
+    this on context exit.
+    """
+
+    def __init__(self, path: str | Path, *, keep_open: bool = False) -> None:
         self.path = Path(path)
         self.path.parent.mkdir(parents=True, exist_ok=True)
         if not self.path.exists():
             self.path.touch()
+        self._fp: IO[str] | None = None
+        if keep_open:
+            self._fp = open(self.path, "a", encoding="utf-8")
 
     def append(self, envelope: Envelope) -> None:
-        line = envelope.model_dump_json()
+        line = envelope.model_dump_json() + "\n"
+        if self._fp is not None:
+            self._fp.write(line)
+            return
         with open(self.path, "a", encoding="utf-8") as f:
-            f.write(line + "\n")
+            f.write(line)
 
     def append_many(self, envelopes: list[Envelope]) -> None:
-        """Append a batch in one open/write — used by ``BufferedStore`` flushes."""
+        """Append a batch in one write — used by ``BufferedStore`` flushes."""
         if not envelopes:
             return
         payload = "".join(e.model_dump_json() + "\n" for e in envelopes)
+        if self._fp is not None:
+            self._fp.write(payload)
+            return
         with open(self.path, "a", encoding="utf-8") as f:
             f.write(payload)
 
     def read_all(self) -> list[Envelope]:
+        self.flush()
         envelopes: list[Envelope] = []
         with open(self.path, encoding="utf-8") as f:
             for line in f:
@@ -47,6 +63,22 @@ def find_by_envelope_id(self, envelope_id: str) -> Envelope | None:
                 return envelope
         return None
 
+    def flush(self) -> None:
+        if self._fp is not None:
+            self._fp.flush()
+
+    def close(self) -> None:
+        if self._fp is not None:
+            self._fp.flush()
+            self._fp.close()
+            self._fp = None
+
+    def __enter__(self) -> EnvelopeStore:
+        return self
+
+    def __exit__(self, *exc: object) -> None:
+        self.close()
+
     @staticmethod
     def load_fixture(path: str | Path) -> Envelope:
         return Envelope.from_file(str(path))