api/routes_query.py

"""routes_query.py — WAVE 32 item 7 (ruling R1, contract C5): the Query module's doors.

    POST /api/v1/query/build   {question, scope}  -> {spec, explain, refused}   WRITES NOTHING
    POST /api/v1/query/save    {question, scope, spec} -> the saved view
    GET  /api/v1/query                            -> the AI-built views this caller may open
    DELETE /api/v1/query/{qid}                    -> forget one

**The AI turns a question about ONE granted database into a view SPEC. It may only build what the
backend already supports** (R1), and the wall on which database it may name is the EXISTING one.

⛔ THE FOUNDATION IS NOT `harness/semantic.py` — MEASURED, W32-T50
(`.claude/wiki/waves/wave32/proto/query-spike/README.md`). That layer is bound to 8 hand-authored
YAML topics over Odoo mirror tables: **0** are `ut_`-bound, and a `ut_` key answers
`ModelError: topic 'ut_…' has no store binding` from both `store_columns` and `store_query`. Its
own `save_view` writes `platform/data/store/views.json`, a chart workspace the product's grid
cannot read. What it lends this file is its POSTURE, quoted from its system prompt: *the model
never writes SQL and never invents a field name — it only picks registry keys.*

The real foundation is the view stack that already ships, and this module is `routes_templates.py`
with a model where the curated registry used to be:

  · the wall            `routes_templates._target_or_refuse`   (imported, never re-implemented)
  · the column contract `core.view_templates.columns_named` / `missing_columns`
  · the view skeleton   `core.view_templates._view`            (one definition of what a view is)
  · the validators      `aios_grid._clean_display` / `clean_filter_tree`
  · the writer          `core.table_store.make(ws).save_view(..., shared=False)`

⛔⛔ THE VALIDATORS DROP; THEY DO NOT REFUSE — and C5 requires a REFUSAL. `_clean_display` returns
None for an unknown mode, and a view with no display block IS A GRID; `clean_filter_tree` drops a
leaf naming an unknown column, and a view whose only condition was dropped SHOWS EVERY ROW under a
name promising a shortlist (`view_templates`' own scar, and `_seed_wave17`'s before it). So
membership is tested HERE, first and explicitly, and the cleaners run second as a belt. They are
the vocabulary ORACLE, never the refusal.

⛔ THE REFUSAL PATH IS A MODEL PROPERTY, so this module declares its OWN provider order rather than
inheriting `analyst.available_providers()`' cheap-first one. Measured over 4 arms x 9 questions
(`proto/query-spike/refusal-arms.json`): `cerebras/gpt-oss-120b` refuses 4/4 unanswerable questions
with a usable sentence; `groq/llama-3.3-70b` refuses **0/4** and instead emits a fully valid spec
answering a different question ("Revenue by Creator" against a database holding no revenue).
Changing refusal from an enum value to its own tool changed cerebras not at all and made groq
worse.
⚠ AND THE LADDER STILL FALLS THROUGH ON A 429, WHICH IS A COMPROMISE, NOT AN OVERSIGHT — stated
here because the first draft of this header claimed the opposite. A door that dies whenever one
free tier rate-limits is worse than one that answers with a weaker model behind the SAME validator
and the SAME human read-back. What "fail closed" means precisely: when the whole ladder is
unreachable the answer is a sentence and never a view. Which provider answered rides back in the
response so a wrong answer stays diagnosable (`_call_model`'s note).

⛔ AND A PROVIDER `400` CAN BE THE MODEL'S REFUSAL. Groq validates the model's tool arguments
against the tool schema server-side and answers `400 tool_use_failed` carrying the model's real
output in `failed_generation` — so a refusal that omits a `required` property arrives as an HTTP
error. `analyst._live_chat` calls `raise_for_status()` and discards that body, which is why this
file has its own small ladder: `required` is `["kind"]` ALONE (a refusal must be a legal argument
object), and a 400 body is READ before it is treated as transport.

⚠ WHAT NO VALIDATOR CAN SEE, stated so nobody thinks the gate covers it: a spec whose every column
is real, every op legal, kind in the vocabulary and target granted, which answers a DIFFERENT
QUESTION. That class is invisible to server-side validation. The mechanism against it is C5's own
shape — `build` returns `{spec, explain}` and WRITES NOTHING, `explain` is DERIVED from the
validated spec (never asked of the model, which would let it flatter its own output), and saving is
a second, explicit call. The person is the last validator, by design.
"""
import datetime as _dt
import hashlib
import json
import os
import re

