"""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}")