api/routes_script_views.py

"""routes_script_views.py — CONTRACT C3: a database View that is a PYTHON SCRIPT (R3 / R5 / R10).

Owner item 6, verbatim (2026-08-18): *"Add code script as an interface (database View) so a user
can build whatever they want through the Agent chat interface. be able to create any dashboard
they want. User should have the ability to see the code AND the dashboard output of course… Limit
the code script View per database… Any agent can add into more AI script, so we can see different
versions or different things the AI code for us."*

    GET    /api/v1/script-views?database=K      the views bound to ONE database
    POST   /api/v1/script-views                 create one   {database, name?, source}
    GET    /api/v1/script-views/{id}            one view, its source and its history
    PUT    /api/v1/script-views/{id}            a NEW VERSION of the source
    DELETE /api/v1/script-views/{id}            drop it
    POST   /api/v1/script-views/{id}/run        run it  ->  {ok, spec | error, stdout, ms}

⭐⭐ **R3 IS "SCOPED, NOT CAPPED", AND THE TWO HALVES POINT OPPOSITE WAYS.** *Scoped*: a view may
read ONLY the database it lives in, and a script that names another database is REFUSED with a
message naming both. *Not capped*: there is **no limit on how many script views a database may
carry**, because that is how an agent offers three attempts and the owner picks one. So nothing
below counts views. What IS bounded is what makes them big — one source is capped, and one view's
edit history is capped and REPORTS what it dropped.

⛔ **THE RUN IS THE CALLER'S, NEVER THE AUTHOR'S (R5).** `script_sandbox.run_view` is handed
`session.user`, so a script written by an administrator and opened by a scoped analyst reads the
ANALYST's rows. The author decides what the code does; the reader decides what it can see.
⚠ And the reverse case is safe rather than lucky: a narrowly-scoped author cannot write a script
that exfiltrates anything, because the only thing a script can return is a render spec drawn on
the screen of the person who ran it. There is no network, no file and no second reader.

⛔ **`run` IS `def`, NOT `async def`.** It waits on a subprocess for up to ten seconds; as a
coroutine that would block the event loop for every other request in the container. FastAPI runs a
plain `def` in the threadpool, which is what makes one slow script one slow REQUEST.
"""
import threading
from datetime import datetime, timezone

from fastapi import APIRouter, Body, Depends

from deps import Session, err, require_session

router = APIRouter(prefix="/api/v1")

#: The tenant's script views: `{id: record}`. Per tenant, so it rides `runtime.store_key`.
VIEWS_KEY = "script_views"

MAX_NAME = 80
MAX_SOURCE_BYTES = 128 * 1024

#: Edit history per view. ⚠ NOT a cap on the NUMBER of views (R3 forbids that) — a cap on how far
#: back ONE view's source is kept. Past this the oldest go and `trimmed` counts them, so a reader
#: can see the history is partial instead of concluding the view was only ever saved twice.
MAX_HISTORY = 40

#: ⛔ HOW MANY SCRIPTS MAY BE RUNNING IN THIS CONTAINER AT ONCE, and it is a REPORTED refusal
#: rather than a queue. Each run is a real subprocess with a ten-second wall clock; without this,
#: holding down refresh forks until the box gives up, and the tenant's ONE FastAPI process is what
#: gives up. A 429 that says so is honest; an unbounded fork is not.
MAX_CONCURRENT_RUNS = 4
_RUN_SLOTS = threading.BoundedSemaphore(MAX_CONCURRENT_RUNS)


def _now():
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def _all(rt):
    """`{id: record}` for one tenant. `{}` on any failure — an unreadable bucket must degrade to
    "this database has no script views", never to a 500 on the view rail."""
    try:
        found = rt.get(VIEWS_KEY) or {}
    except Exception:                                                        # noqa: BLE001
        return {}
    return found if isinstance(found, dict) else {}


