platform/harness/tools.py

"""harness/tools.py — the compounding TOOL REGISTRY (OM-4 spine, 2026-07-11).

The formal tool surface the (small) model calls — the productization directive's "tools we keep
compounding". Every tool wraps the SEMANTIC layer (harness/semantic.py): the model navigates by
registry keys and recipe plans (model/skills/*.skill.yml — recipes and these tools version
TOGETHER), never by SQL. Adding a connector/topic extends what the SAME tools reach — that is the
compounding. Exported in OpenAI function-calling format (`openai_tools()`) — OpenRouter-compatible,
so any cheap model with tool-calling drives the platform.

Correctness posture (Part VI of the plan): whitelisted keys only; values parameterized downstream;
every query result carries a result_id + drill note; artifact tools (save/compose/schedule/alert)
are explicit and confirm-gated by recipe. Errors return a uniform envelope the model can read.
"""
import json
import time
import uuid
from pathlib import Path

import harness.semantic as SEM

VIEWS_PATH = Path(__file__).resolve().parents[1] / "data" / "store" / "views.json"

_RESULTS = {}          # result_id -> query result (session-scoped working memory for chart tools)
_RESULTS_CAP = 40

# The exhaustive chart vocabulary (2026-07-16): every Zelazny comparison form has a kind, so the
# Analyst never lacks a shape. The model picks by the CHART PICKER guide (analyst.py) + the
# charting skill recipes; the platform owns every pixel (app._render_analyst_artifact).
CHART_KINDS = (
    "line", "bar", "area", "scatter", "kpi", "map",              # the original six
    "pie", "donut",                                              # part-to-whole (≤6 slices)
    "stacked_bar", "grouped_bar", "ranked_bar", "stacked_pct",   # composition / rank forms
    "combo", "yoy_bars",                                         # level+rate; this-vs-last-year
    "waterfall", "pareto", "histogram", "heatmap", "treemap",    # bridge / concentration / distribution
    "funnel", "bullet", "bubble", "sparkline",                   # stages / target / 3-measure / mini
)

# Per-kind param contract (beyond x): what else the spec must carry to be renderable.
_KIND_NEEDS = {
    "combo": ("y", "y2"), "bullet": ("y", "y2"), "bubble": ("y", "size"),
    "heatmap": ("y", "value"), "histogram": (),                  # histogram bins x itself
    "stacked_bar": ("y", "series"), "grouped_bar": ("y", "series"),
    "stacked_pct": ("y", "series"),
}
_QUERY_KEYS = ("topic", "measures", "group_by", "grain", "date_from", "date_to",
               "team_id", "filters", "sort", "limit", "exclude_services")


def _remember(res):
    rid = uuid.uuid4().hex[:10]
    _RESULTS[rid] = res
    while len(_RESULTS) > _RESULTS_CAP:
        _RESULTS.pop(next(iter(_RESULTS)))
    return rid


def _ok(data):
    return {"ok": True, "data": data}


def _err(msg):
    return {"ok": False, "error": str(msg)[:400]}


# ------------------------------------------------------------------ schema tools

def list_topics():
    """The 'what data exists' tool."""
    out = []
    for k, t in SEM.topics().items():
        out.append({"topic": k, "label": t.get("label"), "entity": t.get("entity"),
                    "grain": t.get("grain"),
                    "dims": list((t.get("store") or {}).get("dims") or {}),
                    "metrics": [m for m, d in SEM.metrics().items() if d["topic"] == k]})
    return out


def describe_topic(topic):
    """The schema-learning tool: scope, grain, dims, metrics w/ definitions, and ai_context."""
    t = SEM.topics().get(topic)
    if not t:
        raise SEM.ModelError(f"unknown topic {topic!r} (use list_topics)")
    mets = {k: {"label": m.get("label"), "description": m.get("description"),
                "format": m.get("format"), "ai_context": m.get("ai_context")}
            for k, m in SEM.metrics().items() if m["topic"] == topic}
    return {"topic": topic, "label": t.get("label"), "scope": t.get("scope"),
            "grain": t.get("grain"), "ai_context": t.get("ai_context"),
            "dims": {k: v.get("label") for k, v in ((t.get("store") or {}).get("dims") or {}).items()},
            "metrics": mets}