from fastapi import APIRouter, Body, Depends

from deps import Session, err, require_session

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

#: The store key holding the REGISTRY of AI-built views, per tenant: {qid: {...}}. The view itself
#: is a real saved view in its target's own workspace — this is only the index, so listing the
#: Query module never has to read every granted table's workspace to find them. Same split
#: `core/alerts.py` draws between definitions and the inbox.
QUERY_KEY = 'query_views'

MAX_QUESTION = 500
MAX_VISIBLE = 24
MAX_REGISTRY = 200          #: a working list, not an archive (alerts' MAX_NOTIFICATIONS rule)

#: ⭐ THE KINDS THE AI MAY EMIT. Held here as a NAMED SET, and `verify_query.py` asserts it is a
#: subset of BOTH `aios_grid.DISPLAY_MODES` (the stored vocabulary) and the client's
#: `CREATABLE_MODES` (which means "has a mounted renderer" — `verify_icons::mode_parity` enforces
#: that both ways), AND that the excluded remainder is exactly `QUERY_EXCLUDED`. So a mode that
#: lands in `CREATABLE_MODES` later REDS the gate instead of being silently un-offered — the
#: `swipe`/`form` failure ("whole, correct and unreachable") in reverse.
QUERY_KINDS = ('grid', 'chart', 'calendar', 'kanban', 'timeseries', 'map', 'list')

#: Creatable modes the AI deliberately may NOT emit, each with the reason it cannot be derived
#: from a question about existing records. A reason per member, because "excluded" with no cause
#: is how a hold outlives its own justification.
QUERY_EXCLUDED = {
    'form': "a form is a door records come IN through and shows none — a question about existing "
            "records cannot describe one, and its spec carries emails and a public token",
    'catalog': "a catalog is a published artifact needing page authoring and product codes "
               "(MAX_CATALOG_CODES is spent in page order), not a shape of the current rows",
    'swipe': "a swipe deck needs an empty single-select plus a chosen option per direction — two "
             "product decisions a question does not contain",
}

#: Mode-required refs, and the field FAMILY each one demands. ⚠ STRICTER THAN `_clean_display` ON
#: PURPOSE. That function accepts any ref naming a real field and leaves the meaning to the client,
#: which "degrades a wrong-typed ref to that surface's default rather than to an error" — correct
#: for a person clicking a picker who SEES the result, wrong for a generated spec nobody has looked
#: at yet. Measured: a free model stacked a kanban on `category` (a text column) for a contract
#: with ZERO select columns, and every validator passed it.
_MODE_REFS = {
    'kanban': [('stackField', ('select',), True)],
    'calendar': [('dateField', ('date',), True)],
    'timeseries': [('dateField', ('date',), True)],
    'map': [('colorField', ('select',), False), ('sizeField', ('int', 'currency', 'pct'), False)],
}


def _now_iso():
    """UTC WITH the offset — a naive stamp is unsubtractable in a browser (D-18, alerts' rule)."""
    return _dt.datetime.now(_dt.timezone.utc).isoformat()


def _grid():
    import aios_grid
    return aios_grid


def _templates():
    import core.view_templates as view_templates
    return view_templates


def _target(session, scope):
    """`(fields, source)` for a database this session may open — or a refusal.

    ⛔ THE WALL IS `routes_templates._target_or_refuse`, IMPORTED AND NOT COPIED. It is the same
    predicate the nav uses, per topic (`ut_*` -> `user_tables.may_open`; a built-in -> the module
    grant `session.require`; anything else a 404 that is not a directory of what exists), and T51's
    instruction is explicit that the wall is re-used, never re-implemented beside itself. The
    private name is deliberate: an alias would mean editing a file no lane's fence covers.

    It returns field KEYS; this door also needs each field's TYPE (for the prompt and for
    `_MODE_REFS`), so the types are read from the same source the wall consulted — never from a
    list assembled here.
    """
    import routes_templates
    keys, source = routes_templates._target_or_refuse(session, scope)
    key = str(scope or '').strip()
    defs = []
    if key.startswith('ut_'):
        import core.user_tables as user_tables
        defn = user_tables.get(key, st=session.runtime) or {}
        by_key = {str(f.get('key')): f for f in (defn.get('fields') or []) if f.get('key')}
    else:
        g = _grid()
        raw = g.product_fields() if key == 'product_data' else g.FIELDS
        by_key = {str(f.get('key')): f for f in raw if f.get('key')}
    for k in keys:
        f = by_key.get(k) or {}
        defs.append({'key': k, 'label': str(f.get('label') or k)[:80],
                     'type': str(f.get('type') or 'text')})
    return defs, source