def _database_ok(session, database):
    """Does this database EXIST, and may this caller read it? Answered by C1, never by a list.

    ⛔ `perm_scope.may_read` ALONE IS NOT ENOUGH and the reason is easy to miss: it answers True
    for an ADMIN on any key at all, including one no database answers to. So a create validated
    with `may_read` would let an administrator bind a view to a typo and leave an orphan nothing
    can ever run. `scoped_fields` is the cheap half of C1 (a `ut_*` definition, no rows) and it
    RAISES `UnknownTable`, which is exactly the question being asked.
    """
    import core.perm_scope as perm_scope
    try:
        perm_scope.scoped_fields(session.user, database, st=session.runtime)
    except perm_scope.UnknownTable:
        raise err(404, "no_database", f"there is no database '{database}' in this workspace")
    except perm_scope.Denied:
        raise err(403, "forbidden", f"your account may not read '{database}'")
    except perm_scope.Unresolvable as exc:
        # The database is real and cannot be served under this call's constraints. Standing rule
        # 1's second sentence: report the cause and the recommendation, never a bare refusal.
        raise err(409, "unresolvable", str(exc)) from None


def _clean_source(raw):
    source = str(raw or "")
    if not source.strip():
        raise err(400, "no_source", "a script view needs some code")
    if len(source.encode("utf-8", "replace")) > MAX_SOURCE_BYTES:
        raise err(413, "source_too_long",
                  f"a script view is at most {MAX_SOURCE_BYTES // 1024} KB of code")
    return source


def _row(rec, *, source=False):
    """One view as the list door reports it. ⚠ NO SOURCE unless asked: the rail lists names."""
    out = {"id": rec.get("id") or "", "database": rec.get("database") or "",
           "name": rec.get("name") or "", "author": rec.get("author") or "",
           "version": int(rec.get("version") or 1),
           "created": rec.get("created") or "", "updated": rec.get("updated") or "",
           "versions": len(rec.get("history") or []) + 1,
           "trimmed": int(rec.get("trimmed") or 0)}
    if source:
        out["source"] = rec.get("source") or ""
        out["history"] = [{"version": int(h.get("version") or 0), "author": h.get("author") or "",
                           "created": h.get("created") or "",
                           "bytes": len(str(h.get("source") or "").encode("utf-8", "replace"))}
                          for h in reversed(rec.get("history") or []) if isinstance(h, dict)]
    return out


def _limits():
    return {"maxSourceBytes": MAX_SOURCE_BYTES, "maxName": MAX_NAME,
            "maxHistory": MAX_HISTORY, "maxConcurrentRuns": MAX_CONCURRENT_RUNS,
            # ⭐ SAID OUT LOUD, because R3's "not capped" half is the one a reader assumes wrong.
            "maxViewsPerDatabase": None}


def _put(session, view_id, mutate):
    """Read-modify-write ONE view, synchronously — the client re-reads the rail immediately."""
    def _set(cur):
        cur = dict(cur or {})
        nxt = mutate(cur.get(view_id) if isinstance(cur.get(view_id), dict) else None)
        if nxt is None:
            cur.pop(view_id, None)
        else:
            cur[view_id] = nxt
        return cur

    session.runtime.update(VIEWS_KEY, _set, flush="sync")


def _mine_or_admin(session, rec):
    """Who may EDIT or DELETE a view: its author, or an administrator.

    ⚠ RUNNING IS A DIFFERENT QUESTION and deliberately wider — anybody who may read the database
    may run any view on it, under their OWN scope. That is the whole of "so we can see different
    versions or different things the AI code for us": a colleague's attempt is worth nothing if
    only its author can open it.
    """
    if session.admin or str(rec.get("author") or "") == session.uname:
        return
    raise err(403, "forbidden", "only the author or an administrator can change this script view")


# ── the routes ────────────────────────────────────────────────────────────────────────────────
@router.get("/script-views")
def list_script_views(database: str = "", session: Session = Depends(require_session)):
    """Every script view bound to ONE database, newest first. `database` is required."""
    key = str(database or "").strip()
    if not key:
        raise err(400, "no_database", "name the database whose script views you want")
    _database_ok(session, key)
    rows = [_row(rec) for rec in _all(session.runtime).values()
            if isinstance(rec, dict) and str(rec.get("database") or "") == key]
    rows.sort(key=lambda r: (r["created"], r["id"]), reverse=True)
    return {"database": key, "views": rows, "limits": _limits()}


