"""core/script_sandbox.py — WAVE 36 (R5 / R10, contract C1): running a tenant's OWN Python.

Owner item 6: *"Add code script as an interface (database View) so a user can build whatever they
want through the Agent chat interface."* Item 8: *"We need to really guardrail the reach of this
script. So let's really grill this down."* R10 ruled it SERVER-SIDE PYTHON after the trade was
stated, so this file is the guardrail, and one engine serves both items.

════════════════════════════════════════════════════════════════════════════════════════════════
⛔⛔ THE ONE PARAGRAPH TO READ BEFORE CHANGING ANYTHING HERE.

In-process CPython cannot deliver two of this ticket's clauses. An AST allow-list plus a curated
namespace stops import, file, network and environment access — but it **cannot cap memory and
cannot interrupt a runaway loop**, because a `while True:` in the same interpreter is not a slow
request, it is the tenant's ONE FastAPI process gone. So the script runs in a **SUBPROCESS**:
`resource.setrlimit` for address space and CPU, a hard wall-clock kill from the parent, and the
allow-list inside. Neither half is sufficient; both are load-bearing.

⭐ AND THE SUBPROCESS RECEIVES **ROWS, NEVER A STORE**. The parent calls C1's `scoped_table` under
the CALLING user's record and serialises the result; the child imports nothing from this repo and
holds no credential, no runtime and no store handle. Wiring W1 ("the sandbox has no second store
path") is then true by CONSTRUCTION rather than by discipline, and it is checkable: the child
reports its own `sys.modules`, and no `core.*` name may appear in it.

⛔ NEVER A BLACKLIST. Every rule below is an ALLOW-LIST — a set of node types, a set of attribute
names, a dict of builtins. A blacklist of dangerous spellings is bypassable by construction, and
the bypass is usually one string method away (`"{0.__class__}".format(x)` performs its attribute
lookup inside `format`, so there is no `ast.Attribute` node to refuse).
════════════════════════════════════════════════════════════════════════════════════════════════

The two layers, and they refuse DIFFERENT things on purpose:

  1. `check_source()` — a pure function over source text. Refuses a construct the language offers
     and this sandbox does not: `import`, `class`, `with`, `async`, `yield`, `global`, and every
     attribute name outside `ALLOWED_ATTRS`.
  2. `SANDBOX_BUILTINS` — the names that resolve at all. `__import__`, `open`, `eval`, `exec`,
     `compile`, `getattr`, `globals`, `vars` and `type` are simply absent, so a source that gets
     past layer 1 still finds nothing to call.

⚠ THAT DUPLICATION IS DELIBERATE AND IT CHANGES HOW THE GATE MUST BE WRITTEN. `import os` is
refused twice, so a negative control that drops ONE layer sees the other refuse and reports
green — the shape that already cost this wave one missed control in `routes_agent_harness`. So
each layer is tested AT ITS OWN BOUNDARY: `check_source()` is called directly on source strings,
and `run()` is driven end to end. An NC drops one entry from one frozenset and the matching
boundary goes red.
"""
import ast
import json
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path

#: Wall clock, enforced by the PARENT with a kill. The one cap that works on every platform.
DEFAULT_TIMEOUT_S = 10.0

#: Address space for the child (`RLIMIT_AS`). POSIX only — see `run()`'s `caps` report.
DEFAULT_MEMORY_BYTES = 512 * 1024 * 1024

#: CPU seconds for the child (`RLIMIT_CPU`). POSIX only. Deliberately above the wall clock: the
#: wall-clock kill is the primary control and this is the backstop for a child that stops being
#: reachable. A CPU limit BELOW the timeout would make every slow script look like a CPU refusal.
DEFAULT_CPU_SECONDS = 15

#: What the script may print, in bytes. `print` is a curated builtin writing to a capped buffer,
#: and the child's real stdout goes to DEVNULL — so a script cannot fill a pipe, and anything
#: that escaped far enough to write to fd 1 has nowhere for it to land.
MAX_STDOUT_BYTES = 64 * 1024

#: The serialised ROW payload handed to the child. ⛔ A REFUSAL, NEVER A TRUNCATION (standing rule
#: 1): a short answer from a data tool is a wrong answer that looks right. Over this, `run()`
#: returns a named limit carrying its cause and a recommendation.
MAX_PAYLOAD_BYTES = 32 * 1024 * 1024

