""" Python Executor - Persistent Python execution environment. Similar to a Jupyter kernel: maintains state and tool namespace across calls. """ from typing import Any from .base import Executor # noqa: F401 — imported for the conformance check below from .namespace_kernel import MAX_OUTPUT_CHARS, NamespaceKernel # noqa: F401 __all__ = ["PythonExecutor", "MAX_OUTPUT_CHARS"] class PythonExecutor: """ Persistent Python execution environment. Maintains namespace (variables, functions) across multiple execute() calls. This is the default `in_process` executor (`EXECUTOR=in_process`, see `base.get_executor`): it runs generated code with `exec` in THIS process. It conforms structurally to the `Executor` protocol (`base.Executor`) — the module-level assertion below pins that contract. ADR-0007 Phase 1 adds a `sandbox` executor implementing the same `Executor` interface, running code in an isolated per-session container instead of in-process; the agent depends only on the interface, so that swap needs no agent-loop change. The exec/stdout-capture/namespace-seeding semantics now live in the shared `NamespaceKernel` (`namespace_kernel.py`), so the in-process executor and the sandbox exec-kernel can never drift. This class is a thin in-process wrapper over that core; its public surface is unchanged. """ def __init__(self): self._kernel = NamespaceKernel() @property def namespace(self) -> dict[str, Any]: """The live execution namespace (kept for callers that inspect it).""" return self._kernel.namespace def reset_environment(self): """Reset the execution environment to a clean state.""" self._kernel.reset() def send_functions(self, functions: dict[str, Any]): """Inject functions into the execution namespace.""" self._kernel.send_functions(functions) def send_variables(self, variables: dict[str, Any]): """Inject variables into the execution namespace.""" self._kernel.send_variables(variables) def __call__(self, code: str) -> Any: return self.execute(code) def execute(self, code: str) -> Any: """Execute Python code in the persistent namespace, returning stdout output.""" return self._kernel.execute(code) # PythonExecutor is the default in-process implementation of the Executor # seam (ADR-0007 Phase 0). `Executor` is a runtime_checkable Protocol, so this # structural check pins the contract at import time — if a method is renamed or # dropped here, import fails loudly rather than drifting from the interface. # `issubclass` (not `isinstance`) keeps this a class-level, method-presence # check with no instantiation — so importing this module stays free of the # pandas/numpy side effects `PythonExecutor()` would trigger. assert issubclass(PythonExecutor, Executor), ( "PythonExecutor no longer conforms to the Executor protocol (base.Executor)" )