# ── the model ───────────────────────────────────────────────────────────────────────────────

#: This module's OWN provider order — cerebras first. See the header: the refusal is a model
#: property, and the ladder's cheap-first order is exactly wrong for a contract that includes one.
#: Override with `AIOS_QUERY_PROVIDER`. The credentials themselves stay in ONE place
#: (`analyst.PROVIDERS`), so a rotated key or a new model reaches this file too.
QUERY_PROVIDER_ORDER = ('cerebras', 'groq', 'openrouter')


def _providers():
    import harness.analyst as analyst
    pin = os.environ.get("AIOS_QUERY_PROVIDER")
    by_name = {p["name"]: p for p in analyst.PROVIDERS}
    order = [pin] if pin else list(QUERY_PROVIDER_ORDER)
    return [by_name[n] for n in order
            if n in by_name and os.environ.get(by_name[n]["env"])]


def _spec_schema(field_keys):
    """The tool schema the model fills. ⛔ `required` IS `["kind"]` ALONE — see the header: a
    refusal carries no name, and groq rejects a schema-violating refusal with a 400 that reads
    exactly like a transport failure."""
    ops = sorted(_grid().FILTER_OPS)
    col = {"type": "string", "enum": sorted(field_keys)}
    return {
        "type": "object",
        "properties": {
            "kind": {"type": "string", "enum": list(QUERY_KINDS) + ["refused"],
                     "description": "refused = this database cannot answer the question"},
            "name": {"type": "string", "description": "a short title for the view"},
            "refusal": {"type": "string",
                        "description": "when kind=refused: ONE plain sentence naming what is "
                                       "missing"},
            "visible": {"type": "array", "items": col},
            "filters": {"type": "array", "items": {
                "type": "object",
                "properties": {"colId": col, "op": {"type": "string", "enum": ops},
                               "value": {"type": "string"}},
                "required": ["colId", "op"]}},
            "filterConj": {"type": "string", "enum": ["and", "or"]},
            "sorts": {"type": "array", "items": {
                "type": "object",
                "properties": {"colId": col,
                               "dir": {"type": "string", "enum": ["asc", "desc"]}},
                "required": ["colId", "dir"]}},
            "groupBy": col,
            "stackField": col,
            "dateField": col,
            "colorField": col,
            "sizeField": col,
        },
        "required": ["kind"],
    }


def _system_prompt(fields):
    """The standing instructions. The vocabulary is DERIVED (the kinds set, `FILTER_OPS`, the
    target's own columns) — the model is never told about a key this server cannot honour."""
    cols = json.dumps([{"key": f["key"], "label": f["label"], "type": f["type"]}
                       for f in fields], separators=(",", ":"))
    kinds = ", ".join(QUERY_KINDS)
    ops = ", ".join(sorted(_grid().FILTER_OPS))
    selects = [f["key"] for f in fields if f["type"] in ('select', 'multiselect')] or ["(none)"]
    dates = [f["key"] for f in fields if f["type"] == 'date'] or ["(none)"]
    return f"""You turn a question about ONE database into a VIEW SPEC. You never write SQL and you
never invent a column or a kind of view.

THE DATABASE ({len(fields)} columns) — you may name ONLY these column keys:
{cols}

VIEW KINDS you may name, and nothing else: {kinds}
FILTER OPS you may name, and nothing else: {ops}
The only select-type columns (a kanban MUST stack on one of these): {", ".join(selects)}
The only date columns (a calendar or timeseries MUST use one of these): {", ".join(dates)}

RULES:
1. Emit ONE call to build_view. Every colId in visible/filters/sorts/groupBy and every field ref
   MUST be one of the keys above.
2. Pick the kind by what the question asks for: a list of records -> grid (or list); totals or
   counts by a category -> chart with groupBy; cards by a stage -> kanban (stackField REQUIRED,
   and it must be a select column); something over time -> calendar or timeseries (dateField
   REQUIRED, and it must be a date column); geography -> map.
3. `visible` is the columns the question is about, most important first — not every column.
4. Set kind="refused" and put ONE plain sentence in `refusal` when the question needs a column
   this database does not have, a number it does not store, or a picture that is not in the kind
   list. REFUSING IS A CORRECT ANSWER. A view that answers a DIFFERENT question than the one
   asked is the worst possible answer — worse than saying you cannot build it. When in doubt,
   refuse.
5. Every `value` is a string."""


