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