# ------------------------------------------------------------------ query tools

def run_semantic_query(topic, measures, group_by=None, grain=None, date_from=None, date_to=None,
                       team_id=None, filters=None, sort=None, limit=1000, exclude_services=False):
    # limit defaults HIGH (1000): transforms/charts operate on the FULL result while the model
    # only ever sees rows[:100] — a small default silently truncated per-customer analytics
    # (the BELLA FLORIST wrong-decliner incident, 2026-07-17).
    res = SEM.store_query(topic, measures, group_by=group_by, grain=grain, date_from=date_from,
                          date_to=date_to, team_id=team_id, filters=filters, sort=sort,
                          limit=limit, exclude_services=exclude_services)
    # Echo the full query onto the result: chart specs built from it carry the query, so a SAVED
    # view is a re-runnable QUERY (the OM-3 viewer re-executes it live), never a stale snapshot.
    res["query"] = {"topic": topic, "measures": list(measures or []), "group_by": group_by,
                    "grain": grain, "date_from": date_from, "date_to": date_to, "team_id": team_id,
                    "filters": filters, "sort": sort, "limit": limit,
                    "exclude_services": exclude_services}
    rid = _remember(res)
    # The EFFECTIVE window, stated by the platform — the model must repeat this, never guess
    # (a query without dates covers all recorded history; there is no hidden default window).
    if date_from and date_to:
        window = f"{date_from} to {date_to}"
    elif date_from or date_to:
        window = f"{'from ' + date_from if date_from else 'through ' + date_to}"
    else:
        window = "ALL recorded history (no date filter was applied)"
    out = {"result_id": rid, "rows": res["rows"][:100], "row_count": res["row_count"],
           "measures": res["measures"], "group_by": res["group_by"], "grain": res["grain"],
           "window": window,
           "note": "every number here is drillable; cite result_id when charting"}
    if res["row_count"] >= (limit or 1000):        # surface every truncation (plan hard line)
        out["warning"] = (f"TRUNCATED: the result hit limit={limit} — the full set is larger. "
                          "Re-run with a higher limit (max 5000) BEFORE ranking, comparing or "
                          "aggregating, or your answer will be computed on a partial set.")
    return out


def get_field_values(topic, dim, search=None):
    return SEM.store_field_values(topic, dim, search=search)


# ------------------------------------------------------------------ transform tool (governed)

def transform_result(result_id, transforms):
    """Apply governed ANALYTICS TRANSFORMS to a query result -> a NEW result_id to chart/table.
    The chain is recorded on the derived result, so saved views replay query -> transforms live."""
    res = _RESULTS.get(result_id)
    if not res:
        raise SEM.ModelError(f"unknown result_id {result_id!r} — run run_semantic_query first")
    import harness.transforms as TR
    rows, applied = TR.apply(res, transforms, run_query=_run_query)
    new = {**res, "rows": rows, "row_count": len(rows),
           "transforms": (res.get("transforms") or []) + applied}
    rid = _remember(new)
    return {"result_id": rid, "rows": rows[:100], "row_count": len(rows),
            "columns": sorted(rows[0]) if rows else [],
            "note": "derived result — chart THIS result_id to show the transform"}


def _run_query(q):
    return SEM.store_query(**{k: q.get(k) for k in _QUERY_KEYS if q.get(k) is not None})


# ------------------------------------------------------------------ viz tools (emit OUR specs)

