"""The kernel routes, moved verbatim out of `Handler._api`.

First slice of the decomposition. `_api` is one method of ~2,100 lines and 261
branches; this is 220 of them, and the group was chosen because it is the only
one that could be *checked*: it owns eleven of the repo's frozen response
shapes, while `memory`, `permissions`, `connectors` and `compute` own none. A
smaller, less entangled group would have been easier and would have proved
nothing.

The handler bodies and their order remain unchanged. The only structural change
since the extraction is that each HTTP method + path matcher now lives in a
`RouteSpec` consumed by both this runtime and the contract inventory. This keeps
routing facts executable instead of asking contract tooling to rediscover them
by parsing the handler source.

Nothing else here was rewritten, and three things that look like oversights are
not: the `store.get_frame(fid) or {}` fallback on the ten permissive routes, the
duplicate `store.get_frame` in `kernel/variables`, and the six live-Notebook
gates (`official_notebook_enabled`, covering the Stage 8 flag and the older
`notebook_repl` override) with `install` deliberately left ungated.

TWO POSITION DEPENDENCIES, written down because a 2,100-line if-chain is
exactly where this kind of thing hides:

* The call site must stay after the `frame_mutation` guard. That guard is the
  only write-protection on the seven mutating routes here, including the
  arbitrary-code-execution endpoint: a quarantined imported session answers 423
  because of it, and nothing in this module re-checks.
* The call site must stay after the `workbench` guard, which is what makes
  `GET /frames/{id}/execution` return 404 for an unknown session. That handler
  has no frame lookup of its own -- unlike its sibling `kernel/variables`,
  which does. Giving it one would remove this dependency; that is a behaviour
  question, so it is deliberately not bundled into a pure move.

The return value is tri-state on purpose. True means a response was emitted;
False means the route group did not handle the request, and the chain must
continue to its 404. `RouteSpec.match()` includes the HTTP method specifically
so a right-path/wrong-method request is not swallowed as handled.
"""

from __future__ import annotations

import os
import re
from typing import Any

from openai4s.server.notebook_lineage import official_notebook_enabled

from . import contract, errors, team_policy

_EXECUTION = contract.RouteSpec(
    "kernel.execution",
    "GET",
    r"/frames/([^/]+)/execution",
    mutates=False,
)
_EXECUTE = contract.RouteSpec(
    "kernel.execute",
    "POST",
    r"/frames/([^/]+)/kernel/execute",
    mutates=True,
)
_RESTART = contract.RouteSpec(
    "kernel.restart",
    "POST",
    r"/frames/([^/]+)/kernel/restart",
    mutates=True,
)
_STOP = contract.RouteSpec(
    "kernel.stop",
    "POST",
    r"/frames/([^/]+)/kernel/stop",
    mutates=True,
)
_INTERRUPT = contract.RouteSpec(
    "kernel.interrupt",
    "POST",
    r"/frames/([^/]+)/kernel/interrupt",
    mutates=True,
)
_START = contract.RouteSpec(
    "kernel.start",
    "POST",
    r"/frames/([^/]+)/kernel/start",
    mutates=True,
)
_VARIABLES = contract.RouteSpec(
    "kernel.variables",
    "GET",
    r"/frames/([^/]+)/kernel/variables",
    mutates=False,
)
_KERNEL = contract.RouteSpec(
    "kernel.status",
    "GET",
    r"/frames/([^/]+)/kernel",
    mutates=False,
)
_STATUS = contract.RouteSpec(
    "session.status",
    "GET",
    r"/frames/([^/]+)/status",
    mutates=False,
)
_INSTALL = contract.RouteSpec(
    "kernel.install",
    "POST",
    r"/frames/([^/]+)/kernel/install",
    mutates=True,
)
_ENVIRONMENTS = contract.RouteSpec(
    "kernel.environments",
    "GET",
    r"/frames/([^/]+)/environments",
    mutates=False,
)
_ENV = contract.RouteSpec(
    "kernel.env",
    "POST",
    r"/frames/([^/]+)/kernel/env",
    mutates=True,
)

# Ordered exactly as the handler chain below. Contract tooling reads this same
# tuple, so adding a route here is both a runtime and an inventory change.
#
# Wrapped in `validate_routes` so a duplicate or a shadowed route raises when
# this module is imported. The validator used to be reachable only from
# `contract.declared_http_routes()`, which `gateway.py` never calls -- so a
# duplicated registry imported cleanly and the daemon served it, and deleting
# the validation call left every test green.
ROUTES = contract.validate_routes(
    (
        _EXECUTION,
        _EXECUTE,
        _RESTART,
        _STOP,
        _INTERRUPT,
        _START,
        _VARIABLES,
        _KERNEL,
        _STATUS,
        _INSTALL,
        _ENVIRONMENTS,
        _ENV,
    )
)


