"""harness/views.py — saved views + spawned dashboards: the store and the RE-RUNNER (OM-3 viewer). The Analyst's save_view/compose_dashboard tools persist chart SPECS + their semantic QUERY (never rows) into data/store/views.json. This module is the read side: load the specs, re-execute each view's governed query against the tenant store at render time, and hand the page a fresh artifact dict in exactly the shape `_render_analyst_artifact` already renders. The viewer therefore shows LIVE (store-fresh) numbers, not the numbers from whenever the view was saved — same discipline as every dashboard (a saved view is a query, not a snapshot). Legacy note: views saved before 2026-07-11 carry no 'query' (only a session-scoped result_id) and cannot be re-run — run_view raises a readable error the page surfaces per view. """ import json import time import harness.semantic as SEM from harness.tools import VIEWS_PATH _QUERY_KEYS = ("topic", "measures", "group_by", "grain", "date_from", "date_to", "team_id", "filters", "sort", "limit", "exclude_services") def load(): if VIEWS_PATH.exists(): return json.loads(VIEWS_PATH.read_text(encoding="utf-8")) return {"views": {}, "dashboards": {}} def _save(d): VIEWS_PATH.parent.mkdir(parents=True, exist_ok=True) VIEWS_PATH.write_text(json.dumps(d, indent=1), encoding="utf-8") def views(): return load().get("views", {}) def dashboards(): return load().get("dashboards", {}) def delete_view(name): d = load() d.get("views", {}).pop(name, None) for dash in d.get("dashboards", {}).values(): dash["views"] = [v for v in dash.get("views", []) if v != name] _save(d) def delete_dashboard(name): d = load() d.get("dashboards", {}).pop(name, None) _save(d) # ------------------------------------------------------------------ UPDATE (owner directive # 2026-07-12: the Analyst can CREATE **and UPDATE** each workbook in the workspace) _QUERY_PATCH_KEYS = ("measures", "group_by", "grain", "date_from", "date_to", "team_id", "filters", "sort", "limit", "exclude_services", "topic") _SPEC_PATCH_KEYS = ("title", "kind", "x", "y", "series", "metric", "y2", "size", "value", "facet", "columns", "transforms") def update_view(name, changes): """Patch a saved view's chart spec and/or query, VALIDATED BY RE-EXECUTION before saving — an update that cannot run does not land. `changes` may contain spec keys (title/kind/x/y/ series/metric) and/or a `query` dict of query-key patches (a None value REMOVES the key).""" d = load() v = d.get("views", {}).get(name) if not v: raise SEM.ModelError(f"unknown view {name!r} (use list_workspace)") spec = dict(v.get("chart") or {}) q = dict(spec.get("query") or {}) if not q: raise SEM.ModelError(f"view {name!r} carries no query — rebuild it instead") changes = dict(changes or {}) qpatch = changes.pop("query", None) or {} bad = [k for k in changes if k not in _SPEC_PATCH_KEYS] bad += [k for k in qpatch if k not in _QUERY_PATCH_KEYS] if bad: raise SEM.ModelError(f"unknown patch keys {bad} (spec: {list(_SPEC_PATCH_KEYS)}; " f"query: {list(_QUERY_PATCH_KEYS)})") for k, val in qpatch.items(): if val is None: q.pop(k, None) else: q[k] = val res = _run_query(q) # validation: it must RUN new_spec = {**spec, **changes, "query": q} rows = res["rows"] if new_spec.get("transforms"): # the transform chain must replay too import harness.transforms as TR rows, _ = TR.apply({**res, "query": q}, new_spec["transforms"], run_query=_run_query) cols = set(rows[0]) if rows else set() for ref in ("x", "y", "series", "y2", "size", "value", "facet"): if new_spec.get(ref) and cols and new_spec[ref] not in cols: raise SEM.ModelError(f"{ref}={new_spec[ref]!r} not in the patched result columns " f"{sorted(cols)} — patch {ref} too") d["views"][name] = {"chart": new_spec, "saved_at": time.strftime("%Y-%m-%d %H:%M")} _save(d) return {"updated": name, "row_count": res["row_count"], "query": q, "note": "patched query re-ran successfully before saving"} def update_dashboard(name, views_list=None, new_name=None): """Update a workbook (dashboard): recompose its views (order = display order) and/or rename it. Renames follow through to scheduled reports that reference it.""" d = load() dash = d.get("dashboards", {}).get(name) if dash is None: raise SEM.ModelError(f"unknown dashboard {name!r} (use list_workspace)") if views_list is not None: missing = [v for v in views_list if v not in d.get("views", {})] if missing: raise SEM.ModelError(f"unknown views {missing} — save_view them first") dash["views"] = list(views_list) dash["updated_at"] = time.strftime("%Y-%m-%d %H:%M") if new_name and new_name != name: if new_name in d.get("dashboards", {}) or new_name in d.get("views", {}): raise SEM.ModelError(f"{new_name!r} already exists — pick another name") d["dashboards"][new_name] = d["dashboards"].pop(name) try: # follow the rename into schedules import harness.routines as R rt = R.load() if name in rt.get("reports", {}): rt["reports"][new_name] = rt["reports"].pop(name) R._save(rt) except Exception: pass name = new_name _save(d) return {"dashboard": name, "views": d["dashboards"][name]["views"]} def _run_query(q): return SEM.store_query(**{k: q.get(k) for k in _QUERY_KEYS if q.get(k) is not None}) def run_view(name): """Re-execute one saved view's governed query — then REPLAY its recorded transform chain — and return a fresh artifact dict ({'chart': {..., 'rows': [...]}}, {'table': {...}} or {'kpi': {...}}) for the house renderer.""" v = views().get(name) if not v: raise SEM.ModelError(f"unknown view {name!r}") spec = v.get("chart") or {} q = spec.get("query") if not q: raise SEM.ModelError(f"view {name!r} was saved without its query (pre-2026-07-11) — " "ask the Analyst to rebuild and re-save it") res = _run_query(q) if spec.get("kind") == "kpi" or (spec.get("metric") and spec.get("kind") != "table"): metric = spec.get("metric") or spec.get("y") # KPI-shaped view val = (res["rows"][0].get(metric) if res["rows"] else None) kpi = {"metric": metric, "value": val or 0} cq = spec.get("compare_query") if cq: prev_rows = _run_query(cq)["rows"] prev = prev_rows[0].get(metric) if prev_rows else None if prev: kpi["delta_pct"] = ((val or 0) - prev) / abs(prev) return {"kpi": kpi} rows = res["rows"] if spec.get("transforms"): # a saved view replays its transforms import harness.transforms as TR rows, _ = TR.apply({**res, "query": q}, spec["transforms"], run_query=_run_query) if spec.get("kind") == "table": return {"table": {**{k: spec.get(k) for k in ("kind", "title", "columns")}, "rows": rows}} return {"chart": {**{k: spec.get(k) for k in ("kind", "x", "y", "series", "title", "y2", "size", "value", "facet")}, "rows": rows}} def run_dashboard(name): """All of a dashboard's views, freshly re-queried. Returns [(view_name, artifact|None, err)].""" dash = dashboards().get(name) if dash is None: raise SEM.ModelError(f"unknown dashboard {name!r}") out = [] for vn in dash.get("views", []): try: out.append((vn, run_view(vn), None)) except Exception as e: out.append((vn, None, str(e)[:200])) return out def store_freshness(): """Newest sync checkpoint across entities — the staleness footer ('data as of …').""" try: import harness.datastore as DS ts = [s.get("updated") for s in DS.status().values() if s.get("updated")] return max(ts) if ts else None except Exception: return None