feat(lifecycle): add in-memory execution recorder

burr/lifecycle/__init__.py

@@ -29,7 +29,12 @@
     PreRunStepHookAsync,
     PreStartSpanHook,
 )
-from burr.lifecycle.default import StateAndResultsFullLogger
+from burr.lifecycle.default import (
+    ExecutionRecord,
+    InMemoryExecutionRecorder,
+    StateAndResultsFullLogger,
+    StateChange,
+)
 
 __all__ = [
     "PreRunStepHook",
@@ -42,6 +47,9 @@
     "PostApplicationExecuteCallHookAsync",
     "LifecycleAdapter",
     "StateAndResultsFullLogger",
+    "InMemoryExecutionRecorder",
+    "ExecutionRecord",
+    "StateChange",
     "PostApplicationCreateHook",
     "PostEndSpanHook",
     "PreStartSpanHook",

burr/lifecycle/default.py

@@ -15,10 +15,12 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import copy
+import dataclasses
 import datetime
 import json
 import time
-from typing import TYPE_CHECKING, Any, Callable, Literal, Optional
+from typing import TYPE_CHECKING, Any, Callable, Dict, Literal, Optional
 
 if TYPE_CHECKING:
     from burr.core import State, Action
@@ -30,6 +32,139 @@ def safe_json(obj: Any) -> str:
     return json.dumps(obj, default=str)
 
 
+def _values_equal(left: Any, right: Any) -> bool:
+    if left is right:
+        return True
+    try:
+        comparison = left == right
+    except Exception:
+        return False
+    return comparison if isinstance(comparison, bool) else False
+
+
+def _copy_for_recording(value: Any) -> Any:
+    try:
+        return copy.deepcopy(value)
+    except Exception:
+        return value
+
+
+def _copy_mapping_for_recording(values: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+    if values is None:
+        return None
+    return {key: _copy_for_recording(value) for key, value in values.items()}
+
+
+@dataclasses.dataclass(frozen=True)
+class StateChange:
+    """A single state field change captured around an action execution."""
+
+    key: str
+    before_exists: bool
+    before: Any
+    after_exists: bool
+    after: Any
+
+
+@dataclasses.dataclass(frozen=True)
+class ExecutionRecord:
+    """An immutable record of one Burr action execution."""
+
+    app_id: str
+    partition_key: Optional[str]
+    sequence_id: int
+    action: str
+    inputs: Dict[str, Any]
+    result: Optional[Dict[str, Any]]
+    exception: Optional[Exception]
+    state_changes: tuple[StateChange, ...]
+
+
+class InMemoryExecutionRecorder(PreRunStepHook, PostRunStepHook):
+    """Capture action executions in memory for tests and local debugging.
+
+    The recorder stores inputs, results, exceptions, and business-state changes.
+    Burr's private ``__`` state fields are excluded from the change list.
+    """
+
+    def __init__(self):
+        self._records: list[ExecutionRecord] = []
+        self._pending: dict[tuple[str, Optional[str], int], tuple[dict, dict]] = {}
+
+    @property
+    def records(self) -> tuple[ExecutionRecord, ...]:
+        """Return an immutable snapshot of captured execution records."""
+        return tuple(self._records)
+
+    def clear(self) -> None:
+        """Remove all completed and pending records."""
+        self._records.clear()
+        self._pending.clear()
+
+    def pre_run_step(
+        self,
+        *,
+        app_id: str,
+        partition_key: Optional[str],
+        sequence_id: int,
+        state: "State",
+        inputs: Dict[str, Any],
+        **future_kwargs: Any,
+    ) -> None:
+        key = (app_id, partition_key, sequence_id)
+        self._pending[key] = (
+            _copy_mapping_for_recording(state.get_all()),
+            _copy_mapping_for_recording(inputs),
+        )
+
+    def post_run_step(
+        self,
+        *,
+        app_id: str,
+        partition_key: Optional[str],
+        sequence_id: int,
+        state: "State",
+        action: "Action",
+        result: Optional[Dict[str, Any]],
+        exception: Optional[Exception],
+        **future_kwargs: Any,
+    ) -> None:
+        key = (app_id, partition_key, sequence_id)
+        before, inputs = self._pending.pop(key)
+        after = state.get_all()
+        changed_keys = sorted(
+            key
+            for key in before.keys() | after.keys()
+            if not key.startswith("__")
+            and (
+                (key in before) != (key in after)
+                or not _values_equal(before.get(key), after.get(key))
+            )
+        )
+        changes = tuple(
+            StateChange(
+                key=state_key,
+                before_exists=state_key in before,
+                before=_copy_for_recording(before.get(state_key)),
+                after_exists=state_key in after,
+                after=_copy_for_recording(after.get(state_key)),
+            )
+            for state_key in changed_keys
+        )
+        self._records.append(
+            ExecutionRecord(
+                app_id=app_id,
+                partition_key=partition_key,
+                sequence_id=sequence_id,
+                action=action.name,
+                inputs=inputs,
+                result=_copy_mapping_for_recording(result),
+                exception=exception,
+                state_changes=changes,
+            )
+        )
+
+
 class StateAndResultsFullLogger(PostRunStepHook, PreRunStepHook):
     """Logs the state and results of the action in a jsonl file."""
 

docs/concepts/hooks.rst

@@ -74,6 +74,32 @@ To include this in the application, you pass it into the :py:meth:`with_hooks <b
         ...
         .build())
 
+Recording executions in tests
+-----------------------------
+
+``InMemoryExecutionRecorder`` captures action inputs, results, exceptions, and
+business-state changes without configuring a tracking backend. This is useful for
+asserting an agent's route or inspecting a local execution loop:
+
+.. code-block:: python
+
+    from burr.lifecycle import InMemoryExecutionRecorder
+
+    recorder = InMemoryExecutionRecorder()
+    app = ApplicationBuilder().with_hooks(recorder).build()
+
+    app.run(halt_after=["final_answer"])
+    assert [record.action for record in recorder.records] == [
+        "plan",
+        "call_tool",
+        "final_answer",
+    ]
+
+Each record is immutable and includes ``app_id``, ``partition_key``,
+``sequence_id``, inputs, result, exception, and a tuple of ``StateChange`` values.
+Private Burr state keys prefixed with ``__`` are excluded. Call ``clear()`` to
+reuse the recorder between tests.
+
 .. note::
 
     There are synchronous and asynchronous hooks. Synchronous hooks will be called with both synchronous and asynchronous run methods

docs/reference/lifecycle.rst

@@ -70,6 +70,15 @@ and add instances to the application builder to customize your state machines's
 
 These hooks are available for you to use:
 
+.. autoclass:: burr.lifecycle.default.InMemoryExecutionRecorder
+   :members:
+
+.. autoclass:: burr.lifecycle.default.ExecutionRecord
+   :members:
+
+.. autoclass:: burr.lifecycle.default.StateChange
+   :members:
+
 .. autoclass:: burr.lifecycle.default.StateAndResultsFullLogger
 
    .. automethod:: __init__