The design brain: toggleable BI/UX intelligence for dashboard generation (v1 + verified v2 iteration)

.github/workflows/ci.yml

@@ -3,6 +3,11 @@ name: ci
 on:
   push:
     branches: [main]
+  # Branches must clear the same bar as main BEFORE they merge: the offline
+  # matrix (Linux + Windows, 3.11) and the live matrix (three Superset
+  # versions) are exactly the checks a feature branch is most likely to break.
+  pull_request:
+    branches: [main]
   workflow_dispatch:
 
 jobs:

README.md

@@ -108,12 +108,33 @@ Every capability, with the CLI verb reference: [docs/FEATURES.md](https://github
   then ask for a dashboard in plain words; the AI writes the spec, and the
   tool verifies and builds it.
 - **MCP server**: `chartwright-mcp` (installed with `pip install "chartwright[mcp]"`)
-  exposes six tools covering the whole lifecycle, usable from any MCP client.
+  exposes ten tools covering the whole lifecycle, usable from any MCP client.
 - **Open contract**: `chartwright schema` prints the spec's JSON Schema, so any LLM or
   tool can generate valid specs.
 - **Guardrails**: the AI proposes; the tool verifies, using your own Superset
   login. Verification reads names (datasets, columns, metrics), not rows.
 
+## The design brain
+
+Correct-by-construction is table stakes; the design brain makes dashboards
+*read well*. A toggleable BI/UX intelligence layer — audience-aware size and
+scroll budgets, chart-choice limits, layout composition — split into a brief
+the AI reads before authoring and a deterministic critic that reviews the
+result:
+
+```bash
+chartwright brief --audience executive        # the guidance, before writing a spec
+chartwright advise spec.json --fix            # the critique, with safe auto-repairs
+chartwright redesign old-dash --profile prod  # audit + repair a live dashboard
+```
+
+Rules key on stable ids you can suppress per chart in the spec, thresholds
+tune per audience or per deployment (`design.yaml`), and the brain learns
+your house heights from the sizes you polish by hand (`chartwright
+calibrate`). Off is really off: `--design off` restores byte-identical
+behavior. The full rulebook and architecture:
+[docs/DESIGN-BRAIN.md](https://github.com/debabsah/chartwright/blob/main/docs/DESIGN-BRAIN.md).
+
 ## Drawing layouts as text
 
 The spec's `sketch` is the dashboard's shape, drawn with characters that you
@@ -133,7 +154,8 @@ space under `K` deliberately empty. Every rule, drawn and explained:
 
 ## Testing and evidence
 
-Every push runs 125 tests on Linux and Windows, plus the full pipeline (apply,
+Every push and every pull request runs the full offline suite on Linux and
+Windows, plus the full pipeline (apply,
 lifecycle soak, stale-tab adversary, fault injection) against real Superset
 4.1.4, 5.0.0, and 6.1.0 containers. Chart options are checked against
 Superset's own source for every supported version, so a Superset change is

chartwright/absorb.py

@@ -19,6 +19,7 @@
 from __future__ import annotations
 
 import copy
+import json
 from dataclasses import asdict, dataclass, field
 
 from . import ids

chartwright/cli.py

@@ -5,6 +5,10 @@
     chartwright compile spec.json -o out.zip    compile bundle with a stub resolution (golden/debug)
     chartwright check spec.json --profile P     pre-flight referential resolution
     chartwright apply spec.json --profile P     check -> compile -> import -> smoke
+    chartwright brief                           the design brief to read BEFORE authoring a spec
+    chartwright advise spec.json                design review (add --profile for data-aware rules)
+    chartwright redesign <slug> --profile P     decompile + audit + safe fixes -> redesigned spec
+    chartwright calibrate                       learn recommended heights from absorb history
 """
 
 from __future__ import annotations
@@ -37,6 +41,43 @@ def _die(payload: dict, code: int = 1) -> None:
     sys.exit(code)
 
 
+def _design_blocks(advice: dict) -> str | None:
+    """Why `--design strict` should block, or None. A gate that cannot EVALUATE
+    must fail closed: advice degrades to counts of zero when the overlay is
+    broken (see _advice_payload), and a silent pass there would turn one typo
+    in an org-wide design.yaml into a disarmed gate everywhere it is used."""
+    if advice.get("errors"):
+        return ("design advice could not be evaluated: "
+                + "; ".join(e.get("detail", "") for e in advice["errors"])
+                + " -- fix it or pass --design off")
+    if advice["counts"]["error"] or advice["counts"]["warn"]:
+        return ("design findings block under --design strict; fix them, run "
+                "`chartwright advise --fix`, or record deliberate exceptions "
+                "in the spec's design.ignore")
+    return None
+
+
+def _advice_payload(spec, resolution=None) -> dict:
+    """Advice riding along check/apply must never break the pipeline: a bad
+    overlay degrades to an error note inside the advice block, not a crash.
+    Under --design strict that error block BLOCKS (see _design_blocks); under
+    warn it is reported and the pipeline continues."""
+    from .design import advise
+
+    try:
+        return advise(spec, resolution=resolution).payload()
+    except Exception as e:  # noqa: BLE001 - advice must NEVER break check/apply
+        from .design import DESIGN_BRAIN_VERSION
+        from .design.presets import DEFAULT_AUDIENCE
+
+        return {"stage": "design", "ok": True, "design_brain": DESIGN_BRAIN_VERSION,
+                "audience": DEFAULT_AUDIENCE,
+                "counts": {"error": 0, "warn": 0, "info": 0}, "findings": [],
+                "fixed": [], "ignored": [],
+                "errors": [{"code": "overlay" if isinstance(e, ValueError) else "advice",
+                            "detail": str(e)}]}
+
+
 def _client(profile_name: str):
     from .client import SupersetClient
     from .profiles import load_profile, ProfileError
@@ -67,22 +108,58 @@ def main(argv: list[str] | None = None) -> None:
 
 
 def _main(argv: list[str] | None = None) -> None:
-    ap = argparse.ArgumentParser(prog="chartwright", description=__doc__)
+    ap = argparse.ArgumentParser(prog="chartwright", description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
     sub = ap.add_subparsers(dest="cmd", required=True)
 
-    sub.add_parser("schema")
+    sub.add_parser("schema", help="print the JSON Schema a spec must satisfy")
 
-    v = sub.add_parser("validate")
+    v = sub.add_parser("validate", help="schema-validate a spec (offline, no network)")
     v.add_argument("spec")
 
-    comp = sub.add_parser("compile")
+    comp = sub.add_parser("compile", help="compile an import bundle offline (stub dataset ids)")
     comp.add_argument("spec")
     comp.add_argument("-o", "--output", default=None)
 
+    subhelp = {"check": "pre-flight referential resolution against the live instance",
+               "apply": "check -> compile -> import -> verify -> smoke",
+               "plan": "diff the spec against the live dashboard (drift detection)"}
     for name in ("check", "apply", "plan"):
-        p = sub.add_parser(name)
+        p = sub.add_parser(name, help=subhelp[name])
         p.add_argument("spec")
         p.add_argument("--profile", required=True)
+        if name != "plan":
+            p.add_argument("--design", choices=["off", "warn", "strict"], default="warn",
+                           help="design-brain advice: warn (report, default), strict (block), off")
+
+    from .design.presets import AUDIENCE_NAMES
+
+    adv = sub.add_parser("advise", help="design review of a spec against the design-brain rulebook")
+    adv.add_argument("spec")
+    adv.add_argument("--profile", default=None, help="enable data-aware rules (types, cardinality)")
+    adv.add_argument("--audience", choices=AUDIENCE_NAMES, default=None)
+    adv.add_argument("--fix", action="store_true", help="apply safe presentation-only fixes to the spec file")
+    adv.add_argument("--strict", action="store_true", help="exit 1 on warnings, not just errors")
+    adv.add_argument("--ignore", default=None, help="comma-separated rule ids to suppress")
+    adv.add_argument("--no-probe", action="store_true", help="skip cardinality queries (metadata only)")
+
+    br = sub.add_parser("brief", help="the design brief to read BEFORE authoring a spec")
+    br.add_argument("--audience", choices=AUDIENCE_NAMES, default="analytical")
+
+    rd = sub.add_parser("redesign",
+                        help="decompile a live dashboard, audit it, apply safe fixes, write the redesigned spec")
+    rd.add_argument("dashboard", help="slug or numeric id of a live dashboard")
+    rd.add_argument("--profile", required=True)
+    rd.add_argument("--audience", choices=AUDIENCE_NAMES, default=None)
+    rd.add_argument("-o", "--output", default=None,
+                    help="write the redesigned spec here (default: <slug>.json)")
+    rd.add_argument("--no-probe", action="store_true", help="skip cardinality queries (metadata only)")
+
+    cal = sub.add_parser("calibrate", help="propose recommended heights from absorb history")
+    cal.add_argument("--write", action="store_true", help="record proposals in the design overlay")
+    cal.add_argument("--min-samples", type=int, default=5)
+    cal.add_argument("--since", default=None, metavar="90d",
+                     help="only consider absorb events newer than this (decay knob)")
 
     dec = sub.add_parser("decompile")
     dec.add_argument("dashboard", help="slug or numeric id of a live dashboard")
@@ -128,18 +205,144 @@ def _main(argv: list[str] | None = None) -> None:
 
         res = check(spec, client)
         payload = {"ok": res.ok, "stage": "resolve", "errors": [e.as_dict() for e in res.errors]}
+        gate = False
+        if args.design != "off":
+            advice = _advice_payload(spec, resolution=res if res.ok else None)
+            payload["advice"] = advice
+            blocked = _design_blocks(advice) if args.design == "strict" else None
+            gate = blocked is not None
+            if gate:
+                payload["ok"] = False
+                payload["errors"].append({"code": "design_gate", "detail": blocked})
         print(json.dumps(payload, indent=2))
-        sys.exit(0 if res.ok else 1)
+        sys.exit(0 if res.ok and not gate else 1)
 
     if args.cmd == "apply":
         spec = _load(args.spec)
+        advice = None
+        if args.design != "off":
+            # Pre-flight, offline (no probes: applies stay fast; data-aware
+            # advice is `chartwright advise --profile`). Strict blocks BEFORE
+            # anything on the instance is touched.
+            advice = _advice_payload(spec)
+            blocked = _design_blocks(advice) if args.design == "strict" else None
+            if blocked:
+                _die({"stage": "design", "ok": False, "advice": advice,
+                      "errors": [{"code": "design_gate", "detail": blocked}]})
         client = _client(args.profile)
         from .apply import apply as run_apply
 
         report = run_apply(spec, client, args.profile)
-        print(report.to_json())
+        out = json.loads(report.to_json())
+        if advice is not None:
+            out["advice"] = advice
+        print(json.dumps(out, indent=2))
         sys.exit(0 if report.ok else 1)
 
+    if args.cmd == "advise":
+        spec = _load(args.spec)
+        resolution = prober = None
+        if args.profile:
+            client = _client(args.profile)
+            from .resolver import resolve
+
+            resolution = resolve(spec, client)
+            if not args.no_probe:
+                from .design.probe import CardinalityProber
+
+                prober = CardinalityProber(client)
+        ignore = tuple(s.strip() for s in (args.ignore or "").split(",") if s.strip())
+        from .design import advise, advise_and_fix
+
+        written = None
+        try:
+            if args.fix:
+                spec_data = json.loads(Path(args.spec).read_text(encoding="utf-8"))
+                new_data, report = advise_and_fix(
+                    spec_data, audience=args.audience, ignore=ignore,
+                    resolution=resolution, prober=prober)
+                if report.fixed:
+                    # --fix rewrites the whole file (normalized JSON formatting,
+                    # same as absorb); the payload discloses the path.
+                    Path(args.spec).write_text(
+                        json.dumps(new_data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+                    written = str(args.spec)
+            else:
+                report = advise(spec, audience=args.audience, ignore=ignore,
+                                resolution=resolution, prober=prober)
+        except ValueError as e:
+            _die({"stage": "design", "errors": [{"code": "overlay", "detail": str(e)}]})
+        payload = report.payload()
+        if written:
+            payload["written"] = written
+        if resolution is not None and resolution.errors:
+            payload["resolution_errors"] = [e.as_dict() for e in resolution.errors]
+        print(json.dumps(payload, indent=2))
+        sys.exit(1 if report.gate(args.strict) else 0)
+
+    if args.cmd == "redesign":
+        client = _client(args.profile)
+        from .decompile import decompile_live
+
+        try:
+            result = decompile_live(args.dashboard, client)
+        except ValueError as e:
+            _die({"stage": "redesign", "errors": [{"code": "decompile", "detail": str(e)}]})
+        try:
+            spec = load_spec(result.spec)
+        except ValidationError as e:
+            _die({"stage": "redesign", "losses": result.losses_json(), "errors": [
+                {"code": "decompiled_spec_invalid",
+                 "detail": "the decompiled spec does not load; redesign by hand from "
+                           "`chartwright decompile` output"},
+                *json.loads(e.json()),
+            ]})
+        from .apply import _ownership_guard
+        from .design.probe import CardinalityProber
+        from .design.redesign import redesign_spec
+        from .resolver import resolve
+
+        owned = _ownership_guard(spec, client) is None
+        resolution = resolve(spec, client)
+        prober = None if args.no_probe else CardinalityProber(client)
+        try:
+            new_data, payload = redesign_spec(
+                result.spec, result.losses_json(), owned=owned,
+                audience=args.audience, resolution=resolution, prober=prober)
+        except ValueError as e:
+            _die({"stage": "design", "errors": [{"code": "overlay", "detail": str(e)}]})
+        out = Path(args.output or f"{new_data['dashboard']['slug']}.json")
+        if args.output is None and out.exists():
+            _die({"stage": "redesign", "errors": [{
+                "code": "output_exists",
+                "detail": f"{out} already exists (likely a previous redesign); "
+                          f"pass -o to choose where to write"}]})
+        out.write_text(json.dumps(new_data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+        payload["output"] = str(out)
+        if resolution.errors:
+            payload["resolution_errors"] = [e.as_dict() for e in resolution.errors]
+        print(json.dumps(payload, indent=2))
+        sys.exit(0 if payload["ok"] else 1)
+
+    if args.cmd == "brief":
+        from .design.brief import render_brief
+
+        try:
+            print(render_brief(args.audience))
+        except ValueError as e:
+            _die({"stage": "design", "errors": [{"code": "overlay", "detail": str(e)}]})
+        return
+
+    if args.cmd == "calibrate":
+        from .design.calibrate import calibrate
+
+        try:
+            print(json.dumps(calibrate(min_samples=args.min_samples, write=args.write,
+                                       since=args.since), indent=2))
+        except ValueError as e:
+            _die({"stage": "calibrate", "errors": [{"code": "overlay", "detail": str(e)}]})
+        return
+
     if args.cmd == "plan":
         spec = _load(args.spec)
         client = _client(args.profile)
@@ -166,10 +369,18 @@ def _main(argv: list[str] | None = None) -> None:
         live_position = json.loads(detail.get("position_json") or "{}")
         spec_data = json.loads(Path(args.spec).read_text(encoding="utf-8"))
         new_data, report = absorb_heights(spec, spec_data, live_position)
+        # Serialize BEFORE any file side effect: a reporting failure must never
+        # follow a silent mutation of the user's spec file.
+        payload = report.to_json()
         if report.absorbed and not args.dry_run:
             Path(args.spec).write_text(
                 json.dumps(new_data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
-        print(report.to_json())
+            # Feed the design brain's calibration loop (chartwright calibrate):
+            # absorbed heights are ground truth about what humans actually want.
+            from .design.calibrate import record_absorb
+
+            record_absorb(args.profile, spec, report.absorbed)
+        print(payload)
         sys.exit(0)
 
     if args.cmd == "decompile":

chartwright/compiler.py

@@ -200,10 +200,20 @@ def metric(m: str):
         if chart.columns:
             p["query_mode"] = "raw"
             p["all_columns"] = chart.columns
+            if chart.sort_by:
+                # Raw mode sorts on COLUMNS: order_by_cols holds JSON-encoded
+                # [column, ascending] pairs (Table controlPanel, all supported
+                # versions). False = descending, matching the ranking intent.
+                p["order_by_cols"] = [json.dumps([chart.sort_by, False])]
         else:
             p["query_mode"] = "aggregate"
             p["groupby"] = chart.groupby or []
             p["metrics"] = [metric(m) for m in (chart.metrics or [])]
+            if chart.sort_by:
+                # Aggregate mode sorts on a METRIC via timeseries_limit_metric
+                # ("Sort by" in the UI), declared by the Table plugin on 4.1.4,
+                # 5.0.0 and 6.1.0 alike (tools/contracts/params-contract.json).
+                p["timeseries_limit_metric"] = metric(chart.sort_by)
         p["row_limit"] = chart.row_limit or DEFAULT_ROW_LIMIT[t]
         p["server_page_length"] = 10
         if chart.sort_by: