diff --git "a/api/routes_query.py" "b/api/routes_query.py" --- "a/api/routes_query.py" +++ "b/api/routes_query.py" @@ -1,725 +1,1261 @@ -"""Assistant chat and Query-owned virtual artefacts (W33-T72 / C8). - -Query reads only the detached, permission-filtered snapshot from -``deps.assistant_read_scope``. It owns a separate per-user namespace for threads, messages, -citations and virtual views; it never opens or mutates a source workspace. -""" -import copy -import datetime as _dt -import hashlib -import json -import os -import re -import uuid - -from fastapi import APIRouter, Body, Depends - -from deps import (Session, assistant_read_scope, assistant_source_status, err, - require_session) - -router = APIRouter(prefix="/api/v1") - -MAX_QUESTION = 500 -MAX_VISIBLE = 24 -MAX_ARTIFACTS = 200 -MODEL_AUTO = "auto" -QUERY_PROVIDER_ORDER = ("cerebras", "groq", "openrouter") -QUERY_KINDS = ("grid", "chart", "calendar", "kanban", "timeseries", "map", "list") -QUERY_WORKSPACE_EVENTS = {"view_create", "view_upsert", "view_delete"} -QUERY_EXCLUDED = { - "form": "Forms collect new data and are not a read-only Query artefact.", - "catalog": "Catalog is a source-native presentation that Query cannot safely mutate.", - "swipe": "Swipe is an interactive source-native presentation, not an Assistant output.", -} -_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(): - return _dt.datetime.now(_dt.timezone.utc).isoformat() - - -def _grid(): - import aios_grid - return aios_grid - - -def _safe(value): - """Detach values before they enter a durable Query object or an API reply.""" - return json.loads(json.dumps(value, default=str)) - - -def _namespace_key(session): - """A tenant runtime has one isolated key for each user's Query-owned objects.""" - principal = f"{session.tenant}:{session.uname}".encode("utf-8") - return "query_user_" + hashlib.sha256(principal).hexdigest()[:24] - - -def _blank_state(): - return {"version": 1, "threads": {}, "messages": {}, "citations": {}, "views": {}} - - -def _state(session): - try: - raw = session.runtime.get(_namespace_key(session)) or {} - except Exception: - raw = {} - if not isinstance(raw, dict): - return _blank_state() - out = _blank_state() - for key in out: - if key == "version": - continue - if isinstance(raw.get(key), dict): - out[key] = copy.deepcopy(raw[key]) - return out - - -def _new_id(prefix): - return f"{prefix}_{uuid.uuid4().hex[:16]}" - - -def model_choices(): - """Choices are stable even when one is not configured, so explicit means explicit.""" - return [MODEL_AUTO, *QUERY_PROVIDER_ORDER] - - -def _providers(model=MODEL_AUTO): - """Auto returns the permitted ladder; an explicit choice returns at most one provider.""" - import harness.analyst as analyst - - requested = str(model or MODEL_AUTO).strip().lower() - by_name = {p["name"]: p for p in analyst.PROVIDERS} - names = list(QUERY_PROVIDER_ORDER) if requested == MODEL_AUTO else [requested] - return [by_name[name] for name in names - if name in by_name and os.environ.get(by_name[name]["env"])] - - -def _spec_schema(field_keys): - col = {"type": "string", "enum": sorted(field_keys)} - return { - "type": "object", - "properties": { - "kind": {"type": "string", "enum": [*QUERY_KINDS, "refused"]}, - "name": {"type": "string"}, - "refusal": {"type": "string"}, - "visible": {"type": "array", "items": col}, - # ⭐⭐ 2026-08-15 (owner: *"if you can build the view, the AI assistant should also be - # able to do it by using our tools on the backend"*). `rhs` and `important` were the - # only two things a person could express through `view_upsert` and this tool could not. - # - # ⛔ WITHOUT `rhs` THE ASSISTANT CANNOT STATE AN ERROR-CATCHER AT ALL — the one shape - # item 8a named by hand (*"price and COGS not matching"*). Every one of the 20 - # `FILTER_OPS` was already reachable, because they all read `value`; comparing a column - # against ANOTHER COLUMN is a different member (`aios_grid._clean_rhs`, CG-9) and it was - # simply absent here, so the model had no way to ask for it and no way to be told why. - # `kind` is `field` ONLY: `measure` needs a window and `stat` needs the population - # vocabulary, neither of which this snapshot-shaped reader carries — offering them - # would be a control that lies, which is the same rule the source chips follow. - "filters": {"type": "array", "items": {"type": "object", "properties": { - "colId": col, "op": {"type": "string", "enum": sorted(_grid().FILTER_OPS)}, - "value": {"type": "string"}, - "rhs": {"type": "object", "properties": { - "kind": {"type": "string", "enum": ["field"]}, "colId": col, - }, "required": ["kind", "colId"]}, - }, "required": ["colId", "op"]}}, - # A personal legibility mark, exactly as `grid_events.view_upsert` treats it (wave 32 - # R5/C4) — not a lock, and no second permission wall. - "important": {"type": "boolean"}, - "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, - "aggregation": {"type": "object", "properties": { - "op": {"type": "string", "enum": ["count", "sum", "avg", "min", "max"]}, - "field": col, - }, "required": ["op"]}, - "stackField": col, - "dateField": col, - "colorField": col, - "sizeField": col, - }, - "required": ["kind"], - } - - -def _system_prompt(snapshot): - fields = snapshot["fields"] - cols = json.dumps([{"key": field["key"], "label": field.get("label", field["key"]), - "type": field.get("type", "text")} for field in fields], separators=(",", ":")) - return f"""You turn a question about exactly one permitted database into a virtual view. -Never write SQL, invent fields, or name another database. - -DATABASE: {snapshot['database']} -SNAPSHOT VERSION: {json.dumps(snapshot['source_version'], default=str)} -PERMISSION-FILTERED RECORD COUNT: {len(snapshot['records'])} -FIELDS: {cols} -VIEW KINDS: {", ".join(QUERY_KINDS)} - -Choose visible fields, supported filters and an optional aggregation. Use count for record counts; -sum, avg, min and max require one numeric field. The server computes and cites every number from -this exact snapshot. Return kind=refused with one plain sentence if the database cannot answer. - -To compare one column against ANOTHER column rather than a typed value, give the filter an rhs of -{{"kind":"field","colId":""}} and omit value — that is how you express questions like -"priced below what it costs us". Both columns must be in the FIELDS list above. -Set important=true when the view is one somebody should be chased about: an error, a mismatch, or -money at risk. Leave it out otherwise.""" - - -_FAILED_GEN = re.compile(r"(\{.*?\})\s*", re.S) - - -def _spec_from_400(body): - try: - value = ((json.loads(body) or {}).get("error") or {}).get("failed_generation") or "" - raw = (_FAILED_GEN.search(str(value)) or [str(value).strip()])[1] - result = json.loads(raw) - return result if isinstance(result, dict) else None - except Exception: - return None - - -def _call_model(question, snapshot, model=MODEL_AUTO, chat=None): - """Return ``(spec, sentence, provider, reason)`` without any source-data fallback.""" - requested = str(model or MODEL_AUTO).strip().lower() - if requested not in model_choices(): - return None, f"the selected model ({requested or model}) is unavailable", None, "model_unavailable" - tools = [{"type": "function", "function": { - "name": "build_view", "description": "Emit a virtual-view spec or a refusal.", - "parameters": _spec_schema([field["key"] for field in snapshot["fields"]]), - }}] - messages = [{"role": "system", "content": _system_prompt(snapshot)}, - {"role": "user", "content": question}] - if chat is not None: - provider = requested if requested != MODEL_AUTO else "injected" - return chat(messages, tools), None, provider, None - - providers = _providers(requested) - if not providers: - sentence = (f"the selected model ({requested}) is unavailable" if requested != MODEL_AUTO - else "the assistant is not configured on this deployment") - return None, sentence, None, "model_unavailable" if requested != MODEL_AUTO else None - - import requests - last = None - for provider in providers: - try: - response = requests.post( - provider["url"], timeout=60, - headers={"Authorization": f"Bearer {os.environ[provider['env']]}"}, - json={"model": provider["model"], "messages": messages, "tools": tools, - "tool_choice": "required", "temperature": 0.1, "max_tokens": 1200}, - ) - except Exception as exc: - last = f"{provider['name']}: {type(exc).__name__}" - continue - if response.status_code == 400: - spec = _spec_from_400(response.text) - if spec is not None: - return spec, None, provider["name"], None - last = f"{provider['name']}: 400" - continue - if response.status_code != 200: - last = f"{provider['name']}: HTTP {response.status_code}" - continue - try: - call = (response.json()["choices"][0]["message"].get("tool_calls") or [])[0] - spec = json.loads(call["function"].get("arguments") or "{}") - return spec, None, provider["name"], None - except Exception as exc: - last = f"{provider['name']}: unreadable answer ({type(exc).__name__})" - if requested != MODEL_AUTO: - return None, f"the selected model ({requested}) is unavailable", None, "model_unavailable" - return None, "the assistant could not be reached just now (" + (last or "no provider") + ")", None, None - - -def _validate(spec, fields): - """Return a cleaned, source-independent virtual-view config or a named refusal.""" - if not isinstance(spec, dict): - return None, "the assistant did not answer with a view", "no_spec" - by_key = {str(field.get("key")): str(field.get("type") or "text") for field in fields} - keys = set(by_key) - kind = spec.get("kind") - if kind == "refused": - return None, str(spec.get("refusal") or "this database cannot answer that question"), "model_refused" - if kind in QUERY_EXCLUDED or kind not in QUERY_KINDS: - return None, "that kind of view cannot be built from this question", "unsupported_kind" - - named = set(spec.get("visible") or ()) - for item in spec.get("filters") or (): - if isinstance(item, dict) and item.get("colId"): - named.add(str(item["colId"])) - # ⛔ THE RIGHT-HAND COLUMN IS A COLUMN AND MUST FACE THE SAME `missing` CHECK. Collecting - # only the left side is what makes D-229 possible one layer down: `clean_filter_tree` does - # NOT drop a leaf whose field-rhs names a column that does not exist — it keeps the leaf, - # strips the `rhs`, blanks the value, and `filter_sql.is_rule_active` then reports the rule - # INACTIVE. An inactive rule narrows nothing, so "margin under 10%" would come back as a - # view listing the ENTIRE catalogue under an error-catcher's name, with nothing red. - # Naming it here turns that into the ordinary "this database does not have: X" refusal. - rhs = item.get("rhs") if isinstance(item, dict) else None - if isinstance(rhs, dict) and rhs.get("colId"): - named.add(str(rhs["colId"])) - for item in spec.get("sorts") or (): - if isinstance(item, dict) and item.get("colId"): - named.add(str(item["colId"])) - for key in ("groupBy", "stackField", "dateField", "colorField", "sizeField"): - if spec.get(key): - named.add(str(spec[key])) - raw_aggregation = spec.get("aggregation") or {"op": "count"} - if isinstance(raw_aggregation, dict) and raw_aggregation.get("field"): - named.add(str(raw_aggregation["field"])) - missing = sorted(named - keys) - if missing: - return None, "this database does not have: " + ", ".join(missing), "unknown_columns" - - visible = [str(key) for key in (spec.get("visible") or ()) if str(key) in keys][:MAX_VISIBLE] - if not visible: - return None, "that question did not name any fields to show", "no_columns" - raw_filters = [item for item in (spec.get("filters") or ()) if isinstance(item, dict)] - filters = _grid().clean_filter_tree(raw_filters, keys) - if len(filters) != len(raw_filters): - return None, "part of that filter is unsupported", "filter_dropped" - # ⛔⛔ A SECOND, NARROWER CHECK, AND THE LENGTH CHECK ABOVE CANNOT DO ITS JOB (D-229). - # A dropped leaf changes the COUNT; a stripped `rhs` does not — the leaf survives, so - # `len(filters) == len(raw_filters)` and the refusal above never fires. The failure is - # therefore silent in exactly the direction that matters: the condition stops narrowing and - # the view answers with every record. Assert the member survived, per leaf. - # ⚠ The `named` pass above already refuses an rhs naming a column this database lacks, so - # reaching here means something ELSE stripped it (a type the comparand cannot take, a future - # `_clean_rhs` rule). Both doors, because the two catch different causes and the cost of - # missing this one is a wrong answer that looks right. - for sent, kept in zip(raw_filters, filters): - if sent.get("rhs") and not kept.get("rhs"): - return None, "that column cannot be compared against another column", "rhs_dropped" - - aggregation = raw_aggregation if isinstance(raw_aggregation, dict) else {} - op = str(aggregation.get("op") or "").lower() - field = aggregation.get("field") - if op not in {"count", "sum", "avg", "min", "max"}: - return None, "the assistant gave an unsupported aggregation", "bad_aggregation" - if op == "count": - field = None - elif field not in keys or by_key.get(field) not in {"int", "currency", "pct"}: - return None, "that aggregation needs one visible numeric field", "bad_aggregation" - - display = {"mode": kind} - for ref, families, required in _MODE_REFS.get(kind, ()): - value = spec.get(ref) - if required and not value: - return None, f"a {kind} view needs {ref}", "missing_ref" - if value and by_key.get(value) not in families: - return None, f"{ref} has the wrong field type", "wrong_ref_type" - if value: - display[ref] = value - cleaned_display = _grid()._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", "display_dropped" - - return { - "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": item["colId"], "dir": "desc" if item.get("dir") == "desc" else "asc"} - for item in (spec.get("sorts") or []) if isinstance(item, dict) - and item.get("colId") in keys][:3], - "groupBy": spec.get("groupBy") if spec.get("groupBy") in keys else None, - "aggregation": {"op": op, "field": field}, - "display": cleaned_display, - # ⚠ `is True`, not truthy, and UNCONDITIONAL — the same two rules `grid_events.view_upsert` - # follows for this key. `is True` so a model emitting the string "false" does not mark a - # view; unconditional so the mark is REMOVABLE rather than a flag that can be set and never - # cleared (a key written only when present leaves a stored `true` alive forever). - "important": spec.get("important") is True, - }, None, None - - -def _explain(view, fields): - label = {field["key"]: str(field.get("label") or field["key"]) for field in fields} - visible = ", ".join(label.get(key, key) for key in view["visible"][:6]) - result = f"{view['kind']} view of {visible}" - if view["filters"]: - result += "; filtered records only" - if view.get("groupBy"): - result += f"; grouped by {label.get(view['groupBy'], view['groupBy'])}" - agg = view["aggregation"] - if agg["op"] != "count": - result += f"; {agg['op']} of {label.get(agg['field'], agg['field'])}" - return result + "." - - -def _numeric_result(view, records): - aggregation = view["aggregation"] - if aggregation["op"] == "count": - return {"label": "Matching records", "value": len(records), "contributing_record_count": len(records)} - values = [] - for record in records: - try: - value = record.get(aggregation["field"]) - if value is not None and not isinstance(value, bool): - values.append(float(value)) - except (TypeError, ValueError): - continue - if not values: - return {"label": aggregation["op"], "value": None, "contributing_record_count": 0} - op = aggregation["op"] - value = {"sum": sum(values), "avg": sum(values) / len(values), "min": min(values), "max": max(values)}[op] - return {"label": f"{op.title()} of {aggregation['field']}", "value": value, - "contributing_record_count": len(values)} - - -def _referenced_fields(view): - """The citation names every source field that affected the displayed result.""" - out = list(view.get("visible") or ()) - for node in view.get("filters") or (): - if isinstance(node, dict) and node.get("colId"): - out.append(str(node["colId"])) - for node in view.get("sorts") or (): - if isinstance(node, dict) and node.get("colId"): - out.append(str(node["colId"])) - for key in ("groupBy",): - if view.get(key): - out.append(str(view[key])) - aggregation = view.get("aggregation") or {} - if aggregation.get("field"): - out.append(str(aggregation["field"])) - return list(dict.fromkeys(out)) - - -def _effective_filters(snapshot, view): - """Keep the source request and generated-view predicates distinct in provenance.""" - return { - "source": _safe(snapshot.get("filters")), - "view": {"conj": view.get("filterConj", "and"), - "nodes": _safe(view.get("filters") or [])}, - } - - -def _view_records(snapshot, view): - """Apply the exact validated virtual-view filter before calculating a cited number.""" - nodes = view.get("filters") or [] - if not nodes: - return list(snapshot["records"]) - from harness import filter_eval - tree = {"conj": view.get("filterConj", "and"), "nodes": nodes} - return [row for row in snapshot["records"] - if filter_eval.matches(tree, row, snapshot["fields"])] - - -def _citation(citation_id, snapshot, view, numeric, view_id): - return { - "id": citation_id, - "href": f"#/query?view={view_id}&citation={citation_id}", - "database": snapshot["database"], - "snapshot": {"kind": snapshot["source_kind"], "version": _safe(snapshot["source_version"])}, - "fields": [field["key"] for field in snapshot["fields"] - if field.get("key") in set(_referenced_fields(view))], - "filters": _effective_filters(snapshot, view), - "permission_scope_applied": snapshot["permission_scope_applied"], - "aggregation": _safe(view["aggregation"]), - "contributing_record_count": numeric["contributing_record_count"], - "retrieved_at": snapshot["retrieved_at"], - } - - -def _citation_complete(citation): - """A numeric result is not publishable without complete provenance.""" - required = {"database", "snapshot", "fields", "filters", "aggregation", - "contributing_record_count", "retrieved_at", "href"} - snapshot = citation.get("snapshot") if isinstance(citation, dict) else None - return (isinstance(citation, dict) and required <= set(citation) - and bool(citation["database"]) and isinstance(snapshot, dict) - and "version" in snapshot and bool(citation["retrieved_at"]) - and isinstance(citation["fields"], list) - and isinstance(citation["aggregation"], dict)) - - -def _public_view(view): - source = view["source"] - return { - "id": view["id"], "viewId": view["id"], "scope": source["database"], - "name": view["name"], "kind": view["view"]["kind"], "question": view["question"], - "explain": view["explain"], "threadId": view["threadId"], "createdAt": view["createdAt"], - "virtual": True, "source": _safe(source), "view": _safe(view["view"]), - "citationIds": list(view["citationIds"]), "numeric": _safe(view["numeric"]), - } - - -def _source_still_permitted(session, source, permitted=None): - """A persisted artefact never outlives the caller's current data permission. - - ⚠ `permitted` is the BATCH answer from `deps.assistant_source_status` — one rows-free - resolution for every source at once, rather than one whole-document read per saved view. That - function's docstring carries the measurement. The single-source path below stays for callers - holding exactly one artefact (the workspace-event door), and asks the same helper. - - ⛔ `permitted`, NEVER `answerable`. A source whose rows are served through the connector mirror - cannot be ASKED and can still be SEEN: the artefact was built from a snapshot that was legal - when it was taken, and hiding it because the reader can no longer make a NEW one would read as - deletion. The two verdicts are separate for that reason. - """ - database = (source or {}).get("database") if isinstance(source, dict) else None - if not database: - return False - if permitted is None: - permitted = {key for key, row in assistant_source_status(session, [database]).items() - if row["permitted"]} - return database in permitted - - -def _public_state(state, session): - # ONE rows-free batch answers both questions: which sources this caller may still see (which - # artefacts stay listed) and which of them can actually be asked (which chips are live). - status = assistant_source_status(session) - for row in state["views"].values(): - key = (row.get("source") or {}).get("database") if isinstance(row.get("source"), dict) else None - if key and key not in status: - # An artefact whose source is no longer enumerable still gets a verdict rather than a - # KeyError — it resolves to "not permitted" and the artefact drops out, which is the - # same answer the per-view wall gave. - status.update(assistant_source_status(session, [key])) - permitted = {key for key, row in status.items() if row["permitted"]} - allowed_views = [row for row in state["views"].values() - if _source_still_permitted(session, row.get("source"), permitted)] - allowed_ids = {row["id"] for row in allowed_views} - allowed_citations = {citation_id for row in allowed_views - for citation_id in row.get("citationIds") or []} - threads = sorted(state["threads"].values(), key=lambda row: row.get("updatedAt", ""), reverse=True) - messages = sorted(state["messages"].values(), key=lambda row: row.get("createdAt", "")) - views = sorted((_public_view(row) for row in allowed_views), key=lambda row: row["createdAt"], reverse=True) - messages = [row for row in messages if not row.get("viewId") or row.get("viewId") in allowed_ids] - return {"threads": _safe(threads), "messages": _safe(messages), "views": views, - "citations": _safe([row for key, row in state["citations"].items() - if key in allowed_citations]), "models": model_choices(), - # The chip row's own data: a source this caller holds but cannot ask, and WHY. - # ⭐⭐ D-276 — `or row.get("visible")`. A source the caller can open elsewhere in the - # product but the assistant cannot read is now LISTED with its cause, instead of - # vanishing from a picker that shows every other database they hold. It arrives with - # `answerable: False` and a `reason`, which is the SAME shape the mirror-served grids - # already use, so the chip row needs no new state to render it. - # ⛔ `permitted` STILL GATES ARTEFACTS — `allowed_views` above is unchanged. This widens - # what is DESCRIBED, never what can be read or kept. - "sources": _safe([{"database": key, "answerable": row["answerable"], - "reason": row["reason"]} - for key, row in sorted(status.items()) - if row["permitted"] or row.get("visible")])} - - -@router.get("/query") -def list_queries(session: Session = Depends(require_session)): - return _public_state(_state(session), session) - - -@router.post("/query/{qid}/events") -def mutate_query_workspace(qid: str, body: dict = Body(default=None), - session: Session = Depends(require_session)): - """The only grid-mutation transport for a virtual Query workspace. - - Query artefacts are immutable snapshots: creation and changes belong to an Assistant prompt, - never a source-native grid workspace. Deleting the caller's personal artefact is the one - permitted mutation. The binding key, not a client-supplied source scope, identifies it. - """ - qid = str(qid) - body = body if isinstance(body, dict) else {} - binding = body.get("workspaceBinding") - event = body.get("event") - if not (isinstance(binding, dict) and binding.get("kind") == "query" - and str(binding.get("key") or "") == qid): - raise err(400, "query_workspace_mismatch", - "the Query workspace binding must name this virtual artefact") - if not isinstance(event, dict) or str(event.get("type") or "") not in QUERY_WORKSPACE_EVENTS: - raise err(400, "unsupported_query_workspace_event", - "Query accepts only view_create, view_upsert, or view_delete events") - - state = _state(session) - view = state["views"].get(qid) - if view is None or not _source_still_permitted(session, view.get("source")): - # A caller cannot use an opaque Query key to learn about a revoked artefact. - raise err(404, "unknown_query", "that Query artefact does not exist") - - event_type = str(event["type"]) - if event_type != "view_delete": - raise err(409, "query_workspace_immutable", - "AI-created Query views change only through a new Assistant prompt") - if str(event.get("viewId") or "") != qid: - raise err(400, "query_workspace_mismatch", - "a Query delete must name the same virtual artefact as its binding") - - deleted = delete_query(qid, session) - return {"workspaceBinding": {"kind": "query", "key": qid}, - "event": event_type, **deleted} - - -def _sources(body, target, session): - raw = body.get("sources") if isinstance(body, dict) else None - sources = [str(item).strip() for item in raw] if isinstance(raw, list) else [] - sources = list(dict.fromkeys(item for item in sources if item)) - if target not in sources: - sources.insert(0, target) - # Source chips are a permissioned selection, not untrusted labels remembered in a thread. - # The model still receives ONLY `target`'s snapshot below; these calls do not pass a second - # source to it and D's helper refuses cache misses, mirrors and unshared databases. - for source in sources: - assistant_read_scope(session, source, fields=[], filters=None) - return sources - - -def _submit(body, session, chat=None): - body = body if isinstance(body, dict) else {} - question = " ".join(str(body.get("question") or "").split()) - target = str(body.get("database") or body.get("scope") or "").strip() - selected_model = str(body.get("model") or MODEL_AUTO).strip().lower() - 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") - if not target: - raise err(400, "bad_request", "one target database must be selected") - if selected_model not in model_choices(): - raise err(400, "unknown_model", "that model is not available in Query") - - # This call happens before model selection and is the only permitted source-data read. - snapshot = assistant_read_scope(session, target, fields=None, filters=body.get("filters")) - sources = _sources(body, target, session) - state = _state(session) - thread_id = str(body.get("threadId") or "").strip() - if thread_id and thread_id not in state["threads"]: - raise err(404, "unknown_thread", "that chat does not exist") - if not thread_id: - thread_id = _new_id("thread") - now = _now_iso() - user_message_id = _new_id("message") - assistant_message_id = _new_id("message") - spec, sentence, provider, reason = _call_model(question, snapshot, model=selected_model, chat=chat) - view, refusal, refusal_code = _validate(spec, snapshot["fields"]) if spec is not None else (None, sentence, reason) - - artifact = None - citation = None - if view is not None: - if len(state["views"]) >= MAX_ARTIFACTS: - raise err(400, "query_limit", f"you already have {len(state['views'])} Query artefacts") - view_id = _new_id("query") - citation_id = _new_id("citation") - numeric = _numeric_result(view, _view_records(snapshot, view)) - citation = _citation(citation_id, snapshot, view, numeric, view_id) - if not _citation_complete(citation): - raise RuntimeError("Query refused to persist an incomplete numeric citation") - source = {"database": snapshot["database"], "label": snapshot["label"], - "source_kind": snapshot["source_kind"], "source_version": _safe(snapshot["source_version"]), - "retrieved_at": snapshot["retrieved_at"], - "fields": _safe(snapshot["fields"]), "filters": _safe(snapshot["filters"]), - "permission_scope_applied": snapshot["permission_scope_applied"]} - artifact = {"id": view_id, "threadId": thread_id, "name": view["name"], "question": question, - "explain": _explain(view, snapshot["fields"]), "createdAt": now, "source": source, - "view": _safe(view), "citationIds": [citation_id], "numeric": numeric, - "model": provider, "requestedModel": selected_model} - - assistant_message = { - "id": assistant_message_id, "threadId": thread_id, "role": "assistant", "createdAt": now, - "content": artifact["explain"] if artifact else (refusal or "the assistant could not answer"), - "targetDatabase": target, "requestedModel": selected_model, "model": provider, - "reason": refusal_code, "viewId": artifact["id"] if artifact else None, - "citationIds": artifact["citationIds"] if artifact else [], "numeric": artifact["numeric"] if artifact else None, - } - user_message = {"id": user_message_id, "threadId": thread_id, "role": "user", "createdAt": now, - "content": question, "sources": sources, "targetDatabase": target, - "requestedModel": selected_model} - - def update(raw): - current = _blank_state() - if isinstance(raw, dict): - for key in ("threads", "messages", "citations", "views"): - if isinstance(raw.get(key), dict): - current[key] = copy.deepcopy(raw[key]) - thread = current["threads"].get(thread_id) or {"id": thread_id, "createdAt": now} - thread.update({"updatedAt": now, "title": question[:80], "sources": sources, - "model": selected_model, "activeViewId": artifact["id"] if artifact else thread.get("activeViewId")}) - current["threads"][thread_id] = thread - current["messages"][user_message_id] = user_message - current["messages"][assistant_message_id] = assistant_message - if artifact: - current["views"][artifact["id"]] = artifact - current["citations"][citation["id"]] = citation - return current - - if not session.runtime.available(): - raise err(503, "store_unavailable", "the tenant store is unavailable; no chat was saved") - session.runtime.update(_namespace_key(session), update, flush="async") - return {"thread": _safe(update(state)["threads"][thread_id]), "userMessage": _safe(user_message), - "message": _safe(assistant_message), - "view": _public_view(artifact) if artifact else None, "citations": [citation] if citation else []} - - -@router.post("/query/chat") -def submit_chat(body: dict = Body(default=None), session: Session = Depends(require_session)): - return _submit(body, session) - - -@router.delete("/query/threads/{tid}") -def delete_thread(tid: str, session: Session = Depends(require_session)): - """Remove one of the caller's own chats, and only the chat. - - ⛔ THE ARTEFACTS SURVIVE, AND THAT IS THE POINT. A Query view is a durable object in its own - right — it appears in Query's rail, other people's links can point at it, and the owner's - instruction was that AI views LIVE THERE rather than inside the conversation that happened to - produce them. Cascading the delete would make tidying up a chat silently destroy work. Views - are deleted from Query's own rail, one at a time, by `delete_query` below. - - ⚠ It exists because the history had no prune. `delete_query` removed a view and left its - thread, so the panel could only ever grow — a navigation you cannot manage is half a - navigation, and a person clearing test chats found that out first. - """ - tid = str(tid) - state = _state(session) - if tid not in state["threads"]: - raise err(404, "unknown_thread", "that chat does not exist") - - def update(raw): - current = _blank_state() - if isinstance(raw, dict): - for key in ("threads", "messages", "citations", "views"): - if isinstance(raw.get(key), dict): - current[key] = copy.deepcopy(raw[key]) - current["threads"].pop(tid, None) - for message_id in [key for key, row in current["messages"].items() - if (row or {}).get("threadId") == tid]: - current["messages"].pop(message_id, None) - # ⚠ THE VIEWS' `threadId` IS LEFT EXACTLY AS IT WAS, dangling. Blanking it looks tidier and - # is a data-loss bug: `queryApi.saved()` refuses a row with an empty `threadId`, so every - # artefact built in the deleted chat would silently vanish from Query's rail — the exact - # "tidying up destroys work" outcome this door is written not to have - # [[a-record-can-outlive-its-subject]]. Nothing resolves the id but the chat panel, which - # only ever looks up the thread it is showing. - return current - - session.runtime.update(_namespace_key(session), update, flush="async") - return {"deleted": tid} - - -@router.delete("/query/{qid}") -def delete_query(qid: str, session: Session = Depends(require_session)): - qid = str(qid) - state = _state(session) - if qid not in state["views"]: - raise err(404, "unknown_query", "that Query artefact does not exist") - - def update(raw): - current = _blank_state() - if isinstance(raw, dict): - for key in ("threads", "messages", "citations", "views"): - if isinstance(raw.get(key), dict): - current[key] = copy.deepcopy(raw[key]) - view = current["views"].pop(qid, None) - for citation_id in (view or {}).get("citationIds") or []: - current["citations"].pop(citation_id, None) - return current - - session.runtime.update(_namespace_key(session), update, flush="async") - return {"deleted": qid} +"""Assistant chat and Query-owned virtual artefacts (W33-T72 / C8). + +Query reads only the detached, permission-filtered snapshot from +``deps.assistant_read_scope``. It owns a separate per-user namespace for threads, messages, +citations and virtual views; it never opens or mutates a source workspace. +""" +import copy +import csv +import datetime as _dt +import hashlib +import io +import json +import os +import re +import uuid + +from fastapi import APIRouter, Body, Depends, Response + +from deps import (Session, assistant_read_scope, assistant_source_status, err, + require_session) + +router = APIRouter(prefix="/api/v1") + +MAX_QUESTION = 500 +MAX_VISIBLE = 24 +MAX_ARTIFACTS = 200 +# ⭐⭐ R16 — THE THREAD IS REPLAYED, AND THESE TWO NUMBERS ARE THE WHOLE BOUND ON IT. +# The chat was STATELESS: `_call_model` built `[system, user(question)]` and the conversation on +# screen reached the model as nothing at all, so "what about the other one?" could not be answered +# and the assistant could not grill anybody about anything. Replay is bounded twice because the +# two limits fail differently: a turn COUNT stops a long chat costing more every turn, and a +# CHARACTER budget stops one pasted wall of text from being the whole context. Both count from the +# NEWEST end, because that is what a pronoun refers to. +MAX_HISTORY_TURNS = 12 +MAX_HISTORY_CHARS = 6000 +MAX_RATING_REASON = 200 +QUERY_RATINGS = ("up", "down") +MODEL_AUTO = "auto" +# ⛔⛔ WHY QUERY DECLARES THE ENV NAMES RATHER THAN READING `analyst.PROVIDERS`: MEASURED +# 2026-08-16, `import harness.analyst` costs **5 to 8 SECONDS**, warm, on this box (it pulls +# `harness.tools`, which pulls yaml and the skill recipes). `GET /query` is a LIST path that BOTH +# the Assistant and Query call on mount, so importing the ladder to answer "which models can this +# deployment actually call?" would put eight seconds on the first paint of the AI module. The map +# below is Query's own declaration, and it is deliberately the cheap half. +# ⚠ A SECOND DECLARATION IS A DIVERGENCE WAITING TO HAPPEN, so it is CHECKED rather than trusted: +# `_providers()` already holds the platform ladder in memory and reconciles there, and anything it +# finds is REPORTED into the payload instead of a provider silently never being offered. +# The ORDER is Query's own and is deliberate: cerebras first, because this path needs tool calling +# and cerebras carries this account's tool-capable model. +QUERY_PROVIDER_ENV = { + "cerebras": "CEREBRAS_API_KEY", + "groq": "GROQ_API_KEY", + "openrouter": "OPENROUTER_API_KEY", +} +QUERY_PROVIDER_ORDER = tuple(QUERY_PROVIDER_ENV) +# Filled the first time the platform ladder is loaded; empty means "not yet known", never "clean". +_LADDER_DRIFT = [] +QUERY_KINDS = ("grid", "chart", "calendar", "kanban", "timeseries", "map", "list") +QUERY_WORKSPACE_EVENTS = {"view_create", "view_upsert", "view_delete"} +QUERY_EXCLUDED = { + "form": "Forms collect new data and are not a read-only Query artefact.", + "catalog": "Catalog is a source-native presentation that Query cannot safely mutate.", + "swipe": "Swipe is an interactive source-native presentation, not an Assistant output.", +} +_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(): + return _dt.datetime.now(_dt.timezone.utc).isoformat() + + +def _grid(): + import aios_grid + return aios_grid + + +def _safe(value): + """Detach values before they enter a durable Query object or an API reply.""" + return json.loads(json.dumps(value, default=str)) + + +def _namespace_key(session): + """A tenant runtime has one isolated key for each user's Query-owned objects.""" + principal = f"{session.tenant}:{session.uname}".encode("utf-8") + return "query_user_" + hashlib.sha256(principal).hexdigest()[:24] + + +def _blank_state(): + return {"version": 1, "threads": {}, "messages": {}, "citations": {}, "views": {}} + + +def _state(session): + try: + raw = session.runtime.get(_namespace_key(session)) or {} + except Exception: + raw = {} + if not isinstance(raw, dict): + return _blank_state() + out = _blank_state() + for key in out: + if key == "version": + continue + if isinstance(raw.get(key), dict): + out[key] = copy.deepcopy(raw[key]) + return out + + +def _new_id(prefix): + return f"{prefix}_{uuid.uuid4().hex[:16]}" + + +def model_choices(): + """Choices are stable even when one is not configured, so explicit means explicit.""" + return [MODEL_AUTO, *QUERY_PROVIDER_ORDER] + + +def model_status(): + """Per choice: can this deployment actually CALL it, and if not, why not. + + ⭐⭐ THE SAME DEFECT THE SOURCE CHIPS WERE FIXED FOR IN WAVE 33, ONE CONTROL TO THE LEFT. + `permitted` is not `answerable` for a database; OFFERED is not CALLABLE for a model. The picker + listed every name in the ladder whether or not this deployment holds its key, so choosing one + spent a click and a turn to be told "the selected model is unavailable" by the server, which + knew before the click. The reason travels with the flag, exactly as `sources` does. + + ⚠ It names no environment variable. "This model is not configured on this deployment" is the + cause a tenant user can act on (ask an admin); the variable name is our deployment's business. + """ + live = [name for name in QUERY_PROVIDER_ORDER if os.environ.get(QUERY_PROVIDER_ENV[name])] + rows = [{"model": MODEL_AUTO, "available": bool(live), + "reason": "" if live else "no model is configured on this deployment"}] + for name in QUERY_PROVIDER_ORDER: + rows.append({"model": name, "available": name in live, + "reason": "" if name in live + else "this model is not configured on this deployment"}) + # Anything the platform ladder carries that Query does not offer, once a chat has taught us. + rows.extend({"model": name, "available": False, + "reason": "the platform can reach this model but Query does not offer it yet"} + for name in _LADDER_DRIFT) + return rows + + +def _providers(model=MODEL_AUTO): + """Auto returns the permitted ladder; an explicit choice returns at most one provider.""" + import harness.analyst as analyst + + requested = str(model or MODEL_AUTO).strip().lower() + by_name = {p["name"]: p for p in analyst.PROVIDERS} + # ⚠ THE RECONCILIATION, at the ONE point the platform ladder is already in memory, so it costs + # nothing. A provider added to `analyst.PROVIDERS` would otherwise simply never appear here, + # and an env name changed there would make Query's availability answer quietly wrong. Both are + # invisible failures; this turns them into a row a reader can see [[limit-with-no-enforcer]]. + _LADDER_DRIFT[:] = sorted( + set(by_name) - set(QUERY_PROVIDER_ENV) + | {name for name, row in by_name.items() + if name in QUERY_PROVIDER_ENV and row.get("env") != QUERY_PROVIDER_ENV[name]}) + names = list(QUERY_PROVIDER_ORDER) if requested == MODEL_AUTO else [requested] + return [by_name[name] for name in names + if name in by_name and os.environ.get(by_name[name]["env"])] + + +def _history(state, thread_id): + """The prior turns of THIS thread, oldest first, in the transport's own message shape. + + ⛔ SORTED BY `(createdAt, role)`, NOT BY `createdAt` ALONE. `_submit` stamps the user turn and + the assistant turn with the SAME `now`, so on timestamp alone the pair ties and their order is + whatever the store's dict iteration happens to give — which is insertion order today and is + not a guarantee anybody wrote down. A replayed conversation with the answers before the + questions is worse than no replay: it reads as coherent and is backwards. + + ⚠ Scoped to one thread by construction. Replaying another thread's turns would leak one chat's + subject into another's answer, which is the kind of wrong that looks like a good answer. + """ + thread_id = str(thread_id or "") + if not thread_id: + return [] + rows = [row for row in state["messages"].values() + if isinstance(row, dict) and str(row.get("threadId") or "") == thread_id] + rows.sort(key=lambda row: (str(row.get("createdAt") or ""), + 0 if row.get("role") == "user" else 1)) + turns = [] + for row in rows[-MAX_HISTORY_TURNS:]: + content = " ".join(str(row.get("content") or "").split()) + if not content: + continue + role = "assistant" if row.get("role") == "assistant" else "user" + turns.append({"role": role, "content": content}) + # ⭐⭐ R20's "self-improving", and it is the ONLY thing that makes a rating more than a + # flag shipped without its writer. A thumbs-down becomes what it actually was: a turn in + # which the reader said the answer was not useful. Modelled as a USER turn rather than by + # editing the assistant's own words, because that is what happened, and because rewriting + # a stored answer to steer the next one is how a transcript stops being a record. + if role == "assistant" and row.get("rating") == "down": + reason = " ".join(str(row.get("ratingReason") or "").split())[:MAX_RATING_REASON] + turns.append({"role": "user", + "content": f"That answer was not helpful. {reason}".strip()}) + kept, spent = [], 0 + for turn in reversed(turns): + spent += len(turn["content"]) + if spent > MAX_HISTORY_CHARS and kept: + break + kept.append(turn) + kept.reverse() + return kept + + +def _spec_schema(field_keys): + col = {"type": "string", "enum": sorted(field_keys)} + return { + "type": "object", + "properties": { + "kind": {"type": "string", "enum": [*QUERY_KINDS, "refused"]}, + "name": {"type": "string"}, + "refusal": {"type": "string"}, + "visible": {"type": "array", "items": col}, + # ⭐⭐ 2026-08-15 (owner: *"if you can build the view, the AI assistant should also be + # able to do it by using our tools on the backend"*). `rhs` and `important` were the + # only two things a person could express through `view_upsert` and this tool could not. + # + # ⛔ WITHOUT `rhs` THE ASSISTANT CANNOT STATE AN ERROR-CATCHER AT ALL — the one shape + # item 8a named by hand (*"price and COGS not matching"*). Every one of the 20 + # `FILTER_OPS` was already reachable, because they all read `value`; comparing a column + # against ANOTHER COLUMN is a different member (`aios_grid._clean_rhs`, CG-9) and it was + # simply absent here, so the model had no way to ask for it and no way to be told why. + # `kind` is `field` ONLY: `measure` needs a window and `stat` needs the population + # vocabulary, neither of which this snapshot-shaped reader carries — offering them + # would be a control that lies, which is the same rule the source chips follow. + "filters": {"type": "array", "items": {"type": "object", "properties": { + "colId": col, "op": {"type": "string", "enum": sorted(_grid().FILTER_OPS)}, + "value": {"type": "string"}, + "rhs": {"type": "object", "properties": { + "kind": {"type": "string", "enum": ["field"]}, "colId": col, + }, "required": ["kind", "colId"]}, + }, "required": ["colId", "op"]}}, + # A personal legibility mark, exactly as `grid_events.view_upsert` treats it (wave 32 + # R5/C4) — not a lock, and no second permission wall. + "important": {"type": "boolean"}, + "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, + "aggregation": {"type": "object", "properties": { + "op": {"type": "string", "enum": ["count", "sum", "avg", "min", "max"]}, + "field": col, + }, "required": ["op"]}, + "stackField": col, + "dateField": col, + "colorField": col, + "sizeField": col, + }, + "required": ["kind"], + } + + +def _system_prompt(snapshot): + fields = snapshot["fields"] + cols = json.dumps([{"key": field["key"], "label": field.get("label", field["key"]), + "type": field.get("type", "text")} for field in fields], separators=(",", ":")) + return f"""You are the assistant inside this product, talking with one person about exactly one +database they already have permission to read. Reply in plain prose, and keep it short. + +DATABASE: {snapshot['database']} +SNAPSHOT VERSION: {json.dumps(snapshot['source_version'], default=str)} +PERMISSION-FILTERED RECORD COUNT: {len(snapshot['records'])} +FIELDS: {cols} +VIEW KINDS: {", ".join(QUERY_KINDS)} + +You may answer in words alone, and you may ask ONE short question back when the request is +ambiguous: which field is meant, which period, whether they want every record or a narrower set. +Prefer asking over guessing whenever the answer would change what you build. Earlier turns of this +conversation are above; use them, and read a pronoun as referring to what was just discussed. + +Call build_view ONLY when the person wants to SEE records: a table, a chart, a board, a list. Do +not call it to explain something, to confirm something, or to ask your question. When you do call +it you may also write one sentence saying what you built. + +Inside build_view: choose visible fields, supported filters and an optional aggregation. Use count +for record counts; sum, avg, min and max require one numeric field. The server computes and cites +every number from this exact snapshot. Never write SQL, invent fields, or name another database. +Return kind=refused with one plain sentence if the database cannot answer. + +To compare one column against ANOTHER column rather than a typed value, give the filter an rhs of +{{"kind":"field","colId":""}} and omit value. That is how you express questions like +"priced below what it costs us". Both columns must be in the FIELDS list above. +Set important=true when the view is one somebody should be chased about: an error, a mismatch, or +money at risk. Leave it out otherwise. + +Write with ordinary punctuation. Never use an em dash or an en dash: use a comma, a colon or a full +stop instead.""" + + +_FAILED_GEN = re.compile(r"(\{.*?\})\s*", re.S) + + +def _spec_from_400(body): + try: + value = ((json.loads(body) or {}).get("error") or {}).get("failed_generation") or "" + raw = (_FAILED_GEN.search(str(value)) or [str(value).strip()])[1] + result = json.loads(raw) + return result if isinstance(result, dict) else None + except Exception: + return None + + +# ⚠ BUILT FROM `chr`, AND A UNICODE ESCAPE IS NOT ENOUGH. `web_prose` reads a Python +# string's VALUE off the AST, not its spelling in the source, so an escape and the character +# itself are the SAME finding to it. Composing the class at runtime means no string literal in +# this file holds a dash, which is true rather than merely quiet: the only dashes in this module +# are the ones being removed. +_DASH = "[" + chr(0x2014) + chr(0x2013) + "]" + + +def _no_dashes(text): + """R6 applied where it is the ONLY place it can be applied: the model's own words. + + ⛔⛔ THE PROMPT INSTRUCTION IS NOT ENFORCEMENT, AND THIS IS MEASURED, NOT ANTICIPATED. The + system prompt ends with *"Never use an em dash or an en dash"* and the very next live turn came + back with *"Which view type would you like—grid, chart, list, or another?"* (cerebras, + 2026-08-16). Model prose reaches the screen exactly as a string literal does, and `web_prose` + scans SOURCE, so it cannot see a dash that arrives at runtime: the sweep every lane is doing + this wave is undone by our own assistant unless it is undone here. + + ⚠ A DIGIT RANGE IS A DIFFERENT SENTENCE. "10–20" means "10 to 20"; rewriting it as + "10, 20" states two numbers where the model stated a span, which is a wrong answer rather than + a punctuation fix. It gets its own rule, first. + """ + text = str(text or "") + text = re.sub(rf"(?<=\d)\s*{_DASH}\s*(?=\d)", " to ", text) + text = re.sub(rf"\s*{_DASH}\s*(?=[,.;:!?])", "", text) # abutting punctuation: it just goes + text = re.sub(rf"(?<=[,;:])\s*{_DASH}\s*", " ", text) # already punctuated: one space + return re.sub(rf"\s*{_DASH}\s*", ", ", text) + + +def _said(raw): + """The model's own words: one line of whitespace, no dash, and empty means nothing was said.""" + return _no_dashes(" ".join(str(raw or "").split())).strip() or None + + +def _from_chat(answer, provider): + """An injected transport may answer with a SPEC (a dict) or with PROSE (a string). + + ⛔ THE BRANCH BELONGS HERE, NOT AT THE CALLER. `_validate`'s first line refuses anything that + is not a dict, with "the assistant did not answer with a view" — so a prose answer passed + straight through would arrive as a REFUSAL, styled as one, and every conversation check would + pass against the wrong path while looking green [[one-question-two-normalizers]]. + """ + if isinstance(answer, str): + return None, _said(answer), provider, None + return answer, None, provider, None + + +def _call_model(question, snapshot, model=MODEL_AUTO, chat=None, history=None): + """Return ``(spec, said, provider, reason)`` without any source-data fallback. + + ``said`` is what the assistant SAID: the whole answer when it did not build a view, and the + sentence beside the view when it built one and talked as well. ``reason`` is set only when + something went wrong, so a prose ANSWER and a transport FAILURE are distinguishable one layer + up rather than both arriving as a bare sentence. + """ + requested = str(model or MODEL_AUTO).strip().lower() + if requested not in model_choices(): + return None, f"the selected model ({requested or model}) is unavailable", None, "model_unavailable" + tools = [{"type": "function", "function": { + "name": "build_view", "description": "Emit a virtual-view spec or a refusal.", + "parameters": _spec_schema([field["key"] for field in snapshot["fields"]]), + }}] + messages = [{"role": "system", "content": _system_prompt(snapshot)}, + *(history or []), + {"role": "user", "content": question}] + if chat is not None: + provider = requested if requested != MODEL_AUTO else "injected" + return _from_chat(chat(messages, tools), provider) + + providers = _providers(requested) + if not providers: + sentence = (f"the selected model ({requested}) is unavailable" if requested != MODEL_AUTO + else "the assistant is not configured on this deployment") + return None, sentence, None, "model_unavailable" if requested != MODEL_AUTO else "not_configured" + + import requests + last = None + for provider in providers: + try: + response = requests.post( + provider["url"], timeout=60, + headers={"Authorization": f"Bearer {os.environ[provider['env']]}"}, + # ⛔⛔ `auto`, NOT `required`, AND THAT ONE WORD IS R16. Under `required` the model + # had to emit a `build_view` call for EVERY turn: it could not answer a question, + # could not ask one back, and could not decline to build. "The AI chat must TALK + # with the user, not only build queries, and decide for itself whether to create a + # query view" is unreachable with the tool forced, whatever the prompt says. + # ⚠ AND IT IS A LIVE VENDOR-BEHAVIOUR CHANGE THAT NO GATE HERE CAN SEE, because + # `verify_query` injects `chat` and never builds this body. `_spec_from_400` below + # exists precisely because providers differ in how they emit a forced call; under + # `auto` a provider may also decline to build for a view-shaped question. That is + # why this ticket's evidence is a real three-turn run against a real provider and + # not a green gate. + json={"model": provider["model"], "messages": messages, "tools": tools, + "tool_choice": "auto", "temperature": 0.1, "max_tokens": 1200}, + ) + except Exception as exc: + last = f"{provider['name']}: {type(exc).__name__}" + continue + if response.status_code == 400: + spec = _spec_from_400(response.text) + if spec is not None: + return spec, None, provider["name"], None + last = f"{provider['name']}: 400" + continue + if response.status_code != 200: + last = f"{provider['name']}: HTTP {response.status_code}" + continue + try: + answer = response.json()["choices"][0]["message"] + said = _said(answer.get("content")) + calls = answer.get("tool_calls") or [] + if calls: + # The prose rides ALONG with the view when the model wrote both, so the reader gets + # a sentence instead of `_explain`'s machine description of its own output. + spec = json.loads(calls[0]["function"].get("arguments") or "{}") + return spec, said, provider["name"], None + if said: + return None, said, provider["name"], None + # ⚠ Neither a call nor a word is a FAILED turn, not a silent one: falling through to + # the next provider is right, and swallowing it as an empty answer would show the + # reader a blank reply [[empty-answer-vs-unfinished-answer]]. + last = f"{provider['name']}: an empty answer" + except Exception as exc: + last = f"{provider['name']}: unreadable answer ({type(exc).__name__})" + if requested != MODEL_AUTO: + return None, f"the selected model ({requested}) is unavailable", None, "model_unavailable" + return (None, "the assistant could not be reached just now (" + (last or "no provider") + ")", + None, "provider_unreachable") + + +def _validate(spec, fields): + """Return a cleaned, source-independent virtual-view config or a named refusal.""" + if not isinstance(spec, dict): + return None, "the assistant did not answer with a view", "no_spec" + by_key = {str(field.get("key")): str(field.get("type") or "text") for field in fields} + keys = set(by_key) + kind = spec.get("kind") + if kind == "refused": + # Model-authored, so it faces the same R6 wall the model's chat prose does. + return None, _said(spec.get("refusal")) or "this database cannot answer that question", "model_refused" + if kind in QUERY_EXCLUDED or kind not in QUERY_KINDS: + return None, "that kind of view cannot be built from this question", "unsupported_kind" + + named = set(spec.get("visible") or ()) + for item in spec.get("filters") or (): + if isinstance(item, dict) and item.get("colId"): + named.add(str(item["colId"])) + # ⛔ THE RIGHT-HAND COLUMN IS A COLUMN AND MUST FACE THE SAME `missing` CHECK. Collecting + # only the left side is what makes D-229 possible one layer down: `clean_filter_tree` does + # NOT drop a leaf whose field-rhs names a column that does not exist — it keeps the leaf, + # strips the `rhs`, blanks the value, and `filter_sql.is_rule_active` then reports the rule + # INACTIVE. An inactive rule narrows nothing, so "margin under 10%" would come back as a + # view listing the ENTIRE catalogue under an error-catcher's name, with nothing red. + # Naming it here turns that into the ordinary "this database does not have: X" refusal. + rhs = item.get("rhs") if isinstance(item, dict) else None + if isinstance(rhs, dict) and rhs.get("colId"): + named.add(str(rhs["colId"])) + for item in spec.get("sorts") or (): + if isinstance(item, dict) and item.get("colId"): + named.add(str(item["colId"])) + for key in ("groupBy", "stackField", "dateField", "colorField", "sizeField"): + if spec.get(key): + named.add(str(spec[key])) + raw_aggregation = spec.get("aggregation") or {"op": "count"} + if isinstance(raw_aggregation, dict) and raw_aggregation.get("field"): + named.add(str(raw_aggregation["field"])) + missing = sorted(named - keys) + if missing: + return None, "this database does not have: " + ", ".join(missing), "unknown_columns" + + visible = [str(key) for key in (spec.get("visible") or ()) if str(key) in keys][:MAX_VISIBLE] + if not visible: + return None, "that question did not name any fields to show", "no_columns" + raw_filters = [item for item in (spec.get("filters") or ()) if isinstance(item, dict)] + filters = _grid().clean_filter_tree(raw_filters, keys) + if len(filters) != len(raw_filters): + return None, "part of that filter is unsupported", "filter_dropped" + # ⛔⛔ A SECOND, NARROWER CHECK, AND THE LENGTH CHECK ABOVE CANNOT DO ITS JOB (D-229). + # A dropped leaf changes the COUNT; a stripped `rhs` does not — the leaf survives, so + # `len(filters) == len(raw_filters)` and the refusal above never fires. The failure is + # therefore silent in exactly the direction that matters: the condition stops narrowing and + # the view answers with every record. Assert the member survived, per leaf. + # ⚠ The `named` pass above already refuses an rhs naming a column this database lacks, so + # reaching here means something ELSE stripped it (a type the comparand cannot take, a future + # `_clean_rhs` rule). Both doors, because the two catch different causes and the cost of + # missing this one is a wrong answer that looks right. + for sent, kept in zip(raw_filters, filters): + if sent.get("rhs") and not kept.get("rhs"): + return None, "that column cannot be compared against another column", "rhs_dropped" + + aggregation = raw_aggregation if isinstance(raw_aggregation, dict) else {} + op = str(aggregation.get("op") or "").lower() + field = aggregation.get("field") + if op not in {"count", "sum", "avg", "min", "max"}: + return None, "the assistant gave an unsupported aggregation", "bad_aggregation" + if op == "count": + field = None + elif field not in keys or by_key.get(field) not in {"int", "currency", "pct"}: + return None, "that aggregation needs one visible numeric field", "bad_aggregation" + + display = {"mode": kind} + for ref, families, required in _MODE_REFS.get(kind, ()): + value = spec.get(ref) + if required and not value: + return None, f"a {kind} view needs {ref}", "missing_ref" + if value and by_key.get(value) not in families: + return None, f"{ref} has the wrong field type", "wrong_ref_type" + if value: + display[ref] = value + cleaned_display = _grid()._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", "display_dropped" + + return { + "kind": kind, + # The view NAME is model-authored too, and it is the string that ends up in the rail, in + # the flyout and on the artefact card. R6 reaches it here or nowhere. + "name": (_said(spec.get("name")) or "Query")[:60], + "visible": visible, + "filters": filters, + "filterConj": "or" if spec.get("filterConj") == "or" else "and", + "sorts": [{"colId": item["colId"], "dir": "desc" if item.get("dir") == "desc" else "asc"} + for item in (spec.get("sorts") or []) if isinstance(item, dict) + and item.get("colId") in keys][:3], + "groupBy": spec.get("groupBy") if spec.get("groupBy") in keys else None, + "aggregation": {"op": op, "field": field}, + "display": cleaned_display, + # ⚠ `is True`, not truthy, and UNCONDITIONAL — the same two rules `grid_events.view_upsert` + # follows for this key. `is True` so a model emitting the string "false" does not mark a + # view; unconditional so the mark is REMOVABLE rather than a flag that can be set and never + # cleared (a key written only when present leaves a stored `true` alive forever). + "important": spec.get("important") is True, + }, None, None + + +def _explain(view, fields): + label = {field["key"]: str(field.get("label") or field["key"]) for field in fields} + visible = ", ".join(label.get(key, key) for key in view["visible"][:6]) + result = f"{view['kind']} view of {visible}" + if view["filters"]: + result += "; filtered records only" + if view.get("groupBy"): + result += f"; grouped by {label.get(view['groupBy'], view['groupBy'])}" + agg = view["aggregation"] + if agg["op"] != "count": + result += f"; {agg['op']} of {label.get(agg['field'], agg['field'])}" + return result + "." + + +def _said_fallback(artifact): + """What a build turn SAYS when the provider volunteered no sentence of its own. + + ⚠ MEASURED, 2026-08-16, cerebras: a view-shaped question comes back as a tool call with + `content: null`. So this is the sentence a person reads on MOST build turns, not a rare + fallback, and R16 ("the AI chat must TALK with the user") is decided here rather than in the + prompt. `_explain` used to be it, and it is a receipt for our own output: *"grid view of + Company, Owner, Deal value; filtered records only."* It keeps its real job ON THE ARTEFACT, + where it labels the thing it describes. + """ + numeric = artifact.get("numeric") or {} + value = numeric.get("value") + if value is None: + return f"I built {artifact['name']}." + if isinstance(value, float) and value.is_integer(): + value = int(value) + number = f"{value:,}" if isinstance(value, (int, float)) else str(value) + label = " ".join(str(numeric.get("label") or "Records").split()) + return f"I built {artifact['name']}. {label}: {number}." + + +def _numeric_result(view, records): + aggregation = view["aggregation"] + if aggregation["op"] == "count": + return {"label": "Matching records", "value": len(records), "contributing_record_count": len(records)} + values = [] + for record in records: + try: + value = record.get(aggregation["field"]) + if value is not None and not isinstance(value, bool): + values.append(float(value)) + except (TypeError, ValueError): + continue + if not values: + return {"label": aggregation["op"], "value": None, "contributing_record_count": 0} + op = aggregation["op"] + value = {"sum": sum(values), "avg": sum(values) / len(values), "min": min(values), "max": max(values)}[op] + return {"label": f"{op.title()} of {aggregation['field']}", "value": value, + "contributing_record_count": len(values)} + + +def _referenced_fields(view): + """The citation names every source field that affected the displayed result.""" + out = list(view.get("visible") or ()) + for node in view.get("filters") or (): + if isinstance(node, dict) and node.get("colId"): + out.append(str(node["colId"])) + for node in view.get("sorts") or (): + if isinstance(node, dict) and node.get("colId"): + out.append(str(node["colId"])) + for key in ("groupBy",): + if view.get(key): + out.append(str(view[key])) + aggregation = view.get("aggregation") or {} + if aggregation.get("field"): + out.append(str(aggregation["field"])) + return list(dict.fromkeys(out)) + + +def _effective_filters(snapshot, view): + """Keep the source request and generated-view predicates distinct in provenance.""" + return { + "source": _safe(snapshot.get("filters")), + "view": {"conj": view.get("filterConj", "and"), + "nodes": _safe(view.get("filters") or [])}, + } + + +def _view_records(snapshot, view): + """Apply the exact validated virtual-view filter before calculating a cited number.""" + nodes = view.get("filters") or [] + if not nodes: + return list(snapshot["records"]) + from harness import filter_eval + tree = {"conj": view.get("filterConj", "and"), "nodes": nodes} + return [row for row in snapshot["records"] + if filter_eval.matches(tree, row, snapshot["fields"])] + + +def _citation(citation_id, snapshot, view, numeric, view_id): + return { + "id": citation_id, + "href": f"#/query?view={view_id}&citation={citation_id}", + "database": snapshot["database"], + "snapshot": {"kind": snapshot["source_kind"], "version": _safe(snapshot["source_version"])}, + "fields": [field["key"] for field in snapshot["fields"] + if field.get("key") in set(_referenced_fields(view))], + "filters": _effective_filters(snapshot, view), + "permission_scope_applied": snapshot["permission_scope_applied"], + "aggregation": _safe(view["aggregation"]), + "contributing_record_count": numeric["contributing_record_count"], + "retrieved_at": snapshot["retrieved_at"], + } + + +def _citation_complete(citation): + """A numeric result is not publishable without complete provenance.""" + required = {"database", "snapshot", "fields", "filters", "aggregation", + "contributing_record_count", "retrieved_at", "href"} + snapshot = citation.get("snapshot") if isinstance(citation, dict) else None + return (isinstance(citation, dict) and required <= set(citation) + and bool(citation["database"]) and isinstance(snapshot, dict) + and "version" in snapshot and bool(citation["retrieved_at"]) + and isinstance(citation["fields"], list) + and isinstance(citation["aggregation"], dict)) + + +def _public_view(view): + source = view["source"] + return { + "id": view["id"], "viewId": view["id"], "scope": source["database"], + "name": view["name"], "description": str(view.get("description") or ""), + "kind": view["view"]["kind"], "question": view["question"], + "explain": view["explain"], "threadId": view["threadId"], "createdAt": view["createdAt"], + "virtual": True, "source": _safe(source), "view": _safe(view["view"]), + "citationIds": list(view["citationIds"]), "numeric": _safe(view["numeric"]), + } + + +def _source_still_permitted(session, source, permitted=None): + """A persisted artefact never outlives the caller's current data permission. + + ⚠ `permitted` is the BATCH answer from `deps.assistant_source_status` — one rows-free + resolution for every source at once, rather than one whole-document read per saved view. That + function's docstring carries the measurement. The single-source path below stays for callers + holding exactly one artefact (the workspace-event door), and asks the same helper. + + ⛔ `permitted`, NEVER `answerable`. A source whose rows are served through the connector mirror + cannot be ASKED and can still be SEEN: the artefact was built from a snapshot that was legal + when it was taken, and hiding it because the reader can no longer make a NEW one would read as + deletion. The two verdicts are separate for that reason. + """ + database = (source or {}).get("database") if isinstance(source, dict) else None + if not database: + return False + if permitted is None: + permitted = {key for key, row in assistant_source_status(session, [database]).items() + if row["permitted"]} + return database in permitted + + +def _public_state(state, session): + # ONE rows-free batch answers both questions: which sources this caller may still see (which + # artefacts stay listed) and which of them can actually be asked (which chips are live). + status = assistant_source_status(session) + for row in state["views"].values(): + key = (row.get("source") or {}).get("database") if isinstance(row.get("source"), dict) else None + if key and key not in status: + # An artefact whose source is no longer enumerable still gets a verdict rather than a + # KeyError — it resolves to "not permitted" and the artefact drops out, which is the + # same answer the per-view wall gave. + status.update(assistant_source_status(session, [key])) + permitted = {key for key, row in status.items() if row["permitted"]} + allowed_views = [row for row in state["views"].values() + if _source_still_permitted(session, row.get("source"), permitted)] + allowed_ids = {row["id"] for row in allowed_views} + allowed_citations = {citation_id for row in allowed_views + for citation_id in row.get("citationIds") or []} + threads = sorted(state["threads"].values(), key=lambda row: row.get("updatedAt", ""), reverse=True) + messages = sorted(state["messages"].values(), key=lambda row: row.get("createdAt", "")) + views = sorted((_public_view(row) for row in allowed_views), key=lambda row: row["createdAt"], reverse=True) + messages = [row for row in messages if not row.get("viewId") or row.get("viewId") in allowed_ids] + return {"threads": _safe(threads), "messages": _safe(messages), "views": views, + "citations": _safe([row for key, row in state["citations"].items() + if key in allowed_citations]), "models": model_choices(), + # The model picker's own data, the same shape `sources` uses for databases. + "modelStatus": model_status(), + # The chip row's own data: a source this caller holds but cannot ask, and WHY. + # ⭐⭐ D-276 — `or row.get("visible")`. A source the caller can open elsewhere in the + # product but the assistant cannot read is now LISTED with its cause, instead of + # vanishing from a picker that shows every other database they hold. It arrives with + # `answerable: False` and a `reason`, which is the SAME shape the mirror-served grids + # already use, so the chip row needs no new state to render it. + # ⛔ `permitted` STILL GATES ARTEFACTS — `allowed_views` above is unchanged. This widens + # what is DESCRIBED, never what can be read or kept. + "sources": _safe([{"database": key, "answerable": row["answerable"], + "reason": row["reason"]} + for key, row in sorted(status.items()) + if row["permitted"] or row.get("visible")])} + + +@router.get("/query") +def list_queries(session: Session = Depends(require_session)): + return _public_state(_state(session), session) + + +@router.post("/query/{qid}/events") +def mutate_query_workspace(qid: str, body: dict = Body(default=None), + session: Session = Depends(require_session)): + """The only grid-mutation transport for a virtual Query workspace. + + Query artefacts are immutable snapshots: creation and changes belong to an Assistant prompt, + never a source-native grid workspace. Deleting the caller's personal artefact is the one + permitted mutation. The binding key, not a client-supplied source scope, identifies it. + """ + qid = str(qid) + body = body if isinstance(body, dict) else {} + binding = body.get("workspaceBinding") + event = body.get("event") + if not (isinstance(binding, dict) and binding.get("kind") == "query" + and str(binding.get("key") or "") == qid): + raise err(400, "query_workspace_mismatch", + "the Query workspace binding must name this virtual artefact") + if not isinstance(event, dict) or str(event.get("type") or "") not in QUERY_WORKSPACE_EVENTS: + raise err(400, "unsupported_query_workspace_event", + "Query accepts only view_create, view_upsert, or view_delete events") + + state = _state(session) + view = state["views"].get(qid) + if view is None or not _source_still_permitted(session, view.get("source")): + # A caller cannot use an opaque Query key to learn about a revoked artefact. + raise err(404, "unknown_query", "that Query artefact does not exist") + + event_type = str(event["type"]) + if event_type != "view_delete": + raise err(409, "query_workspace_immutable", + "AI-created Query views change only through a new Assistant prompt") + if str(event.get("viewId") or "") != qid: + raise err(400, "query_workspace_mismatch", + "a Query delete must name the same virtual artefact as its binding") + + deleted = delete_query(qid, session) + return {"workspaceBinding": {"kind": "query", "key": qid}, + "event": event_type, **deleted} + + +def _sources(body, target, session): + raw = body.get("sources") if isinstance(body, dict) else None + sources = [str(item).strip() for item in raw] if isinstance(raw, list) else [] + sources = list(dict.fromkeys(item for item in sources if item)) + if target not in sources: + sources.insert(0, target) + # Source chips are a permissioned selection, not untrusted labels remembered in a thread. + # The model still receives ONLY `target`'s snapshot below; these calls do not pass a second + # source to it and D's helper refuses cache misses, mirrors and unshared databases. + for source in sources: + assistant_read_scope(session, source, fields=[], filters=None) + return sources + + +def _submit(body, session, chat=None): + body = body if isinstance(body, dict) else {} + question = " ".join(str(body.get("question") or "").split()) + target = str(body.get("database") or body.get("scope") or "").strip() + selected_model = str(body.get("model") or MODEL_AUTO).strip().lower() + 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") + if not target: + raise err(400, "bad_request", "one target database must be selected") + if selected_model not in model_choices(): + raise err(400, "unknown_model", "that model is not available in Query") + + # This call happens before model selection and is the only permitted source-data read. + snapshot = assistant_read_scope(session, target, fields=None, filters=body.get("filters")) + sources = _sources(body, target, session) + state = _state(session) + thread_id = str(body.get("threadId") or "").strip() + if thread_id and thread_id not in state["threads"]: + raise err(404, "unknown_thread", "that chat does not exist") + if not thread_id: + thread_id = _new_id("thread") + now = _now_iso() + user_message_id = _new_id("message") + assistant_message_id = _new_id("message") + # R16: the turns already on screen go with the question. `state` was read above, BEFORE this + # turn's own messages exist, so the replay is strictly the prior conversation. + spec, said, provider, reason = _call_model(question, snapshot, model=selected_model, chat=chat, + history=_history(state, thread_id)) + view, refusal, refusal_code = _validate(spec, snapshot["fields"]) if spec is not None else (None, said, reason) + + artifact = None + citation = None + if view is not None: + if len(state["views"]) >= MAX_ARTIFACTS: + raise err(400, "query_limit", f"you already have {len(state['views'])} Query artefacts") + view_id = _new_id("query") + citation_id = _new_id("citation") + numeric = _numeric_result(view, _view_records(snapshot, view)) + citation = _citation(citation_id, snapshot, view, numeric, view_id) + if not _citation_complete(citation): + raise RuntimeError("Query refused to persist an incomplete numeric citation") + source = {"database": snapshot["database"], "label": snapshot["label"], + "source_kind": snapshot["source_kind"], "source_version": _safe(snapshot["source_version"]), + "retrieved_at": snapshot["retrieved_at"], + "fields": _safe(snapshot["fields"]), "filters": _safe(snapshot["filters"]), + "permission_scope_applied": snapshot["permission_scope_applied"]} + artifact = {"id": view_id, "threadId": thread_id, "name": view["name"], "description": "", + "question": question, + "explain": _explain(view, snapshot["fields"]), "createdAt": now, "source": source, + "view": _safe(view), "citationIds": [citation_id], "numeric": numeric, + "model": provider, "requestedModel": selected_model} + + assistant_message = { + "id": assistant_message_id, "threadId": thread_id, "role": "assistant", "createdAt": now, + "content": (said or _said_fallback(artifact)) if artifact else (refusal or "the assistant could not answer"), + "targetDatabase": target, "requestedModel": selected_model, "model": provider, + "reason": refusal_code, "viewId": artifact["id"] if artifact else None, + "citationIds": artifact["citationIds"] if artifact else [], "numeric": artifact["numeric"] if artifact else None, + } + user_message = {"id": user_message_id, "threadId": thread_id, "role": "user", "createdAt": now, + "content": question, "sources": sources, "targetDatabase": target, + "requestedModel": selected_model} + + def update(raw): + current = _blank_state() + if isinstance(raw, dict): + for key in ("threads", "messages", "citations", "views"): + if isinstance(raw.get(key), dict): + current[key] = copy.deepcopy(raw[key]) + thread = current["threads"].get(thread_id) or {"id": thread_id, "createdAt": now} + # ⚠ THE FIRST QUESTION NAMES THE CHAT, and R16 is what makes that matter. While every turn + # was a fresh one-shot the title could only be the last thing asked; now that a thread is a + # conversation, retitling it on every follow-up renames the history entry out from under + # the reader, and "and the other one?" is a useless name for anything. + thread.update({"updatedAt": now, "title": thread.get("title") or question[:80], "sources": sources, + "model": selected_model, "activeViewId": artifact["id"] if artifact else thread.get("activeViewId")}) + current["threads"][thread_id] = thread + current["messages"][user_message_id] = user_message + current["messages"][assistant_message_id] = assistant_message + if artifact: + current["views"][artifact["id"]] = artifact + current["citations"][citation["id"]] = citation + return current + + if not session.runtime.available(): + raise err(503, "store_unavailable", "the tenant store is unavailable; no chat was saved") + session.runtime.update(_namespace_key(session), update, flush="async") + return {"thread": _safe(update(state)["threads"][thread_id]), "userMessage": _safe(user_message), + "message": _safe(assistant_message), + "view": _public_view(artifact) if artifact else None, "citations": [citation] if citation else []} + + +@router.post("/query/chat") +def submit_chat(body: dict = Body(default=None), session: Session = Depends(require_session)): + return _submit(body, session) + + +@router.delete("/query/threads/{tid}") +def delete_thread(tid: str, session: Session = Depends(require_session)): + """Remove one of the caller's own chats, and only the chat. + + ⛔ THE ARTEFACTS SURVIVE, AND THAT IS THE POINT. A Query view is a durable object in its own + right — it appears in Query's rail, other people's links can point at it, and the owner's + instruction was that AI views LIVE THERE rather than inside the conversation that happened to + produce them. Cascading the delete would make tidying up a chat silently destroy work. Views + are deleted from Query's own rail, one at a time, by `delete_query` below. + + ⚠ It exists because the history had no prune. `delete_query` removed a view and left its + thread, so the panel could only ever grow — a navigation you cannot manage is half a + navigation, and a person clearing test chats found that out first. + """ + tid = str(tid) + state = _state(session) + if tid not in state["threads"]: + raise err(404, "unknown_thread", "that chat does not exist") + + def update(raw): + current = _blank_state() + if isinstance(raw, dict): + for key in ("threads", "messages", "citations", "views"): + if isinstance(raw.get(key), dict): + current[key] = copy.deepcopy(raw[key]) + current["threads"].pop(tid, None) + for message_id in [key for key, row in current["messages"].items() + if (row or {}).get("threadId") == tid]: + current["messages"].pop(message_id, None) + # ⚠ THE VIEWS' `threadId` IS LEFT EXACTLY AS IT WAS, dangling. Blanking it looks tidier and + # is a data-loss bug: `queryApi.saved()` refuses a row with an empty `threadId`, so every + # artefact built in the deleted chat would silently vanish from Query's rail — the exact + # "tidying up destroys work" outcome this door is written not to have + # [[a-record-can-outlive-its-subject]]. Nothing resolves the id but the chat panel, which + # only ever looks up the thread it is showing. + return current + + session.runtime.update(_namespace_key(session), update, flush="async") + return {"deleted": tid} + + +@router.delete("/query/{qid}") +def delete_query(qid: str, session: Session = Depends(require_session)): + qid = str(qid) + state = _state(session) + if qid not in state["views"]: + raise err(404, "unknown_query", "that Query artefact does not exist") + + def update(raw): + current = _blank_state() + if isinstance(raw, dict): + for key in ("threads", "messages", "citations", "views"): + if isinstance(raw.get(key), dict): + current[key] = copy.deepcopy(raw[key]) + view = current["views"].pop(qid, None) + for citation_id in (view or {}).get("citationIds") or []: + current["citations"].pop(citation_id, None) + return current + + session.runtime.update(_namespace_key(session), update, flush="async") + return {"deleted": qid} + + +# ══ R20 — A QUERY VIEW HAS EVERY FUNCTION A DATABASE VIEW HAS ═══════════════════════════════════ +# +# ⛔⛔ THE DECISION THIS TICKET ASKED FOR, WRITTEN DOWN RATHER THAN IMPLIED, because it is the line +# every later reader will need and there is no other place it exists. +# +# **IMMUTABLE, and unchanged by R20:** the generated SPEC and its provenance. `view` (kind, visible, +# filters, sorts, aggregation, display, important), `source` (the snapshot, its version, its +# retrieval time, the permission scope), `question`, `citationIds`, `numeric`. A cited number is +# only worth citing if the thing it was computed from cannot be edited underneath it, so +# `mutate_query_workspace` still answers 409 `query_workspace_immutable` for `view_create` and +# `view_upsert`, and `QUERY_MUTATION_POLICY` on the client is untouched. +# +# **OPENED:** `name` and `description`. They are LABELS, not the answer: changing them cannot make +# a citation wrong. Plus a server-side DUPLICATE (the client never supplies a spec, so a copy is a +# copy) and an EXPORT, which is a read. +# +# ⛔ AND THE REASON THESE ARE NAMED DOORS RATHER THAN AN OPENED `view_upsert`, which is what the +# ticket's `how:` first suggested: the client refuses `view_upsert` on a Query binding LOCALLY, +# before any request leaves the browser (`queryPreview.ts::routeQueryViewMutation`). Opening the +# server there would give one question two answers, and the client's is the one a person +# experiences. A named door has exactly one answer, and "opening a route widens every field it +# carries unless the cleaner is explicit" is answered by the allow-lists below rather than by hope. + +QUERY_NOTE_MAX = 400 + + +def _artifact_or_404(session, qid, state=None): + """The one wall every artefact door goes through: it exists AND this caller may still see it.""" + state = _state(session) if state is None else state + row = state["views"].get(str(qid)) + if row is None or not _source_still_permitted(session, row.get("source")): + # A caller cannot use an opaque Query key to learn about a revoked artefact. + raise err(404, "unknown_query", "that Query artefact does not exist") + return state, row + + +def _copy_name(name, taken): + """`X copy`, then `X copy 2`, so duplicating twice does not make two rows with one name.""" + base = f"{str(name or 'Query')[:52]} copy" + if base not in taken: + return base[:60] + index = 2 + while f"{base} {index}" in taken and index < 99: + index += 1 + return f"{base} {index}"[:60] + + +@router.patch("/query/{qid}") +def rename_query(qid: str, body: dict = Body(default=None), + session: Session = Depends(require_session)): + """Rename a Query view, or give it a description. Nothing else, by construction. + + ⚠ THE TEXT IS THE READER'S, SO IT IS NOT SWEPT FOR DASHES. R6 governs the copy WE write; a note + somebody typed about their own view is theirs, and rewriting its punctuation would be the + product editing a person's words. Model-authored strings are a different case and are swept + where they are produced (`_said`). + """ + body = body if isinstance(body, dict) else {} + state, row = _artifact_or_404(session, qid) + # An ALLOW-LIST, read key by key. A body carrying `view` or `source` changes neither. + patch = {} + if "name" in body: + patch["name"] = " ".join(str(body.get("name") or "").split())[:60] or row.get("name") or "Query" + if "description" in body: + patch["description"] = " ".join(str(body.get("description") or "").split())[:QUERY_NOTE_MAX] + if not patch: + raise err(400, "bad_request", "a Query view takes a new name or a new description") + + def update(raw): + current = _blank_state() + if isinstance(raw, dict): + for key in ("threads", "messages", "citations", "views"): + if isinstance(raw.get(key), dict): + current[key] = copy.deepcopy(raw[key]) + if str(qid) in current["views"]: + current["views"][str(qid)].update(patch) + return current + + if not session.runtime.available(): + raise err(503, "store_unavailable", "the tenant store is unavailable; nothing was renamed") + session.runtime.update(_namespace_key(session), update, flush="async") + return _public_view(update(state)["views"][str(qid)]) + + +@router.post("/query/{qid}/duplicate") +def duplicate_query(qid: str, session: Session = Depends(require_session)): + """A second artefact with the same spec and its own identity. + + ⛔ THE CITATIONS ARE COPIED, NOT SHARED. `delete_query` pops an artefact's citations by id, so a + copy pointing at the original's citation ids would lose its provenance the moment the original + was deleted, and a number with no citation is the one thing this module refuses to persist. + """ + state, row = _artifact_or_404(session, qid) + if len(state["views"]) >= MAX_ARTIFACTS: + # R6's second sentence: a cap that cannot be lifted here names its cause and the remedy. + raise err(400, "query_limit", + f"you already have {len(state['views'])} Query views, which is the most one " + f"account can hold. Delete one to make room for this copy.") + new_id = _new_id("query") + copied = copy.deepcopy(row) + copied["id"] = new_id + copied["name"] = _copy_name(row.get("name"), {r.get("name") for r in state["views"].values()}) + copied["createdAt"] = _now_iso() + citations = {} + for citation_id in row.get("citationIds") or []: + source = state["citations"].get(citation_id) + if not source: + continue + fresh_id = _new_id("citation") + citation = copy.deepcopy(source) + citation["id"] = fresh_id + citation["href"] = f"#/query?view={new_id}&citation={fresh_id}" + citations[fresh_id] = citation + copied["citationIds"] = list(citations) + + def update(raw): + current = _blank_state() + if isinstance(raw, dict): + for key in ("threads", "messages", "citations", "views"): + if isinstance(raw.get(key), dict): + current[key] = copy.deepcopy(raw[key]) + current["views"][new_id] = copied + current["citations"].update(citations) + return current + + if not session.runtime.available(): + raise err(503, "store_unavailable", "the tenant store is unavailable; nothing was copied") + session.runtime.update(_namespace_key(session), update, flush="async") + return _public_view(copied) + + +# ⛔ A REGISTRY KEY AND A GRID SCOPE ARE DIFFERENT SPELLINGS OF ONE DATABASE, and this product has +# already been bitten by treating them as the same: a `VIEW_OPEN` emit whose topic did not match was +# dropped SILENTLY, 200 OK, and an alert was filed against the wrong database. `customer_data` is +# what the assistant reads; `customer` is what the grid writes. `routes_grid._scope_or_400` refuses +# an unknown scope rather than defaulting it, so this map has to be right rather than nearly right. +_GRID_SCOPE = {"customer_data": "customer", "product_data": "product"} +# The keys a Query spec and a database view actually share. `aggregation` is deliberately absent: +# a database view has no aggregation of its own, which is why the reply below NAMES it as dropped +# instead of letting a "sum of value" answer land as a plain list of rows. +_SAVEABLE = ("visible", "filters", "filterConj", "sorts", "groupBy", "display", "important") + + +def _grid_scope(database): + key = str(database or "").strip() + return _GRID_SCOPE.get(key) or (key if key.startswith("ut_") else "") + + +@router.post("/query/{qid}/save-to-database") +def save_query_to_database(qid: str, session: Session = Depends(require_session)): + """R20: copy this answer into the database's OWN view list, where an ordinary view lives. + + ⛔⛔ `view_upsert` ANSWERS 200 WHILE WRITING NOTHING, IN TWO DIFFERENT WAYS, AND THE WIRE CANNOT + TELL YOU WHICH. `rerender: False` is the documented SUCCESS shape for this event, and it is + also what several silent-refusal branches return: a malformed payload, a missing id or name, an + id that belongs to a view the caller cannot SEE (the wave-9 shared-view guard, which makes + pinned ids effectively tenant-scoped), and a shared view they may see but not edit. So this + door does not believe the response. It READS THE STORE BACK and refuses out loud if the view is + not there. That read is same-process, which is what makes it valid despite `flush="async"`: + the write is visible through the cache long before it is durable. + + ⭐ ONE REQUEST, ONE EVENT. Eighteen POSTs against one JSON document under a coalescing + single-flight once landed ZERO while answering 200 eighteen times; the fix is a batch, never a + retry loop, which treats the symptom and doubles the races. + + ⚠ A FRESH ID, NEVER THE ARTEFACT'S. Reusing the Query id would be a guessable id in a shared + bucket, which is the exact door the wave-9 guard exists to shut. + """ + from routes_grid import grid_events_route + + state, row = _artifact_or_404(session, qid) + scope = _grid_scope((row.get("source") or {}).get("database")) + if not scope: + raise err(409, "unsaveable_source", + "this answer's database does not have a view list to save into.") + spec = row.get("view") or {} + config = {key: copy.deepcopy(spec[key]) for key in _SAVEABLE if spec.get(key) is not None} + config["important"] = spec.get("important") is True + view_id = _new_id("view") + name = " ".join(str(row.get("name") or "Query").split())[:120] or "Query" + + grid_events_route({"scopeKey": scope, "events": [{ + "id": _new_id("event"), "type": "view_upsert", + "view": {"id": view_id, "name": name, "config": config}, + }]}, session) + + document = session.runtime.get(f"{scope}_table_workspace") or {} + landed = (((document.get(session.uname) or {}).get("views") or {}).get(view_id) + if isinstance(document, dict) else None) + if not isinstance(landed, dict): + raise err(409, "view_not_saved", + "the database did not accept this view, so nothing was saved. Open the database " + "and check you can still add a view there.") + + # R6's second sentence, in the payload: what the database's own validator would not take is + # NAMED rather than quietly missing from a view that then looks complete. + stored = landed.get("config") or {} + dropped = [] + if str((spec.get("aggregation") or {}).get("op") or "count") != "count": + dropped.append("the aggregation, which a database view does not carry") + if len(stored.get("filters") or []) != len(config.get("filters") or []): + dropped.append("some of the conditions") + for key in ("visible", "sorts", "groupBy", "display"): + if config.get(key) and not stored.get(key): + dropped.append(key) + return {"scope": scope, "viewId": view_id, "name": landed.get("name") or name, + "dropped": dropped} + + +@router.post("/query/messages/{mid}/rating") +def rate_message(mid: str, body: dict = Body(default=None), + session: Session = Depends(require_session)): + """R20's thumbs, and the ONE thing that stops them being decoration. + + ⛔ A RATING THAT IS WRITTEN AND NEVER READ IS A FLAG SHIPPED WITHOUT ITS WRITER. This one is + read: `_history` turns a thumbs-down into a user turn saying the answer was not helpful, with + the reason if one was given, so the next question in the same thread is answered against that. + The scope is honest and narrow: it steers THIS conversation. It does not train anything, does + not cross threads, and does not cross accounts. + + ⚠ Rating is idempotent and CLEARABLE (`rating: null`), because a mis-click that cannot be taken + back is worse than no control at all, and a flag that can be set and never cleared is the same + defect `important` was fixed for. + """ + body = body if isinstance(body, dict) else {} + mid = str(mid) + raw = body.get("rating") + rating = str(raw or "").strip().lower() + if raw is not None and rating not in QUERY_RATINGS: + raise err(400, "bad_rating", "a rating is up, down, or nothing at all") + state = _state(session) + row = state["messages"].get(mid) + if row is None: + raise err(404, "unknown_message", "that message does not exist") + if row.get("role") != "assistant": + raise err(400, "bad_rating", "only an answer can be rated") + reason = " ".join(str(body.get("reason") or "").split())[:MAX_RATING_REASON] + + def update(raw_state): + current = _blank_state() + if isinstance(raw_state, dict): + for key in ("threads", "messages", "citations", "views"): + if isinstance(raw_state.get(key), dict): + current[key] = copy.deepcopy(raw_state[key]) + message = current["messages"].get(mid) + if message is not None: + message["rating"] = rating if raw is not None else None + # A reason belongs to a thumbs-down; clearing the rating clears it with them. + message["ratingReason"] = reason if (raw is not None and rating == "down") else "" + return current + + if not session.runtime.available(): + raise err(503, "store_unavailable", "the tenant store is unavailable; nothing was recorded") + session.runtime.update(_namespace_key(session), update, flush="async") + return _safe(update(state)["messages"][mid]) + + +@router.get("/query/{qid}/export") +def export_query(qid: str, session: Session = Depends(require_session)): + """The artefact's own rows and columns, as CSV. + + ⚠ THE ARTEFACT IS FROZEN; ITS DATA IS NOT, and the difference is worth stating because the two + are easy to conflate. This re-reads the source through the SAME permission wall that built the + artefact and applies the view's own stored predicate, so the file matches what the grid is + showing right now, not the snapshot the stored `numeric` was computed from. That is the useful + answer: a person exports what they are looking at. + """ + state, row = _artifact_or_404(session, qid) + source = row.get("source") or {} + # ⛔ THIS PASSES A STORED OUTPUT BACK INTO AN INPUT SLOT, WHICH IS WORTH STATING RATHER THAN + # HIDING. `source["filters"]` is `deps.assistant_read_scope`'s OWN return value, and that value + # is `validate_assistant_filter(filters, visible)` — the VALIDATED form of whatever the caller + # asked for. MEASURED: `queryApi.submitChat` sends no `filters` key at all, so in production + # this is `None` on every artefact and the round trip never happens. + # ⚠ It is sent anyway, and deliberately, because the alternative FAILS OPEN. If a caller ever + # does narrow a source, dropping the filter here would export MORE rows than the artefact was + # built from, silently. Sending it means a stored filter that cannot be re-validated comes back + # as a NAMED 400 (`assistant_bad_filter`) instead — the fail-closed direction. The gate asserts + # what this door PASSES rather than what the fixture echoes back, because a double built from + # the producer would make this argument valid by construction. + snapshot = assistant_read_scope(session, source.get("database"), fields=None, + filters=source.get("filters") or None) + labels = {field["key"]: str(field.get("label") or field["key"]) for field in snapshot["fields"]} + columns = [key for key in (row["view"].get("visible") or ()) if key in labels] + if not columns: + raise err(409, "nothing_to_export", + "none of this view's columns exist in the database any more, so there is " + "nothing to export. Ask the assistant to build it again.") + buffer = io.StringIO() + writer = csv.writer(buffer, lineterminator="\n") + writer.writerow([labels[key] for key in columns]) + for record in _view_records(snapshot, row["view"]): + writer.writerow(["" if record.get(key) is None else record.get(key) for key in columns]) + stem = re.sub(r"[^A-Za-z0-9]+", "_", str(row.get("name") or "query")).strip("_") or "query" + return Response( + content=buffer.getvalue(), media_type="text/csv; charset=utf-8", + # ⚠ ASCII-only filename: a header carrying a non-ASCII byte is refused by some servers and + # silently mangled by others, and the stem above comes from a model-authored name. + headers={"Content-Disposition": f'attachment; filename="{stem[:60]}.csv"'})