diff --git a/docs/docs/tutorials/observability/index.md b/docs/docs/tutorials/observability/index.md
index d5ae66ca2c..0ad3d52995 100644
--- a/docs/docs/tutorials/observability/index.md
+++ b/docs/docs/tutorials/observability/index.md
@@ -205,6 +205,10 @@ Sometimes, you may want to implement a custom logging solution. For instance, yo
 |`on_adapter_parse_start` / `on_adapter_parse_end`| Triggered when a `dspy.Adapter` subclass postprocess the output text from an LM. |
 |`on_tool_start` / `on_tool_end` | Triggered when a `dspy.Tool` subclass is invoked. |
 |`on_evaluate_start` / `on_evaluate_end` | Triggered when a `dspy.Evaluate` instance is invoked. |
+|`on_interpreter_execute_start` / `on_interpreter_execute_end` | Triggered around a `CodeInterpreter.execute()` call (e.g. each `dspy.RLM` REPL cell). |
+|`on_interpreter_tool_call_start` / `on_interpreter_tool_call_end` | Triggered around each sandbox->host tool dispatch, including plain closures like RLM's `llm_query`. |
+|`on_interpreter_startup_start` / `on_interpreter_startup_end` | Triggered around `CodeInterpreter.start()` (e.g. Deno process spawn). |
+|`on_interpreter_shutdown_start` / `on_interpreter_shutdown_end` | Triggered around `CodeInterpreter.shutdown()`. |
 
 Here's an example of custom callback that logs the intermediate steps of a ReAct agent:
 
@@ -245,3 +249,49 @@ dspy.configure(callbacks=[AgentLoggingCallback()])
 !!! info "Handling Inputs and Outputs in Callbacks"
 
     Be cautious when working with input or output data in callbacks. Mutating them in-place can modify the original data passed to the program, potentially leading to unexpected behavior. To avoid this, it's strongly recommended to create a copy of the data before performing any operations that may alter it.