def _require_session_control(self: Any, store: Any, frame_id: str) -> None:
    """Keep readable project sessions from becoming shared namespaces."""

    identity = getattr(self, "_team_identity", None)
    if not team_policy.may_control_session(store, identity, frame_id):
        raise errors.GatewayError(
            403,
            "only the session owner or an admin may control its kernel",
            "owner_only",
        )


def _require_instance_admin(self: Any, operation: str) -> None:
    """Gate instance-global mutations while preserving single-user mode."""

    identity = getattr(self, "_team_identity", None)
    if identity is not None and not identity.is_admin:
        raise errors.GatewayError(
            403,
            f"only an admin may {operation}",
            "admin_only",
        )


#: Every pattern above is under `/frames/`, so a request that is not cannot
#: match any of them -- and 56 of the 91 non-`/frames` routes are declared
#: *after* this group's call site in `Handler._api`, so they used to walk all
#: twelve matchers first. Derived rather than written down: a future route that
#: breaks the assumption shortens the prefix instead of being silently skipped.
#:
#: Truncated at the first regex metacharacter, because the shared *pattern*
#: prefix is `/frames/([^/]+)/` and no real path starts with that -- comparing
#: a literal path against it would reject every kernel request.
_PATH_PREFIX = re.split(
    r"[.^$*+?()\[\]{}|\\]",
    os.path.commonprefix([spec.pattern for spec in ROUTES]),
)[0]