#: The emitted spec. A render spec is a description of a picture; one larger than this is data
#: pretending to be a description.
MAX_SPEC_BYTES = 2 * 1024 * 1024

MAX_SOURCE_BYTES = 128 * 1024


# ══════════════════════════════════════════════════════ LAYER 1 — the AST allow-list ═══════════
#: Every `ast` node class a script may contain. ⛔ THE ABSENCES ARE THE POLICY: `Import` /
#: `ImportFrom` (no module reaches the script), `ClassDef` (a class body is a namespace with its
#: own scoping rules and buys a data script nothing), `With` (a context manager is `__enter__`
#: by another spelling), `Global` / `Nonlocal` (rebinding the sandbox's own names), and every
#: `Async*` / `Await` / `Yield` form (this engine is synchronous; a coroutine that is never
#: awaited is a silent no-op that looks like a working script).
ALLOWED_NODES = frozenset("""
Module Expr Assign AugAssign AnnAssign NamedExpr Return Pass Break Continue Delete Assert Raise
If For While Try TryStar ExceptHandler FunctionDef Lambda arguments arg keyword
BoolOp BinOp UnaryOp IfExp Dict Set List Tuple Starred Subscript Slice Compare Call Attribute Name
Constant JoinedStr FormattedValue ListComp SetComp DictComp GeneratorExp comprehension
Load Store Del
And Or Not Invert UAdd USub
Add Sub Mult Div FloorDiv Mod Pow LShift RShift BitOr BitXor BitAnd MatMult
Eq NotEq Lt LtE Gt GtE Is IsNot In NotIn
""".split())

#: Every attribute name a script may READ or CALL. ⛔⛔ THIS IS THE LOAD-BEARING SET, and it is
#: an allow-list of NAMES rather than a refusal of dunders, because the interesting escapes are
#: ordinary-looking: `f.__globals__` on any function reaches the runner's own module namespace,
#: `e.__traceback__.tb_frame.f_globals` reaches it from an exception handler, and `().__class__`
#: reaches `object.__subclasses__`. None of those names is here, and neither is any name this
#: sandbox has not been asked for.
#: ⚠ `format` IS ABSENT DELIBERATELY. `"{0.__class__}".format(x)` performs the attribute lookup
#: INSIDE `str.format`, where no `ast.Attribute` node exists for layer 1 to see. f-strings are
#: fine — `f"{x.__class__}"` compiles to a real `Attribute` node and is refused.
ALLOWED_ATTRS = frozenset("""
append extend insert pop remove clear sort reverse copy count index
keys values items get setdefault update
add discard union intersection difference issubset issuperset
join split rsplit splitlines strip lstrip rstrip lower upper title capitalize casefold
replace startswith endswith find rfind zfill ljust rjust center partition removeprefix removesuffix
isdigit isalpha isalnum isspace isupper islower isnumeric
real imag numerator denominator
""".split())


class Refused(Exception):
    """A named refusal: `code` for a caller to branch on, `message` for a person to read."""

    def __init__(self, code, message):
        self.code, self.message = code, message
        super().__init__(f"{code}: {message}")


def _attr_ok(name):
    """An attribute name passes only if it is on the list AND is not private.

    ⚠ THE SECOND TEST IS NOT A BLACKLIST — it narrows an allow-list that already excludes every
    private name. It is here so that adding a name to `ALLOWED_ATTRS` cannot open a dunder by
    accident, which is the one edit a future reader is most likely to make in a hurry.
    """
    return name in ALLOWED_ATTRS and not name.startswith("_")


def check_source(source):
    """LAYER 1. Return a `Refused` for source this sandbox will not run, or `None`.

    ⭐ PURE, AND THAT IS WHAT MAKES IT TESTABLE AT ITS OWN BOUNDARY. It reads no file, spawns no
    process and touches no store, so a gate can hand it a hundred hostile strings for free and an
    NC can drop one entry from one frozenset and watch exactly this function change its answer.
    """
    text = str(source or "")
    if len(text.encode("utf-8", "replace")) > MAX_SOURCE_BYTES:
        return Refused("source_too_long",
                       f"a script view is at most {MAX_SOURCE_BYTES // 1024} KB of source")
    try:
        tree = ast.parse(text)
    except SyntaxError as exc:
        return Refused("syntax", f"line {exc.lineno or 0}: {exc.msg}")

    for node in ast.walk(tree):
        kind = type(node).__name__
        if kind not in ALLOWED_NODES:
            return Refused("refused_construct",
                           f"line {getattr(node, 'lineno', 0)}: this sandbox does not run "
                           f"{_english(kind)}")
        if isinstance(node, ast.Attribute) and not _attr_ok(node.attr):
            return Refused("refused_attribute",
                           f"line {getattr(node, 'lineno', 0)}: the attribute "
                           f"'{node.attr}' is not available inside a script view")
        # ⛔ A NAME may not be private either. `_` prefixed names are the runner's own, and a
        # script that could bind one could shadow the machinery it runs on top of.
        if isinstance(node, ast.Name) and node.id.startswith("_"):
            return Refused("reserved_name",
                           f"line {getattr(node, 'lineno', 0)}: names starting with an "
                           f"underscore are reserved by the sandbox")
        if isinstance(node, (ast.FunctionDef, ast.arg, ast.ExceptHandler)) and str(
                getattr(node, "name", None) or getattr(node, "arg", "") or "").startswith("_"):
            return Refused("reserved_name",
                           f"line {getattr(node, 'lineno', 0)}: names starting with an "
                           f"underscore are reserved by the sandbox")
        if isinstance(node, ast.keyword) and str(node.arg or "").startswith("_"):
            return Refused("reserved_name",
                           f"line {getattr(node, 'lineno', 0)}: keyword arguments starting with "
                           f"an underscore are reserved by the sandbox")
    return None


_ENGLISH = {
    "Import": "an import", "ImportFrom": "an import", "ClassDef": "a class definition",
    "With": "a with block", "AsyncWith": "a with block", "AsyncFor": "an async loop",
    "AsyncFunctionDef": "an async function", "Await": "await", "Yield": "yield",
    "YieldFrom": "yield from", "Global": "a global statement", "Nonlocal": "a nonlocal statement",
    "Match": "a match statement",
}


def _english(kind):
    return _ENGLISH.get(kind, f"a {kind} expression")


# ══════════════════════════════════════════ LAYER 2 — the namespace, and the child program ═════
#: The builtins a script may reach, BY NAME. Everything else is a `NameError` in the child.
#: ⛔ THE ABSENCES, again, are the policy: `__import__` `open` `eval` `exec` `compile` `input`
#: `getattr` `setattr` `delattr` `globals` `locals` `vars` `dir` `type` `super` `object` `help`
#: `exit` `breakpoint` `memoryview` `id`. Several are harmless on their own; each one is a step
#: on a published escape, and none has ever been asked for by a script that shapes rows.
#: ⚠ THE EXCEPTION CLASSES ARE HERE BECAUSE `try:` IS, and a `try` block whose `except` clause
#: cannot name what it catches is a construct that reads as supported and is not. They are safe
#: for the same reason everything else is: `Exception.__subclasses__` needs an attribute this
#: sandbox does not allow, so a class object in the namespace is a leaf, not a doorway.
SANDBOX_BUILTIN_NAMES = (
    "abs all any bool bytes callable chr dict divmod enumerate filter float frozenset hash hex "
    "int isinstance issubclass iter len list map max min next oct ord pow range repr reversed "
    "round set slice sorted str sum tuple zip True False None "
    "Exception ValueError TypeError KeyError IndexError ZeroDivisionError ArithmeticError "
    "AttributeError StopIteration OverflowError"
).split()

#: The literal program the child runs. It is TEXT rather than a module because the child must
#: import nothing from this repo: a module would be found on `sys.path` and would drag `core`
#: with it, which is exactly the second store path W1 forbids.
#: ⚠ Every name in here is underscore-prefixed and layer 1 refuses a script from binding one, so
#: the runner's own machinery cannot be shadowed by the source it executes.
_RUNNER = r'''
import json as _json, os as _os, sys as _sys

_pay = _json.loads(open(_sys.argv[1], "r", encoding="utf-8").read())
_out = {"ok": False, "code": "not_run", "message": "the script did not run",
        "stdout": "", "spec": None, "caps": {"wallClock": True, "memory": False, "cpu": False}}

# ── the caps this platform can actually apply, reported either way (standing rule 1) ──────────
try:
    import resource as _res
    _mem = int(_pay["memoryBytes"])
    _res.setrlimit(_res.RLIMIT_AS, (_mem, _mem))
    _out["caps"]["memory"] = True
    _cpu = int(_pay["cpuSeconds"])
    _res.setrlimit(_res.RLIMIT_CPU, (_cpu, _cpu))
    _out["caps"]["cpu"] = True
except Exception:
    # `resource` is POSIX only. The wall-clock kill in the parent still applies, and `caps` says
    # which of the three held, never a silent partial.
    pass

_printed = []
_spent = [0]
_LIMIT = int(_pay["maxStdout"])


def _print(*_a, **_k):
    _text = (_k.get("sep") or " ").join(str(_x) for _x in _a) + (_k.get("end") or "\n")
    _room = _LIMIT - _spent[0]
    if _room > 0:
        _printed.append(_text[:_room])
        _spent[0] += len(_text)


class _Refusal(Exception):
    """The SANDBOX refusing, as distinct from the SCRIPT failing.

    Without its own class these arrive as `ValueError`, indistinguishable from a `ValueError` the
    script raised itself, and the answer then says "refused" about an ordinary bug in the tenant's
    own code. Two different facts, two different codes.
    """


_emitted = []


def _emit(_spec):
    if not isinstance(_spec, dict):
        raise _Refusal("emit() takes a view spec, which is a dictionary")
    if _emitted:
        raise _Refusal("emit() was already called; a script view emits exactly one view")
    _emitted.append(_spec)


_rows = _pay["rows"]
_fields = _pay["fields"]
_bound = _pay["table"]


def _scoped_table(_table=None):
    if _table is not None and str(_table) != _bound:
        raise _Refusal(
            "this script view is bound to the database '" + _bound + "' and asked for '"
            + str(_table) + "'. A script view reads its own database only")
    return [dict(_r) for _r in _rows]


def _scoped_fields():
    return [dict(_f) for _f in _fields]


_ns = {"__builtins__": {_n: __builtins__[_n] if isinstance(__builtins__, dict)
                        else getattr(__builtins__, _n)
                        for _n in _pay["builtins"]}}
_ns["__builtins__"]["print"] = _print
_ns["print"] = _print
_ns["emit"] = _emit
_ns["scoped_table"] = _scoped_table
_ns["scoped_fields"] = _scoped_fields
_ns["table"] = _bound

try:
    exec(compile(_pay["source"], "<script view>", "exec"), _ns)
    if not _emitted:
        _out.update(ok=False, code="no_view",
                    message="the script finished without calling emit(spec)")
    else:
        _out.update(ok=True, code="", message="", spec=_emitted[0])
except _Refusal as _e:
    _out.update(ok=False, code="refused", message=str(_e)[:400])
except MemoryError:
    _out.update(ok=False, code="memory",
                message="the script used more memory than a script view is allowed")
except NameError as _e:
    _out.update(ok=False, code="refused_name",
                message=str(_e)[:200] + ". A script view may use only the names the sandbox "
                                        "provides")
except BaseException as _e:
    _out.update(ok=False, code="error",
                message=type(_e).__name__ + ": " + str(_e)[:400])

_out["stdout"] = "".join(_printed)
_out["truncated"] = _spent[0] > _LIMIT
# ⭐ THE PROBE: what this child actually had. The PARENT strips it unless it was asked for, so a
# production run never carries it and a gate can still prove that no `core.*` module and no
# secret-shaped environment key was ever inside this process.
# ⚠ SNAPSHOTTED AFTER `exec`, and `os` is imported at the TOP so this list does not depend on
# dict-literal evaluation order. The first draft called `__import__("os")` inside this very
# expression, so whether `os` appeared depended on which value Python built first: a probe whose
# contents move with an unrelated edit is a probe a gate cannot assert against.
_out["probe"] = {"modules": sorted(_sys.modules), "env": sorted(_os.environ)}
open(_sys.argv[2], "w", encoding="utf-8").write(_json.dumps(_out, default=str))
'''


def _child_env():
    """The child's WHOLE environment. An allow-list of two keys, and neither is a credential.

    ⛔ NOT `os.environ.copy()` MINUS SOMETHING. A subtractive environment ships every key nobody
    thought to name: `HF_TOKEN`, `ODOO_PASSWORD`, `ANTHROPIC_API_KEY` and whatever the next
    connector adds. The three names below are here because Python will not start on Windows
    without them; on Linux this returns `{}` and the child runs with no environment at all.
    """
    env = {}
    for name in ("SystemRoot", "SYSTEMROOT", "WINDIR"):
        if os.environ.get(name):
            env[name] = os.environ[name]
    return env


def run(source, rows, fields, table_key, *, timeout_s=DEFAULT_TIMEOUT_S,
        memory_bytes=DEFAULT_MEMORY_BYTES, cpu_seconds=DEFAULT_CPU_SECONDS, probe=False):
    """Run ONE script over rows that are ALREADY scoped. Returns C3's envelope plus `caps`.

        {ok, code, message, spec, stdout, truncated, ms, caps: {wallClock, memory, cpu}}

    ⛔ THIS FUNCTION NEVER TOUCHES A STORE, AND THAT IS THE POINT: it takes rows. `run_view()`
    below is the door that fetches them through C1; keeping the two apart is what lets a gate
    drive the sandbox with no tenant, no runtime and no credential anywhere in the process.

    ⚠ `caps` IS PART OF THE ANSWER, NOT DEBUG OUTPUT. On Windows `resource` does not exist, so
    the memory and CPU limits are NOT applied and this says so. A caller that reports `ok:true`
    without reading `caps` is claiming an enforcement that did not happen (standing rule 1).
    """
    started = time.monotonic()
    refusal = check_source(source)
    if refusal is not None:
        return _refusal(refusal.code, refusal.message, started)

    payload = {"source": str(source or ""), "rows": rows, "fields": fields,
               "table": str(table_key or ""), "builtins": SANDBOX_BUILTIN_NAMES,
               "maxStdout": MAX_STDOUT_BYTES, "memoryBytes": int(memory_bytes),
               "cpuSeconds": int(cpu_seconds)}
    try:
        blob = json.dumps(payload, default=str)
    except (TypeError, ValueError) as exc:
        return _refusal("bad_rows", f"these rows cannot be handed to a script ({exc})", started)
    if len(blob.encode("utf-8", "replace")) > MAX_PAYLOAD_BYTES:
        # ⛔ REPORTED, NOT TRUNCATED (standing rule 1's second sentence): cause and recommendation,
        # in the words the owner asked for, rather than a quietly short answer.
        return _refusal(
            "payload_too_large",
            f"this database's rows are larger than the {MAX_PAYLOAD_BYTES // (1024 * 1024)} MB a "
            f"script view can be handed at once. Narrow the view with a filter, or raise the "
            f"sandbox payload limit for this deployment", started)

    with tempfile.TemporaryDirectory(prefix="aios-script-") as work:
        pay_path = Path(work) / "payload.json"
        res_path = Path(work) / "result.json"
        pay_path.write_text(blob, encoding="utf-8")
        # `-I` isolates the interpreter (no PYTHON* env, no user site), `-S` skips site-packages,
        # and the program arrives on STDIN so there is no file for anything to import it as.
        # ⛔ `-X utf8` AND AN EXPLICIT `encoding` ARE NOT TIDINESS. Without them this pipe is
        # encoded with the parent's locale codec, which on this Windows box is cp1252: the runner
        # text below cannot be represented in it and `subprocess.run` died with
        # `UnicodeEncodeError` before the child ever started. A sandbox whose behaviour depends on
        # the operator's locale is a sandbox with two behaviours. `-I` implies `-E`, so
        # `PYTHONUTF8` in the environment could not have carried this, it has to be a flag.
        argv = [sys.executable, "-I", "-S", "-X", "utf8", "-", str(pay_path), str(res_path)]
        try:
            done = subprocess.run(
                argv, input=_RUNNER, text=True, encoding="utf-8", errors="replace",
                cwd=work, env=_child_env(),
                stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=timeout_s)
        except subprocess.TimeoutExpired:
            return _refusal("timeout",
                            f"the script ran longer than {timeout_s:g} seconds and was stopped",
                            started)
        except OSError as exc:
            return _refusal("no_sandbox",
                            f"a script view could not be started on this deployment ({exc})",
                            started)

        if not res_path.is_file():
            # The child died without writing an answer: an rlimit signal, an OOM kill, or a crash.
            # ⚠ NAMED BY ITS RETURN CODE rather than reported as a generic failure — a memory kill
            # and a bug in this file must not read identically to an operator.
            return _refusal(*_died(done), started)
        try:
            out = json.loads(res_path.read_text(encoding="utf-8"))
        except (OSError, ValueError) as exc:
            return _refusal("unreadable", f"the script's answer could not be read ({exc})",
                            started)

    out["ms"] = int((time.monotonic() - started) * 1000)
    if out.get("ok"):
        spec_error = _check_spec(out.get("spec"))
        if spec_error:
            out.update(ok=False, code="bad_spec", message=spec_error, spec=None)
    if not probe:
        out.pop("probe", None)
    return out