def _call_model(question, fields, chat=None):
    """`(spec, error_sentence, provider)` — one bounded call down THIS module's ladder.

    `chat` is injectable so `verify_query.py` proves this door end to end with no key and no
    tokens (`analyst.ask`'s own posture, and the reason its loop is testable offline).

    ⚠ THE LADDER IS A DELIBERATE COMPROMISE, AND THE PROVIDER RIDES BACK BECAUSE OF IT. Only the
    FIRST provider is measured to refuse honestly (README section 4b); the ones below it answer
    unanswerable questions with plausible specs. Falling through on a 429 is still the right call —
    a Query module that dies whenever one free tier rate-limits is worse than one that answers with
    a weaker model behind the same validator and the same human read-back — but *which* model
    answered must be knowable, or a wrong answer is undiagnosable after the fact. Failing closed is
    what happens when the whole ladder is unreachable, not when the best rung is.
    """
    tools = [{"type": "function", "function": {
        "name": "build_view",
        "description": "Emit the view spec, or refuse.",
        "parameters": _spec_schema([f["key"] for f in fields])}}]
    messages = [{"role": "system", "content": _system_prompt(fields)},
                {"role": "user", "content": question}]
    if chat is not None:
        return chat(messages, tools), None, "injected"

    provs = _providers()
    if not provs:
        # No key at all: say so. An AI feature that silently does nothing is indistinguishable
        # from one that was never built ([[flag-shipped-without-its-writer]]).
        return None, ("the assistant is not configured on this deployment, so a view cannot be "
                      "built from a question yet"), None
    import requests
    last = None
    for p in provs:
        try:
            r = requests.post(p["url"], timeout=60,
                              headers={"Authorization": f"Bearer {os.environ[p['env']]}"},
                              json={"model": p["model"], "messages": messages, "tools": tools,
                                    "tool_choice": "required", "temperature": 0.1,
                                    "max_tokens": 1200})
        except Exception as e:                       # network blip — try the next provider
            last = f"{p['name']}: {type(e).__name__}"
            continue
        if r.status_code == 400:
            # ⛔ THE MODEL'S ANSWER CAN BE IN HERE. Groq validates tool arguments against the
            # schema and returns the generation it rejected. Reading it turns a "transport
            # failure" back into the refusal it actually was.
            spec = _spec_from_400(r.text)
            if spec is not None:
                return spec, None, p["name"]
            last = f"{p['name']}: 400"
            continue
        if r.status_code != 200:
            last = f"{p['name']}: HTTP {r.status_code}"
            continue                                 # 429 included: the NEXT provider, not a sleep
        try:
            msg = r.json()["choices"][0]["message"]
            calls = msg.get("tool_calls") or []
            if not calls:
                last = f"{p['name']}: no tool call"
                continue
            return json.loads(calls[0]["function"].get("arguments") or "{}"), None, p["name"]
        except Exception as e:
            last = f"{p['name']}: unreadable answer ({type(e).__name__})"
            continue
    # ⛔ FAIL CLOSED. The first provider in the order is the one measured to refuse honestly; if it
    # is rate-limited and the rest are too, the answer is a sentence, never a fabricated view.
    return None, ("the assistant could not be reached just now (" + (last or "no provider") +
                  ") — try the question again in a moment"), None


_FAILED_GEN = re.compile(r'<function=build_view>(\{.*?\})\s*</function>', re.S)


def _spec_from_400(body):
    """The spec out of a `tool_use_failed` body's `failed_generation`, or None."""
    try:
        err_obj = (json.loads(body) or {}).get("error") or {}
    except Exception:
        return None
    gen = str(err_obj.get("failed_generation") or "")
    if not gen:
        return None
    m = _FAILED_GEN.search(gen)
    raw = m.group(1) if m else gen.strip()
    try:
        spec = json.loads(raw)
    except Exception:
        return None
    return spec if isinstance(spec, dict) else None