@router.post("/script-views")
def create_script_view(body: dict = Body(default=None),
                       session: Session = Depends(require_session)):
    """Create one. ⛔ THE SOURCE IS CHECKED BEFORE IT IS STORED, not first at run time.

    A script view that cannot be run is a broken feature the reader discovers by pressing a button,
    and the agent that wrote it is long gone by then. `check_source` is pure and costs no process,
    so the refusal arrives while the author still has the code in front of them.
    """
    import secrets                                                           # noqa: PLC0415
    import core.script_sandbox as sandbox                                    # noqa: PLC0415

    body = body if isinstance(body, dict) else {}
    database = str(body.get("database") or "").strip()
    if not database:
        raise err(400, "no_database", "a script view is bound to one database")
    _database_ok(session, database)
    source = _clean_source(body.get("source"))
    refusal = sandbox.check_source(source)
    if refusal is not None:
        raise err(400, refusal.code, refusal.message)

    view_id = "sv_" + secrets.token_urlsafe(9)
    rec = {"id": view_id, "database": database,
           "name": " ".join(str(body.get("name") or "Script view").split())[:MAX_NAME],
           "source": source, "author": session.uname, "version": 1,
           "created": _now(), "updated": _now(), "history": [], "trimmed": 0}
    _put(session, view_id, lambda _prior: rec)
    fresh = _all(session.runtime).get(view_id)
    if not isinstance(fresh, dict):
        # The store took the write and did not record it. A 200 here would tell the author their
        # script was saved when it was not.
        raise err(503, "store_unavailable", "the script view was NOT created")
    return {"view": _row(fresh, source=True), "limits": _limits()}


@router.get("/script-views/{view_id}")
def get_script_view(view_id: str, session: Session = Depends(require_session)):
    """One view WITH its source and its edit history. Owner item 6's *"see the code"* half."""
    rec = _all(session.runtime).get(str(view_id))
    if not isinstance(rec, dict):
        raise err(404, "no_view", "there is no script view with that id")
    _database_ok(session, str(rec.get("database") or ""))
    return {"view": _row(rec, source=True), "limits": _limits()}


@router.put("/script-views/{view_id}")
def update_script_view(view_id: str, body: dict = Body(default=None),
                       session: Session = Depends(require_session)):
    """A NEW VERSION of one view's source. The prior source is KEPT, never replaced in place."""
    import core.script_sandbox as sandbox                                    # noqa: PLC0415

    view_id = str(view_id)
    rec = _all(session.runtime).get(view_id)
    if not isinstance(rec, dict):
        raise err(404, "no_view", "there is no script view with that id")
    _mine_or_admin(session, rec)
    body = body if isinstance(body, dict) else {}
    source = _clean_source(body.get("source"))
    refusal = sandbox.check_source(source)
    if refusal is not None:
        raise err(400, refusal.code, refusal.message)

    def _mutate(prior):
        prior = dict(prior or rec)
        history = list(prior.get("history") or [])
        history.append({"version": int(prior.get("version") or 1),
                        "source": prior.get("source") or "",
                        "author": prior.get("author") or "", "created": prior.get("updated") or ""})
        dropped = max(0, len(history) - MAX_HISTORY)
        prior["history"] = history[dropped:] if dropped else history
        prior["trimmed"] = int(prior.get("trimmed") or 0) + dropped
        prior["source"] = source
        prior["version"] = int(prior.get("version") or 1) + 1
        prior["updated"] = _now()
        if body.get("name"):
            prior["name"] = " ".join(str(body["name"]).split())[:MAX_NAME]
        return prior

    _put(session, view_id, _mutate)
    fresh = _all(session.runtime).get(view_id)
    if not isinstance(fresh, dict):
        raise err(503, "store_unavailable", "the new version was NOT saved")
    return {"view": _row(fresh, source=True), "limits": _limits()}