def _died(done):
    """`(code, message)` for a child that produced no answer."""
    rc = done.returncode
    tail = " ".join((done.stderr or "").split())[-300:]
    if rc in (-9, 137):
        return "memory", "the script was stopped for using too much memory"
    if rc in (-24, 152):
        return "timeout", "the script used more processor time than a script view is allowed"
    return "crashed", f"the script view engine stopped without an answer{': ' + tail if tail else ''}"


def _refusal(code, message, started):
    return {"ok": False, "code": code, "message": message, "spec": None, "stdout": "",
            "truncated": False, "ms": int((time.monotonic() - started) * 1000),
            "caps": {"wallClock": True, "memory": False, "cpu": False}}


def _check_spec(spec):
    """C3: a spec is a DESCRIPTION the client draws. Never HTML, never a script, never a URL.

    ⛔ THE CHECK IS ON THE KEYS, NOT ON THE STRING CONTENTS. Scanning values for `<script>` is a
    blacklist and would pass `<SCR` + `IPT>`; refusing a spec that carries an `html`, `script`,
    `src` or `onclick` key refuses the SHAPE that would let a renderer be talked into executing
    something. The vocabulary of legal `kind`s is the ROUTE's business (W36-T37) — this is the
    floor every caller gets whether or not the route above it remembers.
    """
    if not isinstance(spec, dict):
        return "the script emitted something that is not a view spec"
    try:
        blob = json.dumps(spec)
    except (TypeError, ValueError):
        return "the emitted view spec is not something the client can be sent"
    if len(blob.encode("utf-8", "replace")) > MAX_SPEC_BYTES:
        return (f"the emitted view spec is over {MAX_SPEC_BYTES // (1024 * 1024)} MB. A view spec "
                f"describes a picture; it is not where the rows go")
    banned = {"html", "innerhtml", "script", "src", "srcdoc", "href", "style", "onclick", "onload"}
    found = sorted(k for k in _keys_of(spec) if str(k).lower() in banned)
    if found:
        return (f"a view spec may not carry {', '.join(found)}. The client DRAWS a spec, so a "
                f"markup or URL key would be a script by another name")
    return None


def _keys_of(value, depth=0):
    """Every key anywhere in a nested spec. Bounded, so a deep structure cannot spin this."""
    if depth > 12:
        return
    if isinstance(value, dict):
        for key, sub in value.items():
            yield key
            yield from _keys_of(sub, depth + 1)
    elif isinstance(value, (list, tuple)):
        for sub in value:
            yield from _keys_of(sub, depth + 1)


# ══════════════════════════════════════════════ THE DOOR — C1 is the ONLY way to a row ═════════
def run_view(user, table_key, source, st=None, **kw):
    """Fetch through C1 under `user`'s scope, then run the script over what came back (R5).

    ⭐⭐ THE FETCH HAPPENS IN THE PARENT AND ONLY ROWS CROSS INTO THE CHILD. That is wiring W1
    made structural: the child has no runtime to ask, no store handle to open and no credential
    to use, so "a script cannot read what its caller cannot read" is not a rule anybody has to
    keep — there is no second path for it to be broken through.

    ⛔ C1'S THREE EXCEPTIONS ARE ANSWERED, NEVER SWALLOWED. `UnknownTable`, `Denied` and
    `Unresolvable` mean three different things to a person; collapsing them into "no rows" is the
    silent-empty answer C1 was written to make impossible. `Unresolvable.as_limit()` is handed
    through in the words it was raised with — standing rule 1's second sentence, verbatim.
    """
    import core.perm_scope as perm_scope

    key = str(table_key or "")
    try:
        rows = perm_scope.scoped_table(user, key, st=st)
        fields = perm_scope.scoped_fields(user, key, st=st)
    except perm_scope.UnknownTable as exc:
        return _refusal("unknown_table", str(exc) or f"there is no database '{key}'",
                        time.monotonic())
    except perm_scope.Denied as exc:
        return _refusal("denied", str(exc) or "this account may not read that database",
                        time.monotonic())
    except perm_scope.Unresolvable as exc:
        out = _refusal("unresolvable", str(exc), time.monotonic())
        out["limit"] = exc.as_limit()
        return out
    return run(source, rows, fields, key, **kw)