# ── validation: the REFUSAL this contract promises ───────────────────────────────────────────

def _validate(spec, fields):
    """`(clean, refusal, code)` — exactly one of `clean` / `refusal` is set.

    Membership FIRST and explicitly (the header's whole subject), then the product's own cleaners
    as a belt. A refusal is a SENTENCE, because C5's user-facing promise is a plain sentence and
    never a broken view.
    """
    g = _grid()
    if not isinstance(spec, dict):
        return None, "the assistant did not answer with a view", "no_spec"
    by_key = {f['key']: f['type'] for f in fields}
    keys = set(by_key)
    kind = spec.get('kind')

    if kind == 'refused':
        return None, (str(spec.get('refusal') or '').strip()
                      or "this database cannot answer that question"), "model_refused"
    if kind in QUERY_EXCLUDED:
        return None, (f"a {kind} view cannot be built from a question — "
                      + QUERY_EXCLUDED[kind]), "unsupported_kind"
    if kind not in QUERY_KINDS:
        # ⛔ NOT `_clean_display`'s job: it would DROP this and hand back a grid under a name
        # promising something else.
        return None, (f"{kind!r} is not a kind of view this product can build"
                      if kind else "the assistant did not name a kind of view"), "unsupported_kind"

    named = set(spec.get('visible') or ())
    for s in spec.get('sorts') or ():
        if isinstance(s, dict) and s.get('colId'):
            named.add(str(s['colId']))
    for f in spec.get('filters') or ():
        if isinstance(f, dict) and f.get('colId'):
            named.add(str(f['colId']))
    if spec.get('groupBy'):
        named.add(str(spec['groupBy']))
    for ref in ('stackField', 'dateField', 'colorField', 'sizeField'):
        if spec.get(ref):
            named.add(str(spec[ref]))
    missing = sorted(named - keys)
    if missing:
        # NAMED, not counted (`routes_templates`' rule): the column keys are what tell the reader
        # this is the wrong database for this question.
        return None, ("this database does not have the columns that answer needs: "
                      + ", ".join(missing)), "unknown_columns"

    visible = [k for k in (spec.get('visible') or ()) if k in keys][:MAX_VISIBLE]
    if not visible:
        # An empty projection is not a view. `_default_view_config` would fill one in, which is
        # how a question about two columns becomes a 285-column wall.
        return None, ("that question did not name anything to show — try naming the columns you "
                      "want to see"), "no_columns"

    # ⛔ D-137, GUARDED WHERE IT REACHES THIS FEATURE. "The chart cannot key a SET-LIKE or CHECKBOX
    # column the way the grid does" — a set row lands in several buckets, so a sum means something
    # else, and a checkbox reads "(blank)". The defect lives in the CLIENT's `chartData` bucket
    # keying, which is in no lane's fence this wave (booked, with that reason). What IS in this
    # lane's power is refusing to GENERATE the broken case: the AI must only build what the backend
    # and the renderer already support, which is R1's own sentence. Delete this guard when D-137
    # closes — the refusal names the debt so the reader knows it is a limit, not a rule.
    if kind == 'chart' and by_key.get(spec.get('groupBy')) in ('multiselect', 'checkbox'):
        return None, (f"a chart cannot group by {spec.get('groupBy')!r} yet — a column that holds "
                      f"several values at once, or a tick box, does not bucket the way a chart "
                      f"needs. Try a table grouped by it instead"), "chart_key_unsupported"

    for ref, families, required in _MODE_REFS.get(kind, ()):
        val = spec.get(ref)
        if val and by_key.get(val) not in families:
            return None, (f"a {kind} view needs {ref[:-5]} to be a "
                          f"{' or '.join(families)} column, and {val!r} is a "
                          f"{by_key.get(val)} column"), "wrong_ref_type"
        if required and not val:
            return None, (f"a {kind} view needs a {' or '.join(families)} column and this "
                          f"database has none that fits"), "missing_ref"

    raw_filters = [f for f in (spec.get('filters') or ()) if isinstance(f, dict)]
    filters = g.clean_filter_tree(raw_filters, keys)
    if len(filters) != len(raw_filters):
        # The belt caught something membership did not — an unknown op. Refuse rather than save a
        # view whose condition was silently dropped (the widening this file exists to prevent).
        return None, ("part of that filter is not something this product can express — try "
                      "asking it more simply"), "filter_dropped"

    display = {'mode': kind}
    for ref, _families, _req in _MODE_REFS.get(kind, ()):
        if spec.get(ref):
            display[ref] = spec[ref]
    cleaned_display = g._clean_display(display, keys) if kind != 'grid' else None
    if kind != 'grid' and not cleaned_display:
        return None, (f"this product could not build a {kind} view from that question"
                      ), "display_dropped"

    clean = {
        'kind': kind,
        'name': ' '.join(str(spec.get('name') or 'Query').split())[:60] or 'Query',
        'visible': visible,
        'filters': filters,
        'filterConj': 'or' if spec.get('filterConj') == 'or' else 'and',
        'sorts': [{'colId': s['colId'], 'dir': 'desc' if s.get('dir') == 'desc' else 'asc'}
                  for s in (spec.get('sorts') or ()) if isinstance(s, dict)
                  and s.get('colId') in keys][:3],
        'groupBy': spec['groupBy'] if spec.get('groupBy') in keys else None,
        'display': cleaned_display,
    }
    return clean, None, None


