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,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

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")