ive harnessed the harness
Something went wrong. Try again.
4.2 kB · 90 lines
Python
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091"""Persistent CPython evaluator; all history and recovery ownership stays outside."""from __future__ import annotations
import __future__import astimport asyncioimport inspectimport linecacheimport tracebackfrom collections import dequefrom 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, "trace": execution.trace.finish()}