def make_chart(result_id, kind, x, y=None, title=None, series=None, y2=None, size=None,
               value=None, facet=None):
    """Returns a validated CHART SPEC the platform renders with its own primitives (design system
    enforced — the model never emits HTML/vega). Extra encodings per kind: combo/bullet need y2
    (line/target), bubble needs size, heatmap needs value (the colour measure); facet (a dim
    column) turns line|bar|area|scatter into small multiples."""
    res = _RESULTS.get(result_id)
    if not res:
        raise SEM.ModelError(f"unknown result_id {result_id!r} — run run_semantic_query first")
    if kind not in CHART_KINDS:
        raise SEM.ModelError(f"kind must be one of {CHART_KINDS}")
    cols = set(res["rows"][0]) if res["rows"] else set()
    if y is None and kind != "histogram":
        raise SEM.ModelError(f"kind={kind!r} needs y (only histogram bins x by itself)")
    for ref, nm in ((x, "x"), (y, "y"), (series, "series"), (y2, "y2"), (size, "size"),
                    (value, "value"), (facet, "facet")):
        if ref and ref not in cols:
            raise SEM.ModelError(f"{nm}={ref!r} not in result columns {sorted(cols)}")
    given = {"y": y, "y2": y2, "size": size, "value": value, "series": series}
    missing = [p for p in _KIND_NEEDS.get(kind, ()) if not given.get(p)]
    if missing:
        raise SEM.ModelError(f"kind={kind!r} also needs {missing} "
                             f"(pick from result columns {sorted(cols)})")
    if facet and kind not in ("line", "bar", "area", "scatter"):
        raise SEM.ModelError("facet (small multiples) works with line|bar|area|scatter only")
    if kind == "yoy_bars" and f"{y}_ly" not in cols:
        raise SEM.ModelError(f"yoy_bars needs a {y}_ly column — run transform_result "
                             "[{'op':'yoy'}] on the result first")
    spec = {"kind": kind, "x": x, "y": y, "series": series,
            "title": title or f"{y or x} by {x}", "result_id": result_id,
            "query": res.get("query"), "rows": res["rows"]}
    for k, v in (("y2", y2), ("size", size), ("value", value), ("facet", facet),
                 ("transforms", res.get("transforms"))):
        if v:
            spec[k] = v
    out = {"chart": spec}
    if kind in ("pie", "donut") and len(res["rows"]) > 6:
        out["note"] = (f"{len(res['rows'])} slices — the platform will show the top 5 plus an "
                       "'Other' bucket; for a cleaner story run transform_result top_n first")
    return out


def make_table(result_id, columns=None, title=None):
    """First-class TABLE artifact: the exact rows, house-formatted (sortable, totals row, the
    drill IS the table). columns (optional) picks and orders a subset."""
    res = _RESULTS.get(result_id)
    if not res:
        raise SEM.ModelError(f"unknown result_id {result_id!r} — run run_semantic_query first")
    rows = res["rows"]
    if columns:
        cols = set(rows[0]) if rows else set()
        bad = [c for c in columns if c not in cols]
        if bad:
            raise SEM.ModelError(f"columns {bad} not in result columns {sorted(cols)}")
        rows = [{c: r.get(c) for c in columns} for r in rows]
    return {"table": {"kind": "table", "title": title, "columns": columns,
                      "result_id": result_id, "query": res.get("query"),
                      "transforms": res.get("transforms"), "rows": rows}}


def make_kpi_card(result_id, metric, compare_result_id=None):
    res = _RESULTS.get(result_id)
    if not res or not res["rows"]:
        raise SEM.ModelError("result_id missing/empty — run a scalar run_semantic_query first")
    val = res["rows"][0].get(metric)
    if val is None:
        raise SEM.ModelError(f"{metric!r} not in result")
    card = {"kpi": {"metric": metric, "value": val, "result_id": result_id,
                    "query": res.get("query")}}
    if compare_result_id and _RESULTS.get(compare_result_id, {}).get("rows"):
        prev = _RESULTS[compare_result_id]["rows"][0].get(metric)
        if prev:
            card["kpi"]["delta_pct"] = (val - prev) / abs(prev)
            card["kpi"]["compare_result_id"] = compare_result_id
            card["kpi"]["compare_query"] = _RESULTS[compare_result_id].get("query")
    return card