def _explain(clean, fields):
    """A plain-language read-back of what the view WILL do — DERIVED from the validated spec.

    ⛔ NEVER ASKED OF THE MODEL. A model describing its own output describes what it MEANT, which
    is precisely the failure this sentence exists to expose: a spec whose every column is real and
    which answers a different question is invisible to validation, so the only thing that catches
    it is a person reading an honest description of the spec that was actually accepted.
    """
    label = {f['key']: f['label'] for f in fields}
    ops = {'eq': 'is', 'neq': 'is not', 'gt': 'is over', 'gte': 'is at least',
           'lt': 'is under', 'lte': 'is at most', 'contains': 'contains',
           'doesNotContain': 'does not contain', 'isEmpty': 'is empty',
           'isNotEmpty': 'is not empty', 'between': 'is between', 'within': 'is within',
           'topN': 'is in the top', 'bottomN': 'is in the bottom'}
    kinds = {'grid': 'a table', 'list': 'a list', 'chart': 'a chart', 'kanban': 'a board',
             'calendar': 'a calendar', 'timeseries': 'a time series', 'map': 'a map'}
    parts = [f"{kinds.get(clean['kind'], clean['kind'])} of "
             + ", ".join(label.get(k, k) for k in clean['visible'][:6])
             + (f" and {len(clean['visible']) - 6} more columns"
                if len(clean['visible']) > 6 else "")]
    if clean['filters']:
        joiner = " or " if clean['filterConj'] == 'or' else " and "
        parts.append("showing only records where " + joiner.join(
            f"{label.get(f['colId'], f['colId'])} {ops.get(f['op'], f['op'])}"
            + (f" {f['value']}" if f.get('value') else "")
            for f in clean['filters'] if isinstance(f, dict) and f.get('colId')))
    else:
        parts.append("showing every record")
    if clean['groupBy']:
        parts.append(f"grouped by {label.get(clean['groupBy'], clean['groupBy'])}")
    if clean['sorts']:
        s = clean['sorts'][0]
        parts.append(f"sorted by {label.get(s['colId'], s['colId'])}"
                     + (" (highest first)" if s['dir'] == 'desc' else " (lowest first)"))
    for ref, word in (('stackField', 'in columns by'), ('dateField', 'placed in time by'),
                      ('colorField', 'coloured by'), ('sizeField', 'sized by')):
        val = (clean.get('display') or {}).get(ref)
        if val:
            parts.append(f"{word} {label.get(val, val)}")
    return ", ".join(parts) + "."


# ── the registry ────────────────────────────────────────────────────────────────────────────

def _qid(scope, question):
    """PINNED per (database, question), the `view_templates` discipline: asking the same thing
    twice UPDATES one view instead of minting "Query 2" — `save_view` de-duplicates NAMES by
    appending a number, so an unpinned id fills the store with numbered near-duplicates."""
    h = hashlib.sha1(f"{scope}|{' '.join(str(question).split()).lower()}".encode("utf-8"))
    return "q_" + h.hexdigest()[:12]


def _registry(session):
    try:
        recs = session.runtime.get(QUERY_KEY) or {}
    except Exception:
        return {}
    return recs if isinstance(recs, dict) else {}


