"""Persistent CPython evaluator; all history and recovery ownership stays outside."""
from __future__ import annotations
import __future__
import ast
import asyncio
import inspect
import linecache
import traceback
from collections import deque
from collections.abc import Mapping
from .context import Bridge, ExecutionScope, TurnYield, current
_FUTURE_MASK = sum(getattr(__future__, name).compiler_flag for name in __future__.all_feature_names)
class Kernel:
def __init__(self, bridge: Bridge, prelude: Mapping[str, object] | None = None):
self.bridge = bridge
# Bound in every generation, and rebuilt by `clear()` from the same place: a reset that
# dropped the vocabulary would make the verbs appear and disappear between cells.
self.prelude: dict[str, object] = dict(prelude or {})
self.globals: dict = self._namespace()
self.flags = ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
self._source_names: deque[str] = deque()
def _namespace(self) -> dict:
return {"__name__": "__klbr_workbench__", "__builtins__": __builtins__, **self.prelude}
def clear(self) -> dict[str, Any]:
"""Clear the namespace globals without claiming to cancel background tasks.
Live asyncio tasks retain their references and continue running;
namespace clear is not task cancellation.
"""
self.globals = self._namespace()
self.flags = ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
return {"cleared": True, "tasks_cancelled": 0}
def display(self, value: object) -> None:
if value is not None:
self.globals["_"] = value
execution = current.get()
if execution is not None:
execution.write("result", repr(value))
async def execute(self, operation_id: str, source: str, purpose: str = "interactive") -> dict:
if purpose not in {"interactive", "maintenance", "eval"}:
raise ValueError(f"invalid execution purpose: {purpose}")
execution = ExecutionScope(operation_id, self.bridge, purpose=purpose)
token = current.set(execution)
self.globals["__purpose__"] = purpose
status, error = "ok", None
filename = f"<klbr:{operation_id}>"
linecache.cache[filename] = (len(source), None, source.splitlines(True), filename)
self._source_names.append(filename)
while len(self._source_names) > 64:
linecache.cache.pop(self._source_names.popleft(), None)
try:
tree = compile(source, filename, "exec", flags=self.flags | ast.PyCF_ONLY_AST, dont_inherit=True)
if tree.body and isinstance(tree.body[-1], ast.Expr):
expression = tree.body[-1]
tree.body[-1] = ast.copy_location(ast.Expr(value=ast.Call(
func=ast.Name(id="__klbr_display__", ctx=ast.Load()),
args=[expression.value], keywords=[])), expression)
ast.fix_missing_locations(tree)
self.globals["__klbr_display__"] = self.display
code = compile(tree, filename, "exec", flags=self.flags, dont_inherit=True)
self.flags |= code.co_flags & _FUTURE_MASK
result = eval(code, self.globals)
if inspect.isawaitable(result):
await result
except TurnYield:
status = "yielded"
except asyncio.CancelledError:
status = "interrupted"
except BaseException as failure:
# SystemExit/KeyboardInterrupt are cell outcomes, not instructions to
# kill the host. os._exit and native failures still kill the worker.
status = "error"
error = "".join(traceback.format_exception(failure))[-16_384:]
finally:
self.bridge.close_execution(execution)
current.reset(token)
# Catching the yield sentinel does not restore effect authority.
status = execution.terminal or status
return {"kind": "cell", "status": status, "output": execution.output,
"truncated_bytes": execution.truncated_bytes, "error": error}