# ------------------------------------------------------------------ artifact tools (v0: local)

def _load_views():
    if VIEWS_PATH.exists():
        return json.loads(VIEWS_PATH.read_text(encoding="utf-8"))
    return {"views": {}, "dashboards": {}}


def _save_views(d):
    VIEWS_PATH.parent.mkdir(parents=True, exist_ok=True)
    VIEWS_PATH.write_text(json.dumps(d, indent=1), encoding="utf-8")


def save_view(name, chart):
    """Persist a chart/KPI spec (from make_chart / make_kpi_card) as a named view. Specs persist
    WITH their semantic query and WITHOUT rows — the OM-3 viewer re-executes the query live, so a
    saved view is always current, never a snapshot."""
    d = _load_views()
    spec = chart.get("chart") or chart.get("kpi") or chart.get("table") or chart
    spec = {k: v for k, v in spec.items() if k != "rows"}
    if "kpi" in chart and not spec.get("kind"):
        spec["kind"] = "kpi"
    if "table" in chart and not spec.get("kind"):
        spec["kind"] = "table"
    if not spec.get("query"):
        raise SEM.ModelError("spec carries no query — pass the exact object returned by "
                             "make_chart / make_kpi_card (from a fresh run_semantic_query)")
    d["views"][name] = {"chart": spec, "saved_at": time.strftime("%Y-%m-%d %H:%M")}
    _save_views(d)
    return {"saved": name, "views": list(d["views"])}


def compose_dashboard(name, views):
    """'Spawn a dashboard': compose saved views into a named dashboard spec (rendered at OM-3)."""
    d = _load_views()
    missing = [v for v in views if v not in d["views"]]
    if missing:
        raise SEM.ModelError(f"unknown views {missing} — save_view them first")
    d["dashboards"][name] = {"views": views, "created_at": time.strftime("%Y-%m-%d %H:%M")}
    _save_views(d)
    return {"dashboard": name, "views": views}


# ------------------------------------------------------------------ the GAP LOOP (rung 4)

GAP_KINDS = ("dimension", "metric", "transform", "chart_kind", "data_source", "other")
GAPS_KEY = "analyst_gaps"
GAPS_CAP = 500


def report_gap(kind, missing, question, workaround=None):
    """Log a CAPABILITY GAP: the model determined (after checking the schema) that no registered
    dim/metric/transform/kind can answer. The entry lands in telemetry AND the durable store —
    the admin Gaps view aggregates them into the platform build backlog. This is how every
    honest 'I can't' becomes the next dim, transform, or recipe."""
    if kind not in GAP_KINDS:
        raise SEM.ModelError(f"kind must be one of {GAP_KINDS}")
    entry = {"ts": time.strftime("%Y-%m-%d %H:%M:%S"), "kind": kind,
             "missing": str(missing)[:120], "question": str(question)[:300],
             "workaround": (str(workaround)[:200] if workaround else None)}
    import harness.telemetry as TEL
    TEL.log("analyst_gap", **{("gap_kind" if k == "kind" else k): v for k, v in entry.items()})
    try:                                    # durable copy — async, the chatlog write posture
        import threading

        import core.store as store
        if store.available():
            def _fn(data):
                data = list(data or [])
                data.append(entry)
                return data[-GAPS_CAP:]
            threading.Thread(target=lambda: store.update(GAPS_KEY, _fn),
                             daemon=True, name="analyst-gap").start()
    except Exception:
        pass
    return {"logged": True,
            "note": "gap recorded for the platform backlog — now tell the user in ONE sentence "
                    "what is missing and offer the nearest ask that IS answerable"}


# ------------------------------------------------------------------ workspace tools (OM-3/P11)