def _mine(session, rec):
    return session.admin or str(rec.get('createdBy') or '') == session.uname


#: The BUILT-IN topics `routes_templates._target_or_refuse` accepts beside any `ut_*` key. ⛔ THE
#: CLIENT'S PICKER MUST MIRROR THIS, and `GET /query` publishes it (`builtins`) rather than leaving
#: the page to re-derive it: the nav hands `QueryPage` every granted entry, which is a WIDER set
#: than this door accepts, so a picker filtered on "not a surface" alone would offer a database the
#: build door 404s. `view_templates`' own rule — "a picker that offered a template the apply door
#: would then refuse is a control that lies" — and it is the same predicate twice, published once.
BUILTIN_SCOPES = ('customer_data', 'product_data')


@router.get("/query")
def list_queries(session: Session = Depends(require_session)):
    """The AI-built views this caller may open: their own, still-granted, still-existing.

    ⚠ PRUNED ON READ against the CURRENT wall, never merely on write — a grant can be withdrawn
    and a database deleted after a view was saved, and the Query module is chrome: it may render
    nothing the server would not grant today (`nav.ts`'s chrome law).
    """
    import routes_templates
    out = []
    for qid, rec in sorted(_registry(session).items(),
                           key=lambda kv: str((kv[1] or {}).get('createdAt') or ''),
                           reverse=True):
        if not isinstance(rec, dict) or not _mine(session, rec):
            continue
        try:
            routes_templates._target_or_refuse(session, rec.get('scope'))
        except Exception:
            continue                                  # no longer granted, or gone
        out.append({"id": qid, "scope": rec.get('scope'), "viewId": rec.get('viewId'),
                    "name": rec.get('name'), "kind": rec.get('kind'),
                    "question": rec.get('question'), "explain": rec.get('explain'),
                    "createdAt": rec.get('createdAt')})
    return {"views": out, "kinds": list(QUERY_KINDS), "builtins": list(BUILTIN_SCOPES)}


@router.post("/query/build")
def build_query(body: dict = Body(default=None),
                session: Session = Depends(require_session)):
    """A question + a granted database -> `{spec, explain, refused}`. **WRITES NOTHING.**

    A refusal is a 200 carrying a sentence, not a 5xx: "this database cannot answer that" is a
    successful answer to the person asking, and an error envelope would paint the error page R1
    forbids. The 4xx cases are the ones that ARE the caller's fault — an unnamed database, an
    over-long question, a database this session may not open.

    ⛔ THE MODEL SEAM IS **NOT** ON THIS SIGNATURE, AND THAT IS A CORRECTION. The first draft took
    `chat=None` as a parameter of the ROUTE — and FastAPI reads an un-annotated defaulted argument
    as a **QUERY PARAMETER**, so `chat` appeared in `openapi()["paths"]` (measured) and
    `POST /query/build?chat=x` would have reached `_call_model(..., chat="x")` and 500'd on
    `"x"(messages, tools)`. The gate never saw it because the gate calls the handler directly —
    which is exactly the shape of a test seam that becomes an input. The seam now lives on the
    module-level `_build` below: the gate calls that, and the wire cannot.
    """
    body = body if isinstance(body, dict) else {}
    return _build(body.get("question"), body.get("scope"), session)


def _build(question, scope, session, chat=None):
    """`build_query`'s body, with the model injectable — see that route's note on why the seam is
    here and not on the signature FastAPI reads."""
    question = ' '.join(str(question or "").split())
    if not question:
        raise err(400, "bad_request", "no question was asked")
    if len(question) > MAX_QUESTION:
        raise err(400, "question_too_long",
                  f"a question must be under {MAX_QUESTION} characters")
    fields, _source = _target(session, scope)
    if not fields:
        raise err(400, "no_columns", "that database has no columns to build a view from")

    spec, transport, provider = _call_model(question, fields, chat=chat)
    if transport:
        return {"spec": None, "explain": None, "refused": transport, "provider": provider}
    clean, refusal, code = _validate(spec, fields)
    if refusal:
        return {"spec": None, "explain": None, "refused": refusal, "reason": code,
                "provider": provider}
    return {"spec": clean, "explain": _explain(clean, fields),
            "refused": None, "scope": scope,
            "id": _qid(scope, question), "provider": provider}


