src/managers/execution/python_executor.py

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