def list_workspace():
    """Everything in the tenant workspace (uniform modular objects) + available templates."""
    import harness.workspace as W
    return {"objects": W.items(), "templates": W.templates()}


def instantiate_template(filename, new_name=None):
    """Stamp a tenant-agnostic template into this workspace as NEW objects (never overwrites)."""
    import harness.workspace as W
    return W.instantiate_template(filename, new_name)


def update_view(name, changes):
    """Patch an existing view (spec and/or query) — validated by re-execution before saving."""
    import harness.views as V
    return V.update_view(name, changes)


def update_workbook(name, views=None, new_name=None):
    """Recompose and/or rename a workbook (dashboard); renames follow into schedules."""
    import harness.views as V
    return V.update_dashboard(name, views_list=views, new_name=new_name)


def delete_object(kind, name):
    """Delete any workspace object. DESTRUCTIVE — the recipe requires explicit confirmation."""
    import harness.workspace as W
    W.delete(kind, name)
    return {"deleted": f"{kind} · {name}"}


# ------------------------------------------------------------------ registry + dispatch

def _p(props, required):
    return {"type": "object", "properties": props, "required": required}


TOOLS = {
    "list_topics": {"fn": lambda **kw: list_topics(),
        "description": "List the datasets (topics) available: their metrics, dims, and grain. Start here.",
        "parameters": _p({}, [])},
    "describe_topic": {"fn": lambda **kw: describe_topic(kw["topic"]),
        "description": "Full schema of one topic: scope rules, metric definitions, dims, and the business context you must respect.",
        "parameters": _p({"topic": {"type": "string"}}, ["topic"])},
    "get_field_values": {"fn": lambda **kw: get_field_values(kw["topic"], kw["dim"], kw.get("search")),
        "description": "Resolve real filter values (ids+names) for a dim. ALWAYS use before filtering by a typed name.",
        "parameters": _p({"topic": {"type": "string"}, "dim": {"type": "string"},
                          "search": {"type": "string"}}, ["topic", "dim"])},
    "run_semantic_query": {"fn": lambda **kw: run_semantic_query(**kw),
        "description": "Run a governed query: registered measures over a topic, optional group_by dims / time grain / filters. The ONLY way to read data.",
        "parameters": _p({"topic": {"type": "string"},
                          "measures": {"type": "array", "items": {"type": "string"}},
                          "group_by": {"type": "array", "items": {"type": "string"}},
                          "grain": {"type": "string", "enum": ["month", "week", "day"]},
                          "date_from": {"type": "string"}, "date_to": {"type": "string"},
                          "team_id": {"type": "integer"},
                          "filters": {"type": "object"},
                          "sort": {"type": "string"}, "limit": {"type": "integer"},
                          "exclude_services": {"type": "boolean"}},
                         ["topic", "measures"])},
    "transform_result": {"fn": lambda **kw: transform_result(kw["result_id"], kw["transforms"]),
        "description": "Apply governed analytics transforms to a result -> a NEW result_id (a "
                       "CHAINABLE list of {op, ...}). The Tableau-class analytics library — pick "
                       "op names (full catalog + recipes in the analytics skill). Families: "
                       "ordering/rank (sort, head, bottom_n, rank, rank_pct, ntile, top_n, "
                       "add_total) · part-to-whole (share_of_total, cum_share) · running/moving "
                       "(running_total/avg/max/min, running_count, moving_average/sum/median, "
                       "rolling_std) · period-over-period (diff, pct_change, lag, lead, "
                       "diff_from_first, index_to_100, percent_of_max, compare) · distribution/"
                       "stats (bin, describe, zscore, outliers, winsorize, clip, normalize, "
                       "correlate, weighted_average, safe_ratio, product) · business (abc_classify, "
                       "concentration, contribution_to_change, rfm, funnel_rates) · modeling "
                       "(trend_line, regression, cagr, growth_rate) · reference lines as columns "
                       "(reference_line, reference_band, target_line, xmr_limits) · reshape (pivot, "
                       "unpivot, filter_rows, dedupe, resample) · re-query windows (yoy, ytd, "
                       "rolling, forecast). ADDITIVITY LAW: accumulating ops "
                       "(running_total/share_of_total/cum_share/moving_sum/abc_classify/"
                       "concentration) work on ADDITIVE measures (revenue/units/margin/orders); "
                       "for a cumulative/trailing DISTINCT count (customers) or ratio use ytd/"
                       "rolling (they re-query) — never running_total. NEVER compute any of these "
                       "yourself; transform, then chart/table the new result_id.",
        "parameters": _p({"result_id": {"type": "string"},
                          "transforms": {"type": "array", "items": {"type": "object"}}},
                         ["result_id", "transforms"])},
    "make_chart": {"fn": lambda **kw: make_chart(**kw),
        "description": "Turn a query result into a platform chart. kinds: line|bar|area|scatter|"
                       "map|pie|donut|stacked_bar|grouped_bar|ranked_bar|stacked_pct|combo|"
                       "yoy_bars|waterfall|pareto|histogram|heatmap|treemap|funnel|bullet|bubble|"
                       "sparkline. x/y/series must be result columns. Extra encodings: combo "
                       "(bars y + line y2, dual axis), bullet (value y vs target y2), bubble "
                       "(scatter + size), heatmap (dims x,y + colour value), histogram (bins x, "
                       "no y), facet (a dim column -> small multiples of line|bar|area|scatter). "
                       "yoy_bars needs the yoy transform first. kind='map' plots customers "
                       "geographically: x = the customer dim, dot size = y.",
        "parameters": _p({"result_id": {"type": "string"}, "kind": {"type": "string", "enum": list(CHART_KINDS)},
                          "x": {"type": "string"}, "y": {"type": "string"},
                          "title": {"type": "string"}, "series": {"type": "string"},
                          "y2": {"type": "string"}, "size": {"type": "string"},
                          "value": {"type": "string"}, "facet": {"type": "string"}},
                         ["result_id", "kind", "x"])},
    "make_table": {"fn": lambda **kw: make_table(kw["result_id"], kw.get("columns"),
                                                 kw.get("title")),
        "description": "Turn a query result into a first-class TABLE artifact (house-formatted, "
                       "totals row, drillable). Use when the user wants exact figures, many "
                       "columns, or a list — not a shape. columns (optional) picks and orders.",
        "parameters": _p({"result_id": {"type": "string"},
                          "columns": {"type": "array", "items": {"type": "string"}},
                          "title": {"type": "string"}}, ["result_id"])},
    "make_kpi_card": {"fn": lambda **kw: make_kpi_card(**kw),
        "description": "Turn a scalar query result into a KPI card; optional compare_result_id adds a YoY delta.",
        "parameters": _p({"result_id": {"type": "string"}, "metric": {"type": "string"},
                          "compare_result_id": {"type": "string"}}, ["result_id", "metric"])},
    "report_gap": {"fn": lambda **kw: report_gap(kw["kind"], kw["missing"], kw["question"],
                                                 kw.get("workaround")),
        "description": "LAST RESORT — log a capability gap. Call ONLY after list_topics/"
                       "describe_topic confirm that NO registered dimension, metric, transform "
                       "or chart kind can answer the user's question (e.g. stock on hand, "
                       "which has no topic). NOT a gap: YoY/decline/growth compares "
                       "(transform_result yoy), rankings/top-N, shares, running totals, "
                       "distributions — those are ANSWERABLE via transform_result. Then tell "
                       "the user plainly what is missing and offer the nearest answerable ask. "
                       "NEVER call this for something the tools support, and NEVER guess "
                       "instead of calling it.",
        "parameters": _p({"kind": {"type": "string", "enum": list(GAP_KINDS)},
                          "missing": {"type": "string",
                                      "description": "what does not exist, short (e.g. 'inventory/stock-on-hand topic')"},
                          "question": {"type": "string",
                                       "description": "the user's question, verbatim"},
                          "workaround": {"type": "string",
                                         "description": "the nearest answerable alternative you offered"}},
                         ["kind", "missing", "question"])},
    "save_view": {"fn": lambda **kw: save_view(kw["name"], kw["chart"]),
        "description": "Save a chart as a named view (confirm with the user first).",
        "parameters": _p({"name": {"type": "string"}, "chart": {"type": "object"}}, ["name", "chart"])},
    "compose_dashboard": {"fn": lambda **kw: compose_dashboard(kw["name"], kw["views"]),
        "description": "Compose saved views into a named dashboard (confirm with the user first).",
        "parameters": _p({"name": {"type": "string"},
                          "views": {"type": "array", "items": {"type": "string"}}}, ["name", "views"])},
    "list_workspace": {"fn": lambda **kw: list_workspace(),
        "description": "List the tenant workspace: every saved view/dashboard/alert/report "
                       "(modular objects) plus available templates. Use when the user asks what "
                       "exists, wants to reuse/manage artifacts, or before composing.",
        "parameters": _p({}, [])},
    "instantiate_template": {"fn": lambda **kw: instantiate_template(kw["filename"],
                                                                     kw.get("new_name")),
        "description": "Stamp a tenant-agnostic template (from list_workspace) into the "
                       "workspace as NEW objects — never overwrites. Confirm with the user first.",
        "parameters": _p({"filename": {"type": "string"}, "new_name": {"type": "string"}},
                         ["filename"])},
    "update_view": {"fn": lambda **kw: update_view(kw["name"], kw.get("changes")),
        "description": "UPDATE an existing saved view: patch spec keys (title/kind/x/y/series) "
                       "and/or 'query' subkeys (measures, group_by, grain, date_from, date_to, "
                       "team_id, filters, sort, limit; null REMOVES a key). The patched query is "
                       "re-executed before saving — invalid updates are rejected. Confirm first.",
        "parameters": _p({"name": {"type": "string"}, "changes": {"type": "object"}},
                         ["name", "changes"])},
    "update_workbook": {"fn": lambda **kw: update_workbook(kw["name"], kw.get("views"),
                                                           kw.get("new_name")),
        "description": "UPDATE a workbook (dashboard): recompose its views (list = new display "
                       "order; add/remove by including/omitting) and/or rename it. Confirm first.",
        "parameters": _p({"name": {"type": "string"},
                          "views": {"type": "array", "items": {"type": "string"}},
                          "new_name": {"type": "string"}}, ["name"])},
    "delete_object": {"fn": lambda **kw: delete_object(kw["kind"], kw["name"]),
        "description": "DELETE a workspace object (view|dashboard|alert|report). DESTRUCTIVE — "
                       "requires the user's explicit confirmation in this conversation first.",
        "parameters": _p({"kind": {"type": "string",
                                   "enum": ["view", "dashboard", "alert", "report"]},
                          "name": {"type": "string"}}, ["kind", "name"])},
}


def openai_tools():
    """The registry in OpenAI function-calling format (OpenRouter-compatible)."""
    return [{"type": "function",
             "function": {"name": k, "description": v["description"], "parameters": v["parameters"]}}
            for k, v in TOOLS.items()]


def dispatch(name, arguments):
    """Uniform tool execution for the Analyst loop: JSON-safe result or a readable error the
    model can act on. Never raises."""
    t = TOOLS.get(name)
    if not t:
        return _err(f"unknown tool {name!r} (tools: {list(TOOLS)})")
    try:
        args = json.loads(arguments) if isinstance(arguments, str) else dict(arguments or {})
        missing = [r for r in t["parameters"].get("required", []) if r not in args]
        if missing:
            return _err(f"missing required arguments: {missing}")
        return _ok(t["fn"](**args))
    except SEM.ModelError as e:
        return _err(e)
    except Exception as e:
        return _err(f"{type(e).__name__}: {e}")