feat(dspy): add interpreter-level callbacks (on_interpreter_*)

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.

dspy/primitives/code_interpreter.py

@@ -80,6 +80,14 @@ 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 `execute`, `start`, and
+        `shutdown`, and route sandbox->host tool dispatch through a decorated seam so the
+        `on_interpreter_*` handlers fire. `dspy.PythonInterpreter` does this out of the box.
     """
 
     @property

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,23 @@ 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.
+
+        Extracted from `_handle_tool_call` so it can be decorated with `@with_callbacks`:
+        this is the decoratable seam where the real exception is still in flight, before
+        `_handle_tool_call` converts it into a JSON-RPC error response. Decorating
+        `_handle_tool_call` directly would be useless because it swallows every exception,
+        so the end handler's `exception` would always be 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 +344,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"},
@@ -507,6 +526,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,
@@ -588,6 +608,7 @@ def execute(
 
         raise CodeInterpreterError(f"Too many non-JSON lines ({skipped}) during execution")
 
+    @with_callbacks
     def start(self) -> None:
         """Initialize the Deno/Pyodide sandbox.
 
@@ -612,6 +633,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")

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,24 @@ 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.
+_INTERPRETER_START_HANDLERS = {
+    "execute": "on_interpreter_execute_start",
+    "start": "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",
+    "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 +525,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 +557,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