+
+### Observing code execution with interpreter callbacks
+
+Code-executing modules such as [`dspy.RLM`](../../api/modules/RLM.md) run inside a sandboxed
+interpreter. The `on_interpreter_*` handlers let you observe each REPL cell in real time,
+along with the sandbox->host tool calls it triggers (including RLM's built-in `llm_query`).
+The example below logs per-cell execution time and every sub-tool call:
+
+```python
+import time
+import dspy
+from dspy.utils.callback import BaseCallback
+
+class InterpreterLoggingCallback(BaseCallback):
+    def __init__(self):
+        self._start = {}
+
+    def on_interpreter_execute_start(self, call_id, instance, inputs):
+        # inputs["variables"] can be very large (RLM passes input data through it every
+        # iteration), so avoid printing it verbatim - truncation is the handler's job.
+        self._start[call_id] = time.perf_counter()
+        code = inputs.get("code", "")
+        print(f"[cell] executing:\n{code[:200]}")
+
+    def on_interpreter_execute_end(self, call_id, outputs, exception):
+        elapsed = time.perf_counter() - self._start.pop(call_id, time.perf_counter())
+        status = f"error: {type(exception).__name__}" if exception else "ok"
+        print(f"[cell] finished in {elapsed:.3f}s ({status})")
+
+    def on_interpreter_tool_call_start(self, call_id, instance, inputs):
+        print(f"[tool] {inputs['tool_name']}(**{inputs['kwargs']})")
+
+dspy.configure(callbacks=[InterpreterLoggingCallback()])
+
+rlm = dspy.RLM("context, query -> answer")
+rlm(context="...very long text...", query="What is the magic number?")
+```
+
+Because the tool call runs synchronously inside `execute()`, its `call_id` nests under the
+enclosing cell's `call_id`, so you can reconstruct which cell issued which sub-tool call.
+
+!!! info "Custom interpreters"
+
+    The `CodeInterpreter` protocol cannot enforce these hooks. `dspy.PythonInterpreter` emits them
+    out of the box; a custom interpreter should decorate its `execute`/`start`/`shutdown` methods
+    (and its tool-dispatch seam) with `dspy.utils.callback.with_callbacks` to participate.
diff --git a/dspy/primitives/code_interpreter.py b/dspy/primitives/code_interpreter.py
index 9eccb79605..52276b5bb0 100644
--- a/dspy/primitives/code_interpreter.py
+++ b/dspy/primitives/code_interpreter.py
@@ -80,6 +80,25 @@ class CodeInterpreter(Protocol):
     Pooling:
         For interpreter pooling, call start() to pre-warm instances, then
         distribute execute() calls across the pool.
+
+    Callbacks:
+        The Protocol cannot enforce observability, since callbacks are a per-method
+        behavior rather than part of the interface. Implementations that want to
+        participate in DSPy's callback system (`dspy.BaseCallback`) should apply the
+        `@dspy.utils.callback.with_callbacks` decorator to their methods. Callback routing
+        dispatches by method name (mirroring the Adapter's `format`/`parse` routing), so the
+        decorated methods must use these exact names to be recognized:
+
+        - `execute`  -> `on_interpreter_execute_*`
+        - `start`    -> `on_interpreter_startup_*`
+        - `shutdown` -> `on_interpreter_shutdown_*`
+        - `invoke_tool(self, tool_name, kwargs)` -> `on_interpreter_tool_call_*`, a public
+          seam wrapping a single sandbox->host tool invocation. Route your tool dispatch
+          through a method of this name (decorating a differently named method will not emit
+          tool-call events).
+
+        `dspy.PythonInterpreter` does this out of the box (it additionally emits startup from
+        its internal process-spawn seam so lazy start via `execute()` is covered).
     """
 
     @property
diff --git a/dspy/primitives/python_interpreter.py b/dspy/primitives/python_interpreter.py
index b700dfffc3..1d95cad836 100644
--- a/dspy/primitives/python_interpreter.py
+++ b/dspy/primitives/python_interpreter.py
@@ -19,6 +19,7 @@
 from typing import Any, Callable
 
 from dspy.primitives.code_interpreter import SIMPLE_TYPES, CodeInterpreterError, FinalOutput
+from dspy.utils.callback import with_callbacks
 
 __all__ = ["PythonInterpreter", "FinalOutput", "CodeInterpreterError"]
 
@@ -129,6 +130,7 @@ def __init__(
         sync_files: bool = True,
         tools: dict[str, Callable[..., str]] | None = None,
         output_fields: list[dict] | None = None,
+        callbacks: list | None = None,
     ) -> None:
         """
         Args:
@@ -144,6 +146,9 @@ def __init__(
                    Tools are callable directly from sandbox code by name.
             output_fields: List of output field definitions for typed SUBMIT signature.
                    Each dict should have 'name' and optionally 'type' keys.
+            callbacks: Instance-level `dspy.BaseCallback` handlers, combined with any globally
+                   configured callbacks. Interpreter execute/startup/shutdown and sandbox->host
+                   tool dispatch are surfaced through the `on_interpreter_*` handlers.
         """
         if isinstance(deno_command, dict):
             raise TypeError("deno_command must be a list of strings, not a dict")
@@ -155,6 +160,7 @@ def __init__(
         self.sync_files = sync_files
         self.tools = dict(tools) if tools else {}
         self.output_fields = output_fields
+        self.callbacks = callbacks or []
         self._tools_registered = False
         # TODO later on add enable_run (--allow-run) by proxying subprocess.run through Deno.run() to fix 'emscripten does not support processes' error
 
@@ -313,6 +319,27 @@ def _register_tools(self) -> None:
         self._send_request("register", params, "registering tools/outputs")
         self._tools_registered = True
 
+    @with_callbacks
+    def invoke_tool(self, tool_name: str, kwargs: dict) -> Any:
+        """Invoke a single host-side tool by name.
+
+        This is the callback seam for sandbox->host tool dispatch, routed to the
+        `on_interpreter_tool_call_*` handlers. It is deliberately a public method so custom
+        `CodeInterpreter` implementations can decorate an identically named method to
+        participate in the callback system (see `CodeInterpreter`'s docstring).
+
+        It is extracted from `_handle_tool_call` so the callback observes the real exception
+        while it is still in flight: `_handle_tool_call` catches every exception to convert it
+        into a JSON-RPC error response, so decorating it directly would always report
+        `exception=None`.
+        """
+        if tool_name not in self.tools:
+            raise CodeInterpreterError(f"Unknown tool: {tool_name}")
+        result = self.tools[tool_name](**kwargs)
+        if asyncio.iscoroutine(result):
+            result = _await_in_sync(result)
+        return result
+
     def _handle_tool_call(self, request: dict) -> None:
         """Handle a tool call request from the sandbox."""
         request_id = request["id"]
@@ -321,11 +348,7 @@ def _handle_tool_call(self, request: dict) -> None:
         kwargs = params.get("kwargs", {})
 
         try:
-            if tool_name not in self.tools:
-                raise CodeInterpreterError(f"Unknown tool: {tool_name}")
-            result = self.tools[tool_name](**kwargs)
-            if asyncio.iscoroutine(result):
-                result = _await_in_sync(result)
+            result = self.invoke_tool(tool_name, kwargs)
             is_json = isinstance(result, (list, dict))
             response = _jsonrpc_result(
                 {"value": json.dumps(result) if is_json else (str(result) if result is not None else ""), "type": "json" if is_json else "string"},
@@ -341,30 +364,42 @@ def _handle_tool_call(self, request: dict) -> None:
 
     def _ensure_deno_process(self) -> None:
         if self.deno_process is None or self.deno_process.poll() is not None:
-            # Process identity changed (or process missing), so replay setup.
-            self._tools_registered = False
-            self._mounted_files = False
-            try:
-                self.deno_process = subprocess.Popen(
-                    self.deno_command,
-                    stdin=subprocess.PIPE,
-                    stdout=subprocess.PIPE,
-                    stderr=subprocess.PIPE,
-                    text=True,
-                    encoding="UTF-8",
-                    env=os.environ.copy()
-                )
-            except FileNotFoundError as e:
-                install_instructions = (
-                    "Deno executable not found. Please install Deno to proceed.\n"
-                    "Installation instructions:\n"
-                    "> curl -fsSL https://deno.land/install.sh | sh\n"
-                    "*or*, on macOS with Homebrew:\n"
-                    "> brew install deno\n"
-                    "For additional configurations: https://docs.deno.com/runtime/getting_started/installation/"
-                )
-                raise CodeInterpreterError(install_instructions) from e
-            self._health_check()
+            self._spawn_process()
+
+    @with_callbacks
+    def _spawn_process(self) -> None:
+        """Spawn the Deno subprocess and health-check it.
+
+        This is the real startup seam: it runs both when start() is called explicitly and,
+        more commonly, lazily on the first execute() (the default RLM path). Decorating it
+        (rather than the public start()) means on_interpreter_startup_* fires exactly once
+        per actual process spawn on either path, instead of on every execute() or only on an
+        explicit start(). start() is intentionally left undecorated to avoid double-emission.
+        """
+        # Process identity changed (or process missing), so replay setup.
+        self._tools_registered = False
+        self._mounted_files = False
+        try:
+            self.deno_process = subprocess.Popen(
+                self.deno_command,
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                encoding="UTF-8",
+                env=os.environ.copy()
+            )
+        except FileNotFoundError as e:
+            install_instructions = (
+                "Deno executable not found. Please install Deno to proceed.\n"
+                "Installation instructions:\n"
+                "> curl -fsSL https://deno.land/install.sh | sh\n"
+                "*or*, on macOS with Homebrew:\n"
+                "> brew install deno\n"
+                "For additional configurations: https://docs.deno.com/runtime/getting_started/installation/"
+            )
+            raise CodeInterpreterError(install_instructions) from e
+        self._health_check()
 
     _MAX_SKIP_LINES = 100
 
@@ -507,6 +542,7 @@ def _inject_large_var(self, name: str, value: str) -> None:
         """Inject a large variable via the virtual filesystem."""
         self._send_request("inject_var", {"name": name, "value": value}, f"injecting variable '{name}'")
 
+    @with_callbacks
     def execute(
         self,
         code: str,
@@ -596,6 +632,11 @@ def start(self) -> None:
         on first execute().
 
         Idempotent: safe to call multiple times.
+
+        Note: this method is not decorated with @with_callbacks. The startup callbacks
+        (on_interpreter_startup_*) fire from the underlying process-spawn seam (_spawn_process),
+        so they are emitted once per actual spawn whether startup is explicit (start()) or lazy
+        (first execute()), and are not emitted when the process is already running.
         """
         self._ensure_deno_process()
 
@@ -612,6 +653,7 @@ def __call__(
     ) -> Any:
         return self.execute(code, variables)
 
+    @with_callbacks
     def shutdown(self) -> None:
         if self.deno_process and self.deno_process.poll() is None:
             self.deno_process.stdin.write(_jsonrpc_notification("shutdown") + "\n")
diff --git a/dspy/utils/callback.py b/dspy/utils/callback.py
index cd9094f7e0..d6b183d2fa 100644
--- a/dspy/utils/callback.py
+++ b/dspy/utils/callback.py
@@ -254,6 +254,147 @@ def on_evaluate_end(
         """
         pass
 
+    def on_interpreter_execute_start(
+        self,
+        call_id: str,
+        instance: Any,
+        inputs: dict[str, Any],
+    ):
+        """A handler triggered when execute() method of a CodeInterpreter is called.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            instance: The CodeInterpreter instance (e.g. dspy.PythonInterpreter).
+            inputs: The inputs to the interpreter's execute() method, stored as key-value pairs.
+                Includes ``code`` (the Python source to run) and ``variables`` (the namespace
+                injected before execution).
+
+                Note: ``inputs["variables"]`` can be very large. For example, dspy.RLM passes its
+                input data through it on every iteration. The framework does not truncate or copy
+                payloads, mirroring the ``on_lm_start`` precedent of passing full messages, so
+                handlers are responsible for any truncation or redaction they need.
+        """
+        pass
+
+    def on_interpreter_execute_end(
+        self,
+        call_id: str,
+        outputs: Any | None,
+        exception: Exception | None = None,
+    ):
+        """A handler triggered after execute() method of a CodeInterpreter is executed.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            outputs: The outputs of the interpreter's execute() method. This may be a
+                ``FinalOutput`` (when SUBMIT() was called), a string, a list, or None. If the
+                method is interrupted by an exception, this will be None.
+            exception: If an exception is raised during the execution, it will be stored here.
+                Interpreter execution errors surface as ``CodeInterpreterError`` and invalid
+                Python source surfaces as ``SyntaxError``.
+        """
+        pass
+
+    def on_interpreter_tool_call_start(
+        self,
+        call_id: str,
+        instance: Any,
+        inputs: dict[str, Any],
+    ):
+        """A handler triggered when the sandbox invokes a host-side tool.
+
+        This fires for every entry in ``interpreter.tools``, including plain closures such as
+        dspy.RLM's ``llm_query``/``llm_query_batched``, not just dspy.Tool objects.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            instance: The CodeInterpreter instance that dispatched the tool call.
+            inputs: The inputs to the tool dispatch, stored as key-value pairs. Includes
+                ``tool_name`` (the name of the invoked tool) and ``kwargs`` (the keyword
+                arguments passed from the sandbox).
+        """
+        pass
+
+    def on_interpreter_tool_call_end(
+        self,
+        call_id: str,
+        outputs: Any | None,
+        exception: Exception | None = None,
+    ):
+        """A handler triggered after the sandbox invokes a host-side tool.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            outputs: The value returned by the tool. If the tool raises, this will be None.
+            exception: If the tool raises during dispatch, the exception is stored here. The
+                interpreter still converts it into a sandbox-side error afterwards, so this
+                handler observes the real exception without changing sandbox behavior.
+        """
+        pass
+
+    def on_interpreter_startup_start(
+        self,
+        call_id: str,
+        instance: Any,
+        inputs: dict[str, Any],
+    ):
+        """A handler triggered when start() method of a CodeInterpreter is called.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            instance: The CodeInterpreter instance being started.
+            inputs: The inputs to the interpreter's start() method, stored as key-value pairs.
+        """
+        pass
+
+    def on_interpreter_startup_end(
+        self,
+        call_id: str,
+        outputs: Any | None,
+        exception: Exception | None = None,
+    ):
+        """A handler triggered after start() method of a CodeInterpreter is executed.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            outputs: The outputs of the interpreter's start() method. If the method is interrupted
+                by an exception, this will be None.
+            exception: If an exception is raised during startup (e.g. the sandbox process fails to
+                spawn), it will be stored here.
+        """
+        pass
+
+    def on_interpreter_shutdown_start(
+        self,
+        call_id: str,
+        instance: Any,
+        inputs: dict[str, Any],
+    ):
+        """A handler triggered when shutdown() method of a CodeInterpreter is called.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            instance: The CodeInterpreter instance being shut down.
+            inputs: The inputs to the interpreter's shutdown() method, stored as key-value pairs.
+        """
+        pass
+
+    def on_interpreter_shutdown_end(
+        self,
+        call_id: str,
+        outputs: Any | None,
+        exception: Exception | None = None,
+    ):
+        """A handler triggered after shutdown() method of a CodeInterpreter is executed.
+
+        Args:
+            call_id: A unique identifier for the call. Can be used to connect start/end handlers.
+            outputs: The outputs of the interpreter's shutdown() method. If the method is
+                interrupted by an exception, this will be None.
+            exception: If an exception is raised during the execution, it will be stored here.
+        """
+        pass
+
 
 def with_callbacks(fn):
     """Decorator to add callback functionality to instance methods."""
@@ -348,6 +489,32 @@ def sync_wrapper(instance, *args, **kwargs):
         return sync_wrapper
 
 
+# Maps the decorated CodeInterpreter method name to the corresponding callback handler.
+#
+# The tool-call hook is routed via the extracted `invoke_tool` seam (see PythonInterpreter)
+# rather than the exception-swallowing `_handle_tool_call`, so the end handler observes real
+# exceptions. `invoke_tool` is a public method name so custom implementations can adopt it.
+#
+# Startup routes both `start` (the public method some implementations decorate directly, e.g.
+# MockInterpreter) and `_spawn_process` (PythonInterpreter's real process-spawn seam, so the
+# common lazy-start path via execute() emits startup exactly once per actual spawn).
+_INTERPRETER_START_HANDLERS = {
+    "execute": "on_interpreter_execute_start",
+    "start": "on_interpreter_startup_start",
+    "_spawn_process": "on_interpreter_startup_start",
+    "shutdown": "on_interpreter_shutdown_start",
+    "invoke_tool": "on_interpreter_tool_call_start",
+}
+
+_INTERPRETER_END_HANDLERS = {
+    "execute": "on_interpreter_execute_end",
+    "start": "on_interpreter_startup_end",
+    "_spawn_process": "on_interpreter_startup_end",
+    "shutdown": "on_interpreter_shutdown_end",
+    "invoke_tool": "on_interpreter_tool_call_end",
+}
+
+
 def _get_on_start_handler(callback: BaseCallback, instance: Any, fn: Callable) -> Callable:
     """Selects the appropriate on_start handler of the callback based on the instance and function name."""
     if isinstance(instance, dspy.BaseLM):
@@ -366,6 +533,16 @@ def _get_on_start_handler(callback: BaseCallback, instance: Any, fn: Callable) -
     if isinstance(instance, dspy.Tool):
         return callback.on_tool_start
 
+    # CodeInterpreter is a runtime_checkable Protocol, so this isinstance check is structural
+    # (it only checks for the presence of start/execute/shutdown/tools). None of the branches
+    # above are structural interpreters, so ordering relative to them is safe; this branch must
+    # stay before the module fallback so interpreter events do not masquerade as module events.
+    if isinstance(instance, dspy.CodeInterpreter):
+        handler_name = _INTERPRETER_START_HANDLERS.get(fn.__name__)
+        if handler_name is None:
+            raise ValueError(f"Unsupported interpreter method for using callback: {fn.__name__}.")
+        return getattr(callback, handler_name)
+
     # We treat everything else as a module.
     return callback.on_module_start
 
@@ -388,5 +565,13 @@ def _get_on_end_handler(callback: BaseCallback, instance: Any, fn: Callable) ->
     if isinstance(instance, dspy.Tool):
         return callback.on_tool_end
 
+    # See the note in _get_on_start_handler: this structural Protocol check must stay before the
+    # module fallback so interpreter events do not masquerade as module events.
+    if isinstance(instance, dspy.CodeInterpreter):
+        handler_name = _INTERPRETER_END_HANDLERS.get(fn.__name__)
+        if handler_name is None:
+            raise ValueError(f"Unsupported interpreter method for using callback: {fn.__name__}.")
+        return getattr(callback, handler_name)
+
     # We treat everything else as a module.
     return callback.on_module_end
diff --git a/tests/callback/test_interpreter_callback.py b/tests/callback/test_interpreter_callback.py
new file mode 100644
index 0000000000..a7200d6532
--- /dev/null
+++ b/tests/callback/test_interpreter_callback.py
@@ -0,0 +1,588 @@
+"""Tests for interpreter-level callbacks (`on_interpreter_*`).
+
+Test organization mirrors tests/primitives/test_python_interpreter.py:
+- Unit tests (no Deno required): MockInterpreter, PythonInterpreter tool-dispatch seam,
+  dispatch-routing integrity.
+- Integration tests (@pytest.mark.deno): real PythonInterpreter execute/tool/lifecycle events
+  and a full dspy.RLM run.
+"""
+
+import json
+
+import pytest
+
+import dspy
+from dspy.predict.rlm import RLM
+from dspy.primitives import python_interpreter
+from dspy.primitives.code_interpreter import CodeInterpreterError, FinalOutput
+from dspy.primitives.prediction import Prediction
+from dspy.primitives.python_interpreter import PythonInterpreter
+from dspy.utils.callback import ACTIVE_CALL_ID, BaseCallback
+from dspy.utils.dummies import DummyLM
+from tests.mock_interpreter import MockInterpreter
+
+
+@pytest.fixture(autouse=True)
+def reset_settings():
+    original_settings = dspy.settings.copy()
+    yield
+    dspy.configure(**original_settings)
+
+
+class RecordingCallback(BaseCallback):
+    """Records every handler invocation, capturing the active (parent) call id at call time."""
+
+    def __init__(self):
+        self.calls = []
+
+    def _record(self, handler, call_id, **kwargs):
+        self.calls.append({
+            "handler": handler,
+            "call_id": call_id,
+            "parent_call_id": ACTIVE_CALL_ID.get(),
+            **kwargs,
+        })
+
+    # Module / LM / Tool handlers (used to assert dispatch integrity).
+    def on_module_start(self, call_id, instance, inputs):
+        self._record("on_module_start", call_id, instance=instance, inputs=inputs)
+
+    def on_module_end(self, call_id, outputs, exception):
+        self._record("on_module_end", call_id, outputs=outputs, exception=exception)
+
+    def on_lm_start(self, call_id, instance, inputs):
+        self._record("on_lm_start", call_id, instance=instance, inputs=inputs)
+
+    def on_lm_end(self, call_id, outputs, exception):
+        self._record("on_lm_end", call_id, outputs=outputs, exception=exception)
+
+    def on_tool_start(self, call_id, instance, inputs):
+        self._record("on_tool_start", call_id, instance=instance, inputs=inputs)
+
+    def on_tool_end(self, call_id, outputs, exception):
+        self._record("on_tool_end", call_id, outputs=outputs, exception=exception)
+
+    # Interpreter handlers.
+    def on_interpreter_execute_start(self, call_id, instance, inputs):
+        self._record("on_interpreter_execute_start", call_id, instance=instance, inputs=inputs)
+
+    def on_interpreter_execute_end(self, call_id, outputs, exception):
+        self._record("on_interpreter_execute_end", call_id, outputs=outputs, exception=exception)
+
+    def on_interpreter_tool_call_start(self, call_id, instance, inputs):
+        self._record("on_interpreter_tool_call_start", call_id, instance=instance, inputs=inputs)
+
+    def on_interpreter_tool_call_end(self, call_id, outputs, exception):
+        self._record("on_interpreter_tool_call_end", call_id, outputs=outputs, exception=exception)
+
+    def on_interpreter_startup_start(self, call_id, instance, inputs):
+        self._record("on_interpreter_startup_start", call_id, instance=instance, inputs=inputs)
+
+    def on_interpreter_startup_end(self, call_id, outputs, exception):
+        self._record("on_interpreter_startup_end", call_id, outputs=outputs, exception=exception)
+
+    def on_interpreter_shutdown_start(self, call_id, instance, inputs):
+        self._record("on_interpreter_shutdown_start", call_id, instance=instance, inputs=inputs)
+
+    def on_interpreter_shutdown_end(self, call_id, outputs, exception):
+        self._record("on_interpreter_shutdown_end", call_id, outputs=outputs, exception=exception)
+
+    def handlers(self):
+        return [c["handler"] for c in self.calls]
+
+    def by_handler(self, handler):
+        return [c for c in self.calls if c["handler"] == handler]
+
+
+class _FakeStdin:
+    """Captures lines written by _handle_tool_call so tests can assert JSON-RPC responses."""
+
+    def __init__(self):
+        self.written = []
+
+    def write(self, data):
+        self.written.append(data)
+
+    def flush(self):
+        pass
+
+
+class _FakeProcess:
+    def __init__(self):
+        self.stdin = _FakeStdin()
+
+
+# ============================================================================
+# Unit Tests: execute() events via MockInterpreter (no Deno required)
+# ============================================================================
+
+
+def test_execute_fires_start_and_end_with_code_and_output():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    interp = MockInterpreter(responses=["hello world\n"])
+    result = interp.execute("print('hello world')", variables={"x": 1})
+
+    assert result == "hello world\n"
+    assert callback.handlers() == ["on_interpreter_execute_start", "on_interpreter_execute_end"]
+
+    start = callback.by_handler("on_interpreter_execute_start")[0]
+    assert start["inputs"]["code"] == "print('hello world')"
+    assert start["inputs"]["variables"] == {"x": 1}
+
+    end = callback.by_handler("on_interpreter_execute_end")[0]
+    assert end["outputs"] == "hello world\n"
+    assert end["exception"] is None
+    # start/end share the same call_id.
+    assert start["call_id"] == end["call_id"]
+
+
+def test_execute_submit_path_reports_final_output():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    interp = MockInterpreter(responses=[FinalOutput({"answer": "42"})])
+    result = interp.execute("SUBMIT('42')")
+
+    assert isinstance(result, FinalOutput)
+    end = callback.by_handler("on_interpreter_execute_end")[0]
+    assert isinstance(end["outputs"], FinalOutput)
+    assert end["outputs"].output == {"answer": "42"}
+    assert end["exception"] is None
+
+
+def test_execute_error_path_surfaces_exception_and_propagates():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    interp = MockInterpreter(responses=[CodeInterpreterError("NameError: name 'x' is not defined")])
+
+    with pytest.raises(CodeInterpreterError, match="NameError"):
+        interp.execute("print(x)")
+
+    end = callback.by_handler("on_interpreter_execute_end")[0]
+    assert isinstance(end["exception"], CodeInterpreterError)
+    assert end["outputs"] is None
+
+
+# ============================================================================
+# Unit Tests: tool-dispatch seam via PythonInterpreter (no Deno process needed)
+# ============================================================================
+
+
+def test_invoke_tool_fires_events_with_tool_name_and_kwargs():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    def my_tool(a: str = "", b: str = "") -> str:
+        return f"{a}:{b}"
+
+    interp = PythonInterpreter(tools={"my_tool": my_tool})
+    # invoke_tool is the decorated seam; it does not require a running Deno process.
+    result = interp.invoke_tool("my_tool", {"a": "x", "b": "y"})
+
+    assert result == "x:y"
+    assert callback.handlers() == ["on_interpreter_tool_call_start", "on_interpreter_tool_call_end"]
+
+    start = callback.by_handler("on_interpreter_tool_call_start")[0]
+    assert start["inputs"] == {"tool_name": "my_tool", "kwargs": {"a": "x", "b": "y"}}
+
+    end = callback.by_handler("on_interpreter_tool_call_end")[0]
+    assert end["outputs"] == "x:y"
+    assert end["exception"] is None
+
+
+def test_plain_closure_tool_fires_events():
+    """A plain (non-dspy.Tool) callable, like RLM's llm_query, still fires tool-call events."""
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    def llm_query(prompt: str = "") -> str:
+        return f"answer to {prompt}"
+
+    interp = PythonInterpreter(tools={"llm_query": llm_query})
+    result = interp.invoke_tool("llm_query", {"prompt": "hi"})
+
+    assert result == "answer to hi"
+    start = callback.by_handler("on_interpreter_tool_call_start")[0]
+    assert start["inputs"]["tool_name"] == "llm_query"
+    assert callback.by_handler("on_interpreter_tool_call_end")[0]["outputs"] == "answer to hi"
+
+
+def test_raising_tool_delivers_exception_and_sandbox_still_gets_jsonrpc_error():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    def failing_tool() -> str:
+        raise RuntimeError("Tool failed!")
+
+    interp = PythonInterpreter(tools={"failing_tool": failing_tool})
+    interp.deno_process = _FakeProcess()
+
+    # _handle_tool_call preserves its JSON-RPC error conversion: the exception reaches the end
+    # handler, and is then caught and written to the sandbox exactly as before.
+    interp._handle_tool_call({"id": 7, "params": {"name": "failing_tool", "kwargs": {}}})
+
+    # The end handler saw the real exception...
+    end = callback.by_handler("on_interpreter_tool_call_end")[0]
+    assert isinstance(end["exception"], RuntimeError)
+    assert str(end["exception"]) == "Tool failed!"
+    assert end["outputs"] is None
+
+    # ...and the sandbox still received a JSON-RPC error response for request id 7.
+    assert len(interp.deno_process.stdin.written) == 1
+    written = json.loads(interp.deno_process.stdin.written[0])
+    assert written["id"] == 7
+    assert "error" in written
+    assert written["error"]["data"]["type"] == "RuntimeError"
+    assert written["error"]["message"] == "Tool failed!"
+
+
+def test_unknown_tool_still_converted_to_jsonrpc_error():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    interp = PythonInterpreter(tools={})
+    interp.deno_process = _FakeProcess()
+
+    interp._handle_tool_call({"id": 3, "params": {"name": "nope", "kwargs": {}}})
+
+    end = callback.by_handler("on_interpreter_tool_call_end")[0]
+    assert isinstance(end["exception"], CodeInterpreterError)
+
+    written = json.loads(interp.deno_process.stdin.written[0])
+    assert written["id"] == 3
+    assert written["error"]["data"]["type"] == "CodeInterpreterError"
+
+
+# ============================================================================
+# Unit Tests: nesting (tool call nests under the enclosing execute call)
+# ============================================================================
+
+
+def test_tool_call_nests_under_execute_call_id():
+    """A decorated tool call invoked synchronously inside a decorated execute() nests under it.
+
+    This exercises the same ACTIVE_CALL_ID mechanism PythonInterpreter relies on: invoke_tool
+    runs on the same thread inside execute(), so its parent is the execute call_id.
+    """
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    def my_tool(x: str = "") -> str:
+        return x
+
+    interp = PythonInterpreter(tools={"my_tool": my_tool})
+
+    def execute_fn(code, variables):
+        # Simulate the sandbox calling back into a host tool mid-execution.
+        return interp.invoke_tool("my_tool", {"x": "nested"})
+
+    mock = MockInterpreter(execute_fn=execute_fn)
+    mock.execute("my_tool(x='nested')")
+
+    execute_start = callback.by_handler("on_interpreter_execute_start")[0]
+    tool_start = callback.by_handler("on_interpreter_tool_call_start")[0]
+
+    # The tool call's parent (active call id at handler time) is the execute call id.
+    assert tool_start["parent_call_id"] == execute_start["call_id"]
+    # The execute call itself has no interpreter parent here.
+    assert execute_start["parent_call_id"] is None
+
+
+# ============================================================================
+# Unit Tests: lifecycle + idempotent shutdown
+# ============================================================================
+
+
+def test_startup_and_shutdown_events_fire_and_shutdown_is_idempotent():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    # PythonInterpreter.shutdown() with no live process is a no-op; decoration must not change that.
+    interp = PythonInterpreter()
+    interp.shutdown()
+    interp.shutdown()  # idempotent: safe to call twice
+
+    handlers = callback.handlers()
+    assert handlers == [
+        "on_interpreter_shutdown_start",
+        "on_interpreter_shutdown_end",
+        "on_interpreter_shutdown_start",
+        "on_interpreter_shutdown_end",
+    ]
+    for end in callback.by_handler("on_interpreter_shutdown_end"):
+        assert end["exception"] is None
+
+
+def _stub_spawn(monkeypatch):
+    """Stub the Deno subprocess spawn + health check so startup can be tested without Deno."""
+
+    class _FakeAliveProcess:
+        def poll(self):
+            return None  # report the process as alive
+
+    spawns = {"count": 0}
+
+    def fake_popen(*args, **kwargs):
+        spawns["count"] += 1
+        return _FakeAliveProcess()
+
+    monkeypatch.setattr(python_interpreter.subprocess, "Popen", fake_popen)
+    monkeypatch.setattr(PythonInterpreter, "_health_check", lambda self: None)
+    return spawns
+
+
+def test_lazy_spawn_emits_startup_events_once(monkeypatch):
+    """The lazy Deno spawn inside execute()/_ensure_deno_process (the default RLM path) emits
+    startup events, exactly once per actual spawn."""
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+    spawns = _stub_spawn(monkeypatch)
+
+    interp = PythonInterpreter()
+    # _ensure_deno_process is what execute() calls before talking to the sandbox (lazy start).
+    interp._ensure_deno_process()
+    assert spawns["count"] == 1
+    assert callback.handlers() == ["on_interpreter_startup_start", "on_interpreter_startup_end"]
+    assert callback.by_handler("on_interpreter_startup_end")[0]["exception"] is None
+
+    # Process already running: no re-spawn and no new startup events.
+    interp._ensure_deno_process()
+    assert spawns["count"] == 1
+    assert callback.handlers() == ["on_interpreter_startup_start", "on_interpreter_startup_end"]
+
+
+def test_explicit_start_emits_startup_events_once(monkeypatch):
+    """Explicit start() also emits startup events (via the same spawn seam), once per spawn."""
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+    spawns = _stub_spawn(monkeypatch)
+
+    interp = PythonInterpreter()
+    interp.start()
+    assert spawns["count"] == 1
+    assert callback.handlers() == ["on_interpreter_startup_start", "on_interpreter_startup_end"]
+
+    # start() is idempotent: no duplicate startup events when already running.
+    interp.start()
+    assert spawns["count"] == 1
+    assert callback.handlers() == ["on_interpreter_startup_start", "on_interpreter_startup_end"]
+
+
+def test_mock_interpreter_lifecycle_events():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    interp = MockInterpreter(responses=["ok"])
+    interp.start()
+    interp.execute("print(1)")
+    interp.shutdown()
+
+    assert callback.handlers() == [
+        "on_interpreter_startup_start",
+        "on_interpreter_startup_end",
+        "on_interpreter_execute_start",
+        "on_interpreter_execute_end",
+        "on_interpreter_shutdown_start",
+        "on_interpreter_shutdown_end",
+    ]
+
+
+# ============================================================================
+# Unit Tests: dispatch integrity + zero-callback path
+# ============================================================================
+
+
+def test_dispatch_integrity_lm_tool_module_not_misrouted():
+    """Interpreter events must not fire module handlers, and LM/Tool/Module keep their own."""
+    callback = RecordingCallback()
+    dspy.configure(
+        lm=DummyLM({"How are you?": {"answer": "test output", "reasoning": "No more responses"}}),
+        callbacks=[callback],
+    )
+
+    # Module + LM + Adapter events.
+    cot = dspy.ChainOfThought("question -> answer")
+    cot(question="How are you?")
+
+    # Tool events.
+    def tool_1(query: str) -> str:
+        return "result 1"
+
+    dspy.Tool(tool_1)(query="x")
+
+    # Interpreter events.
+    interp = MockInterpreter(responses=["ok"])
+    interp.execute("print(1)")
+
+    handlers = callback.handlers()
+    # Module/LM/Tool events are all present.
+    assert "on_module_start" in handlers
+    assert "on_lm_start" in handlers
+    assert "on_tool_start" in handlers
+    # Interpreter events are present...
+    assert "on_interpreter_execute_start" in handlers
+
+    # ...and no module-start event was emitted by the interpreter (they all come from the
+    # ChainOfThought module, never masquerading through the new dispatch branch).
+    for call in callback.by_handler("on_module_start"):
+        assert not isinstance(call["instance"], MockInterpreter)
+
+
+def test_interpreter_only_run_emits_no_module_events():
+    """Running only an interpreter must not trip any module/lm/tool handlers."""
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    interp = MockInterpreter(responses=["ok"])
+    interp.start()
+    interp.execute("print(1)")
+    interp.shutdown()
+
+    handlers = set(callback.handlers())
+    assert not (handlers & {
+        "on_module_start", "on_module_end",
+        "on_lm_start", "on_lm_end",
+        "on_tool_start", "on_tool_end",
+    })
+
+
+def test_zero_callback_path_unchanged():
+    # No callbacks registered: behavior and results are unchanged, nothing recorded.
+    callback = RecordingCallback()  # not registered anywhere
+
+    interp = MockInterpreter(responses=["out", FinalOutput({"answer": "42"})])
+    interp.start()
+    assert interp.execute("print(1)", variables={"a": 1}) == "out"
+    result = interp.execute("SUBMIT('42')")
+    assert isinstance(result, FinalOutput)
+    interp.shutdown()
+
+    assert callback.calls == []
+    # Recording of call history (the mock's own bookkeeping) is unaffected.
+    assert interp.call_history == [("print(1)", {"a": 1}), ("SUBMIT('42')", {})]
+
+
+# ============================================================================
+# Integration: dspy.RLM run emits interpreter execute events (no Deno required)
+# ============================================================================
+
+
+def _make_mock_predictor(responses):
+    class MockPredictor:
+        def __init__(self):
+            self.idx = 0
+
+        def __call__(self, **kwargs):
+            result = responses[self.idx % len(responses)]
+            self.idx += 1
+            return Prediction(**result)
+
+    return MockPredictor()
+
+
+def test_rlm_run_emits_one_execute_event_per_iteration():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    mock = MockInterpreter(responses=[
+        "explored\n",
+        "still exploring\n",
+        FinalOutput({"answer": "42"}),
+    ])
+    rlm = RLM("query -> answer", max_iters=5, interpreter=mock)
+    rlm.generate_action = _make_mock_predictor([
+        {"reasoning": "Explore", "code": "print('explore 1')"},
+        {"reasoning": "Explore", "code": "print('explore 2')"},
+        {"reasoning": "Submit", "code": "SUBMIT('42')"},
+    ])
+
+    result = rlm.forward(query="test")
+    assert result.answer == "42"
+
+    execute_starts = callback.by_handler("on_interpreter_execute_start")
+    execute_ends = callback.by_handler("on_interpreter_execute_end")
+    # One execute event per RLM iteration (3 iterations here).
+    assert len(execute_starts) == 3
+    assert len(execute_ends) == 3
+
+    # Interpreter events are interleaved with the module events emitted by RLM.
+    handlers = callback.handlers()
+    assert "on_interpreter_execute_start" in handlers
+    # The final iteration's output is the FinalOutput.
+    assert isinstance(execute_ends[-1]["outputs"], FinalOutput)
+
+    # RLM injects llm_query as a plain closure into the interpreter; those are non-Tool callables,
+    # so no on_tool_* events fire for them (the mock does not dispatch tools).
+    assert "on_tool_start" not in handlers
+
+
+# ============================================================================
+# Deno integration tests (real PythonInterpreter)
+# ============================================================================
+
+
+@pytest.mark.deno
+def test_deno_execute_events_and_nesting():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    calls = {"count": 0}
+
+    def host_tool(value: str = "") -> str:
+        calls["count"] += 1
+        return f"host:{value}"
+
+    with PythonInterpreter(tools={"host_tool": host_tool}) as interp:
+        # A successful execute with code present in inputs and output in outputs.
+        result = interp.execute("print('hi')")
+        assert result == "hi\n"
+
+        exec_start = callback.by_handler("on_interpreter_execute_start")[0]
+        assert exec_start["inputs"]["code"] == "print('hi')"
+        exec_end = callback.by_handler("on_interpreter_execute_end")[0]
+        assert exec_end["outputs"] == "hi\n"
+
+        # The Deno process was spawned lazily inside this first execute(), so the startup
+        # events fired and nest under the execute call_id.
+        startup_start = callback.by_handler("on_interpreter_startup_start")
+        assert len(startup_start) == 1
+        assert startup_start[0]["parent_call_id"] == exec_start["call_id"]
+
+        # A tool call dispatched from inside execute nests under that execute call_id.
+        callback.calls.clear()
+        out = interp.execute("print(host_tool(value='x'))")
+        assert "host:x" in out
+
+        tool_starts = callback.by_handler("on_interpreter_tool_call_start")
+        assert len(tool_starts) == 1
+        assert tool_starts[0]["inputs"]["tool_name"] == "host_tool"
+        assert tool_starts[0]["inputs"]["kwargs"] == {"value": "x"}
+
+        this_execute = callback.by_handler("on_interpreter_execute_start")[0]
+        assert tool_starts[0]["parent_call_id"] == this_execute["call_id"]
+
+        # SUBMIT path yields a FinalOutput in the end handler.
+        callback.calls.clear()
+        submit_result = interp.execute("SUBMIT('done')")
+        assert isinstance(submit_result, FinalOutput)
+        assert isinstance(callback.by_handler("on_interpreter_execute_end")[0]["outputs"], FinalOutput)
+
+    # Startup and shutdown events fired for the real interpreter across the context.
+    assert callback.by_handler("on_interpreter_shutdown_start")
+
+
+@pytest.mark.deno
+def test_deno_execute_error_surfaces_exception():
+    callback = RecordingCallback()
+    dspy.configure(callbacks=[callback])
+
+    with PythonInterpreter() as interp:
+        with pytest.raises(CodeInterpreterError):
+            interp.execute("1/0")
+
+    end = callback.by_handler("on_interpreter_execute_end")[-1]
+    assert isinstance(end["exception"], CodeInterpreterError)
+    assert end["outputs"] is None
diff --git a/tests/mock_interpreter.py b/tests/mock_interpreter.py
index 255f0282f2..e7cbced3ed 100644
--- a/tests/mock_interpreter.py
+++ b/tests/mock_interpreter.py
@@ -11,6 +11,7 @@
 from typing import Any, Callable
 
 from dspy.primitives.code_interpreter import CodeInterpreterError, FinalOutput
+from dspy.utils.callback import with_callbacks
 
 __all__ = ["MockInterpreter"]
 
@@ -45,6 +46,7 @@ def __init__(
         responses: list[str | FinalOutput | Exception] | None = None,
         execute_fn: Callable[[str, dict[str, Any]], Any] | None = None,
         tools: dict[str, Callable[..., str]] | None = None,
+        callbacks: list | None = None,
     ):
         """Initialize the mock interpreter.
 
@@ -56,17 +58,23 @@ def __init__(
                        returns the result. Takes precedence over responses.
             tools: Dictionary mapping tool names to callable functions.
                    MockInterpreter doesn't use tools, but stores them for protocol compliance.
+            callbacks: Instance-level `dspy.BaseCallback` handlers, combined with globally
+                   configured callbacks. Mirrors PythonInterpreter so the mock participates
+                   in the `on_interpreter_*` callback system.
         """
         self.responses = list(responses) if responses else []
         self.execute_fn = execute_fn
         self.tools = tools or {}
+        self.callbacks = callbacks or []
         self.call_count = 0
         self.call_history: list[tuple[str, dict[str, Any]]] = []
         self._shutdown = False
 
+    @with_callbacks
     def start(self) -> None:
         pass
 
+    @with_callbacks
     def execute(
         self,
         code: str,
@@ -107,6 +115,7 @@ def execute(
 
         return response
 
+    @with_callbacks
     def shutdown(self) -> None:
         self._shutdown = True