def handle(self, method: str, sub: str, q: dict, runner: Any, store: Any) -> bool:
    """Answer a kernel route, or report that this group does not own it.

    `q` is not optional decoration: `kernel/variables` reads the requested
    language from it, on one line out of 220. An earlier reading of this
    block's dependencies missed it, and the resulting signature raised
    NameError on that route alone -- including on the default path, since
    `q.get` is evaluated before the "python" fallback applies.
    """
    if not sub.startswith(_PATH_PREFIX):
        return False
    m = _EXECUTION.match(method, sub)
    if m:
        self._json(runner.executions.snapshot(m.group(1)))
        return True
    m = _EXECUTE.match(method, sub)
    if m:
        if not official_notebook_enabled(runner.cfg):
            self._json(
                {
                    "error": "notebook REPL is disabled; send a message to resume the agent"
                },
                403,
            )
            return True
        fid = m.group(1)
        _require_session_control(self, store, fid)
        f = store.get_frame(fid) or {}
        pid = f.get("project_id") or "default"
        body = self._body()
        code = body.get("code") or ""
        language = str(body.get("language") or "python").lower()
        if language not in {"python", "r"}:
            self._json({"error": "language must be python or r"}, 400)
            return True
        requested_execution_id = body.get("execution_id")
        if requested_execution_id and not re.fullmatch(
            r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}",
            str(requested_execution_id),
        ):
            self._json({"error": "invalid execution_id"}, 400)
            return True
        job = runner.submit_repl(
            fid,
            pid,
            code,
            language=language,
            execution_id=(
                str(requested_execution_id) if requested_execution_id else None
            ),
        )
        if body.get("wait") is True:
            self._json(job.wait_result())
            return True
        snapshot = runner.executions.snapshot(fid)
        queued = next(
            (
                item
                for item in snapshot.get("queue", [])
                if item.get("execution_id") == job.execution_id
            ),
            (
                snapshot.get("owner")
                if (snapshot.get("owner") or {}).get("execution_id") == job.execution_id
                else None
            ),
        )
        self._json(
            {
                "status": "accepted",
                "frame_id": fid,
                "job_id": job.job_id,
                "execution_id": job.execution_id,
                "owner": job.execution_owner,
                "queue_position": (queued or {}).get("queue_position"),
            },
            202,
        )
        return True
    m = _RESTART.match(method, sub)
    if m:
        if not official_notebook_enabled(runner.cfg):
            self._json(
                {
                    "error": "notebook REPL is disabled; send a message to resume the agent"
                },
                403,
            )
            return True
        fid = m.group(1)
        f = store.get_frame(fid) or {}
        _require_session_control(self, store, fid)
        pid = f.get("project_id") or "default"
        self._json(runner.restart_kernel(fid, pid))
        return True
    m = _STOP.match(method, sub)
    if m:
        if not official_notebook_enabled(runner.cfg):
            self._json(
                {
                    "error": "notebook REPL is disabled; send a message to resume the agent"
                },
                403,
            )
            return True
        fid = m.group(1)
        f = store.get_frame(fid) or {}
        _require_session_control(self, store, fid)
        self._json(runner.stop_kernel(fid, f.get("project_id") or "default"))
        return True
    m = _INTERRUPT.match(method, sub)
    if m:
        if not official_notebook_enabled(runner.cfg):
            self._json(
                {
                    "error": "notebook REPL is disabled; send a message to resume the agent"
                },
                403,
            )
            return True
        fid = m.group(1)
        _require_session_control(self, store, fid)
        body = self._body()
        owner = body.get("owner") or body.get("owner_kind")
        owner_kind = owner.get("kind") if isinstance(owner, dict) else owner
        owner_id = owner.get("id") if isinstance(owner, dict) else body.get("owner_id")
        if not body.get("execution_id") or not owner_kind or not owner_id:
            self._json(
                {
                    "ok": False,
                    "frame_id": fid,
                    "error": ("execution_id, owner.kind, and owner.id are required"),
                    "reason": ("execution_id, owner.kind, and owner.id are required"),
                },
                400,
            )
            return True
        kwargs = {
            "execution_id": body.get("execution_id"),
            "owner": owner,
            "owner_id": str(owner_id),
        }
        self._json(runner.interrupt_kernel(fid, **kwargs))
        return True
    m = _START.match(method, sub)
    if m:
        if not official_notebook_enabled(runner.cfg):
            self._json(
                {
                    "error": "notebook REPL is disabled; send a message to resume the agent"
                },
                403,
            )
            return True
        fid = m.group(1)
        f = store.get_frame(fid) or {}
        _require_session_control(self, store, fid)
        self._json(runner.start_kernel(fid, f.get("project_id") or "default"))
        return True
    m = _VARIABLES.match(method, sub)
    if m:
        fid = m.group(1)
        frame = store.get_frame(fid)
        if frame is None:
            raise errors.GatewayError(404, "session not found")
        if (frame.get("root_frame_id") or fid) != fid:
            raise errors.GatewayError(
                409,
                "variable inspection requires the current root session",
            )
        language = str((q.get("language") or ["python"])[0]).lower()
        if language not in {"python", "r"}:
            self._json({"error": "language must be python or r"}, 400)
            return True
        self._json(runner.variables.inspect(fid, language))
        return True
    m = _KERNEL.match(method, sub)
    if m:
        self._json(runner.kernel_status(m.group(1)))
        return True
    m = _STATUS.match(method, sub)
    if m:
        fid = m.group(1)
        self._json(
            {
                "frame_id": fid,
                "running": runner.is_running(fid),
                # The frame's own terminal state, which `running` cannot give:
                # `running: false` is true of a session that completed, one
                # that was cancelled and one that failed alike. A client
                # reopening a session had no authoritative way to learn which,
                # so it could not restore a failure that had already ended --
                # and the alternative, reading a cached session list, is stale
                # by construction.
                "status": (runner.store.get_frame(fid) or {}).get("status"),
                "kernel": runner.kernel_status(fid),
            }
        )
        return True
    m = _INSTALL.match(method, sub)
    if m:
        # NOT gated by notebook_repl: prebuilt-env package install is a
        # separate Customize → Compute affordance, not the code REPL. Both
        # this route and global /kernel/install are nevertheless admin-only
        # in team mode because they mutate the shared runtime environment.
        _require_instance_admin(self, "install shared kernel packages")
        fid = m.group(1)
        f = store.get_frame(fid) or {}
        pid = f.get("project_id") or "default"
        b = self._body()
        pkgs = b.get("packages") or ([b["package"]] if b.get("package") else [])
        self._json(
            runner.install_packages(
                pkgs,
                root_frame_id=fid,
                project_id=pid,
                restart=b.get("restart", True),
            )
        )
        return True
    # prebuilt-environment selection for this session's kernel
    m = _ENVIRONMENTS.match(method, sub)
    if m:
        self._json(runner.list_environments(m.group(1)))
        return True
    m = _ENV.match(method, sub)
    if m:
        if not official_notebook_enabled(runner.cfg):
            self._json(
                {
                    "error": "notebook REPL is disabled; send a message to resume the agent"
                },
                403,
            )
            return True
        fid = m.group(1)
        f = store.get_frame(fid) or {}
        _require_session_control(self, store, fid)
        pid = f.get("project_id") or "default"
        b = self._body()
        name = b.get("env") or b.get("name") or ""
        self._json(runner.set_env(fid, name, pid))
        return True
    return False