@router.delete("/script-views/{view_id}")
def delete_script_view(view_id: str, session: Session = Depends(require_session)):
    view_id = str(view_id)
    rec = _all(session.runtime).get(view_id)
    if not isinstance(rec, dict):
        raise err(404, "no_view", "there is no script view with that id")
    _mine_or_admin(session, rec)
    _put(session, view_id, lambda _prior: None)
    return {"deleted": view_id}


@router.post("/script-views/{view_id}/run")
def run_script_view(view_id: str, body: dict = Body(default=None),
                    session: Session = Depends(require_session)):
    """CONTRACT C3's run door: `{ok, spec | error, stdout, ms}`.

    ⛔ `spec` IS A DESCRIPTION THE CLIENT DRAWS. It is never HTML and never text the browser
    executes — the sandbox refuses a spec carrying an `html`, `script`, `src`, `href` or `on*` key
    before this function ever sees it, so a renderer cannot be talked into running something by a
    script that was itself perfectly well behaved.

    ⛔ AND IT IS PLAIN `def`, NOT `async def` — see this module's header. A ten-second subprocess
    wait on the event loop is a ten-second outage for the whole container.
    """
    import core.script_sandbox as sandbox                                    # noqa: PLC0415

    rec = _all(session.runtime).get(str(view_id))
    if not isinstance(rec, dict):
        raise err(404, "no_view", "there is no script view with that id")
    database = str(rec.get("database") or "")
    # A DRAFT run: the editor sends unsaved code so the author can try it before committing to it.
    # It is checked exactly as a stored one is, because "unsaved" is not a permission.
    draft = (body or {}).get("source") if isinstance(body, dict) else None
    source = _clean_source(draft) if draft else str(rec.get("source") or "")
    # ⚠ NO `_database_ok` CALL HERE. `run_view` asks C1 the identical question a line later, and
    # asking twice builds a registry topic's pool twice. The three refusals are translated below
    # instead, which is the same wall reached through the same door.

    if not _RUN_SLOTS.acquire(blocking=False):
        raise err(429, "busy",
                  f"{MAX_CONCURRENT_RUNS} script views are already running on this server. "
                  f"Try again in a moment")
    try:
        out = sandbox.run_view(session.user, database, source, st=session.runtime)
    finally:
        _RUN_SLOTS.release()

    # ⛔ C1'S THREE REFUSALS ARE HTTP STATUSES, NOT `ok:false`. "You may not read this database"
    # answered 200 would be a permission decision the client has to go looking for, and every
    # other door in this app answers 403 for it. Everything BELOW this line is a well-formed
    # request whose ANSWER is that the script did not produce a view — that is a 200 with
    # `ok:false`, the shape `routes_web_agent.test` already uses for the same reason.
    if out.get("code") == "unknown_table":
        raise err(404, "no_database", out.get("message") or f"there is no database '{database}'")
    if out.get("code") == "denied":
        raise err(403, "forbidden", out.get("message") or "your account may not read that database")
    if out.get("code") == "unresolvable":
        refusal = err(409, "unresolvable", out.get("message") or "these rows cannot be served")
        refusal.detail["error"]["limit"] = out.get("limit") or {}
        raise refusal

    answer = {"ok": bool(out.get("ok")), "spec": out.get("spec"),
              "stdout": out.get("stdout") or "", "truncated": bool(out.get("truncated")),
              "ms": int(out.get("ms") or 0), "code": out.get("code") or "",
              # ⭐ `caps` RIDES ON THE ANSWER (standing rule 1). On a POSIX host all three limits
              # were applied; on a Windows host the memory and CPU ones were not, and a screen
              # that claims an enforcement which did not happen is the failure the rule is about.
              "caps": out.get("caps") or {}}
    if not answer["ok"]:
        answer["error"] = out.get("message") or "the script view did not produce a view"
    return answer