@router.post("/query/save")
def save_query(body: dict = Body(default=None), session: Session = Depends(require_session)):
    """Save an accepted spec as the CALLING USER's own view, and index it under Query.

    ⛔ THE SPEC IS RE-VALIDATED HERE, against the target's contract as it is NOW. The build door
    wrote nothing, so this is the first write and it is the only place authorisation and validity
    have to hold together — and a client is not a wall. An unsupported kind or an unknown column
    is a **400 with a named code** on this door (the caller sent it), where the same fact is a
    plain sentence on `build` (the model produced it).
    """
    body = body if isinstance(body, dict) else {}
    scope = str(body.get("scope") or '').strip()
    question = ' '.join(str(body.get("question") or "").split())[:MAX_QUESTION]
    fields, _source = _target(session, scope)
    clean, refusal, code = _validate(body.get("spec"), fields)
    if refusal:
        raise err(400, code or "bad_spec", refusal)
    if not session.runtime.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable — nothing was saved")

    vt = _templates()
    ws_key = vt.workspace_key(scope)
    if not ws_key:
        raise err(404, "unknown_table", "that database does not exist")
    qid = _qid(scope, question) if question else _qid(scope, clean['name'])

    # ⛔ THE SKELETON IS `view_templates._view`, NOT A SECOND ONE. "A view assembled from a
    # DIFFERENT skeleton is a second definition of what a view is — and the two drift on the day
    # one of them gains a key." It also puts the mode at `config.display.mode`, which is the key
    # the grid actually reads (`config.displayMode` was the first draft, and nothing reads it).
    payload = vt._view(qid, clean['name'], f"Built by the assistant from: {question}"[:300],
                       clean['visible'], mode=clean['kind'],
                       sorts=clean['sorts'], filters=clean['filters'],
                       group_by=clean['groupBy'], conj=clean['filterConj'])
    if clean.get('display'):
        payload['config']['display'] = dict(clean['display'])
    payload['createdBy'] = session.uname

    # ⛔ THE CAP REFUSES; IT DOES NOT EVICT — and the first draft evicted. Dropping the oldest entry
    # would leave its saved VIEW behind in the table's own workspace: "stranded in the target's view
    # list with no way back to the question that made it", which is the thing `delete_query` below
    # exists to prevent. And an eviction nobody was told about is W30/R6's second sentence broken —
    # a limit that cannot be removed must be REPORTED with its cause. So the 201st is a refusal
    # naming the number and the way out, checked BEFORE the view is written.
    existing = _registry(session)
    if qid not in existing and len(existing) >= MAX_REGISTRY:
        raise err(400, "query_limit",
                  f"you already have {len(existing)} saved questions, which is the limit "
                  f"({MAX_REGISTRY}). Delete one you no longer need and save this again.")

    import core.table_store as table_store
    ops = table_store.make(ws_key, st=session.runtime)
    saved = ops.save_view(session.uname, payload, shared=False)

    entry = {"scope": scope, "viewId": qid, "name": saved.get('name') or clean['name'],
             "kind": clean['kind'], "question": question,
             "explain": _explain(clean, fields), "createdBy": session.uname,
             "createdAt": _now_iso()}

    def _up(data):
        data = data if isinstance(data, dict) else {}
        data[qid] = entry
        return data

    session.runtime.update(QUERY_KEY, _up, flush='async')
    return {"id": qid, **entry}


@router.delete("/query/{qid}")
def delete_query(qid: str, session: Session = Depends(require_session)):
    """Forget an AI-built view — the index entry AND the view itself.

    Leaving the view behind would strand it in the target's view list with no way back to the
    question that made it ([[a-record-can-outlive-its-subject]]).
    """
    rec = _registry(session).get(str(qid))
    if not isinstance(rec, dict) or not _mine(session, rec):
        raise err(404, "unknown_query", "that view does not exist")
    _target(session, rec.get('scope'))                # the wall, again, on the write path
    vt = _templates()
    ws_key = vt.workspace_key(rec.get('scope'))
    if ws_key and session.runtime.available():
        import core.table_store as table_store
        table_store.make(ws_key, st=session.runtime).delete_view(
            session.uname, rec.get('viewId') or str(qid))

    def _up(data):
        data = data if isinstance(data, dict) else {}
        data.pop(str(qid), None)
        return data

    session.runtime.update(QUERY_KEY, _up, flush='async')
    return {"deleted": str(qid)}