"""harness/analyst.py — the AIOS Analyst loop (OM-4, 2026-07-11). A CHEAP small model (via OpenRouter, OpenAI-compatible) drives the compounding tool registry (harness/tools.py) to answer questions, build charts, and spawn dashboards over the semantic layer. The model NEVER writes SQL — it picks registry keys, guided by the in-product SKILL recipes (model/skills/*.skill.yml) injected into its system prompt. Plan: omni-adoption addendum. Architecture (mirrors the Omni coordinator pattern): ask(question) -> bounded tool loop: chat -> tool_calls -> tools.dispatch() -> results -> chat … until a plain answer or MAX_ITERS. chat_fn is INJECTABLE: live = OpenRouter POST; tests = a scripted double — the loop mechanics are provable offline (no key, no tokens). Correctness posture: bounded iterations (never endless); tool errors go BACK to the model as readable envelopes (it self-corrects); the final answer carries the tool trace (auditability + the drill guarantee); exhausted-budget is reported AS exhausted, never as success (loop rules). Cost posture: DEFAULT_MODEL is a cheap tool-calling model; override per-tenant/env; prompt = the skills + topic summary (small, cacheable), NOT raw data. """ import json import os from pathlib import Path import harness.tools as T # Provider LADDER (owner 2026-07-12: "use Groq and Cerebras"): all OpenAI-compatible chat # endpoints; tried in order, failing over on payment/rate/server errors — an exhausted or # rate-limited provider degrades to the next instead of taking the Analyst down (the # OpenRouter-402 incident). Pin one with ANALYST_PROVIDER=groq|cerebras|openrouter; override # its model with ANALYST_MODEL. Every model here must re-pass the eval gate (§4b). PROVIDERS = [ {"name": "groq", "env": "GROQ_API_KEY", "url": "https://api.groq.com/openai/v1/chat/completions", "model": "llama-3.3-70b-versatile"}, {"name": "cerebras", "env": "CEREBRAS_API_KEY", "url": "https://api.cerebras.ai/v1/chat/completions", "model": "gpt-oss-120b"}, # this account's tool-capable model (from /v1/models) {"name": "openrouter", "env": "OPENROUTER_API_KEY", "url": "https://openrouter.ai/api/v1/chat/completions", "model": "openai/gpt-4o-mini"}, ] DEFAULT_MODEL = None # None = the selected provider's default model MAX_ITERS = 8 def available_providers(): pin = os.environ.get("ANALYST_PROVIDER") ps = [p for p in PROVIDERS if os.environ.get(p["env"])] if pin: ps = [p for p in ps if p["name"] == pin] or ps return ps def llm_configured(): return bool(available_providers()) SKILLS_DIR = Path(__file__).resolve().parents[1] / "model" / "skills" def _skills_text(): parts = [] for f in sorted(SKILLS_DIR.glob("*.skill.yml")): parts.append(f.read_text(encoding="utf-8")) return "\n\n".join(parts) or "(no skill files installed)" def system_prompt(): """The small model's standing instructions: the platform rules + the skill recipes. Stable content (cache-friendly); per-question content goes in the user turn.""" topics = T.list_topics() topic_summary = "\n".join( f"- {t['topic']}: {t['label']} | metrics: {', '.join(t['metrics'])} | dims: {', '.join(t['dims'])}" for t in topics) return f"""You are the AIOS Analyst for a wholesale business. You answer data questions by calling TOOLS over a governed semantic layer. THE RULES (non-negotiable): 1. You NEVER write SQL or invent field names — only registry keys from the tools' schemas. 2. Start unknown asks with list_topics/describe_topic. Resolve typed names (customers, products) with get_field_values BEFORE filtering. Business units: Fisch=5, Royal=6 — never mix them in one series unless explicitly comparing. 3. Every number you present must come from a tool result and keep its result_id (drillable). If a tool errors, read the error and correct your call — do not make numbers up, ever. 4. Artifact actions (save_view, compose_dashboard, schedule_report, create_alert) need the user's explicit confirmation first. 5. Answer plainly and lead with the finding. State the window and scope you used — READ the window from the tool result's 'window' field, never guess it: a query without dates covers ALL recorded history (there is no hidden default window). Time-bound asks ("this year", "past 4 weeks") must pass explicit date_from/date_to. 6. A single-number question gets THE SINGLE NUMBER first (the total for the asked window/scope), THEN any breakdown. Never answer a "what was X" question with only components: no monthly lists without the total, no per-BU splits when the consolidated figure was asked. If the ask is consolidated, query WITHOUT group_by (or sum before answering). 7. NEVER state an aggregate you did not obtain from a tool result. If you want to mention a total across the rows you fetched, either run the scalar query for it or ADD UP those exact rows — a round "approximately" figure is a fabrication and forbidden. When in doubt, present the rows without a total. Joining or diffing two result sets BY HAND (e.g. this year vs last year per customer) is equally forbidden — that is transform_result's job ({{"op":"yoy"}} adds _ly, _delta, _yoy_pct per row); if no transform can compute it, say you cannot. 8. THE GAP PROTOCOL. If — and only if — after checking the schema (list_topics/describe_topic) no registered dimension, metric, transform or chart kind can answer, do NOT guess and do NOT force a wrong query: call report_gap(kind, missing, question, workaround) ONCE, then answer in one honest sentence what is missing and offer the nearest ask that IS answerable. Before concluding a gap, check the TRANSFORMS: year-over-year / decline / growth comparisons (yoy), rankings (top_n/sort/head), shares, running totals and distributions are ALL answerable via transform_result — a derived-analytics question is never a gap. Both directions are failures: refusing without report_gap, and calling report_gap for something the tools support. 9. NEVER emit markdown images, data: URIs, or links to charts — you cannot draw. Charts exist ONLY through make_chart / make_table; the platform renders them below your answer. Refer to a chart as "the chart below". Derived analytics (top-N, YoY, shares, running totals, moving averages, ranks, bins) exist ONLY through transform_result — never arithmetic in your head. CHART PICKER (the form follows the comparison — pick by what the question compares): - trend over time -> line (few periods -> bar); seasonal/two-dim intensity -> heatmap - this year vs last year by period -> transform_result [{{"op":"yoy"}}] then yoy_bars - ranking / "top N" -> transform_result top_n then ranked_bar (sorted, labeled) - part-to-whole -> pie|donut ONLY when <=6 slices (top_n first); over time -> stacked_bar; shares of a shifting total -> stacked_pct; hierarchy of shares -> treemap - what drove the change / contribution -> waterfall (signed steps; platform adds the Total) - concentration ("do the top X carry the book?") -> pareto - correlation of two measures -> scatter; + a third measure -> bubble (size) - distribution of a measure -> histogram (or transform bin -> bar) - stage conversion -> funnel; actual vs target -> bullet (y=actual, y2=target) - level + rate together (e.g. GM$ + GM%) -> combo (y=bars left, y2=line right) - same small chart per segment -> facet on line|bar|area|scatter - exact figures, many columns, lists -> make_table (totals row included) - customer geography -> map. When two forms tie, pick the simpler one. DATASETS: {topic_summary} YOUR SKILL RECIPES (follow these plans; adjust the [bracketed] parameters): {_skills_text()}""" _DEAD_STATUS = {401, 402, 403} # payment/auth: skip the provider immediately, no retry _RETRY_STATUS = {429, 500, 502, 503} # rate/server: retry with backoff before failing over _MAX_ANSWER_TOKENS = 1500 # answers are short; also keeps low-credit providers usable _COOLDOWN = {} # provider name -> unix ts until which we skip it (in-proc) def _live_chat(messages, tools, model, tool_choice="auto"): """POST down the provider ladder with LATENCY-FIRST failover (a chat user is waiting): a rate-limited provider is SKIPPED IMMEDIATELY in favor of the next one — we only sleep when the ENTIRE ladder is limited (then short ladder-level backoffs). A LONG Retry-After (daily quota, e.g. Groq TPD) puts the provider in an in-process COOLDOWN; 401/402/403 skip the provider outright.""" import time as _time import requests provs = available_providers() if not provs: raise RuntimeError("no LLM provider configured — set GROQ_API_KEY / CEREBRAS_API_KEY / " "OPENROUTER_API_KEY in platform/.env") last_err = None # Two patience profiles: CHAT (default) fails over fast — a user is waiting; BATCH # (ANALYST_PATIENT=1, set by the eval gate / insight runner) waits out per-minute rate # windows across more ladder rounds instead of exhausting the ladder in seconds. patient = os.environ.get("ANALYST_PATIENT") == "1" round_sleeps = (5, 10, 20, 30, 45, 60) if patient else (2, 6, 12) for rnd in range(len(round_sleeps)): # ladder rounds saw_retryable = False for p in provs: if _COOLDOWN.get(p["name"], 0) > _time.time(): last_err = RuntimeError(f"{p['name']}: cooling down (quota)") continue try: r = requests.post(p["url"], timeout=120, headers={"Authorization": f"Bearer {os.environ[p['env']]}", "HTTP-Referer": "https://aios.local", "X-Title": "AIOS Analyst"}, json={"model": (model or os.environ.get("ANALYST_MODEL") or p["model"]), "messages": messages, "tools": tools, "tool_choice": tool_choice, "temperature": 0.1, "max_tokens": _MAX_ANSWER_TOKENS}) if r.status_code in _DEAD_STATUS: last_err = RuntimeError(f"{p['name']}: HTTP {r.status_code}") _COOLDOWN[p["name"]] = _time.time() + 300 # don't re-poke dead auth/payment continue if r.status_code == 400: # ⛔ WAVE 32 (W32-T50/T54) — A 400 FROM THIS LADDER IS NOT A TRANSPORT FACT, # AND THIS FUNCTION USED TO THROW ITS EXPLANATION AWAY. Groq validates the # model's tool ARGUMENTS against the tool schema server-side and answers # `400 tool_use_failed` with the generation it rejected in `failed_generation`. # Measured 2026-08-13: a correct refusal that omitted a `required` property # arrived here, `raise_for_status()` turned it into "400 Client Error", and the # caller was told `all LLM providers failed` — for a request the model had # answered. The control flow is unchanged (still fail over); what changes is # that the message NAMES the cause, because a schema this ladder sent is the # one thing the operator can actually fix. detail = "" try: e400 = (r.json() or {}).get("error") or {} detail = str(e400.get("message") or "")[:300] if e400.get("failed_generation"): detail += f" | rejected generation: " \ f"{str(e400['failed_generation'])[:300]}" except Exception: detail = (r.text or "")[:300] last_err = RuntimeError(f"{p['name']}: HTTP 400 {detail}") continue if r.status_code in _RETRY_STATUS: last_err = RuntimeError(f"{p['name']}: HTTP {r.status_code}") try: wait = float(r.headers.get("retry-after") or 0) except ValueError: wait = 0 if wait > 60: # daily/long quota — cool down _COOLDOWN[p["name"]] = _time.time() + wait else: saw_retryable = True # NO sleep — next provider NOW continue r.raise_for_status() usage = r.json().get("usage", {}) return r.json()["choices"][0]["message"], usage except requests.RequestException as e: last_err = e continue # network blip — next provider if not saw_retryable: break # nothing retryable left anywhere _time.sleep(round_sleeps[rnd]) # whole ladder limited — pause a round raise RuntimeError(f"all LLM providers failed (last: {last_err})") #: Wave 15 R2 — what a restricted account is told instead of an answer. ANALYST_DENIED = ("The Analyst is not available on your account. It answers from the full " "governed model, which is wider than what your account is permitted to see.") def may_ask(user): """⛔ R2 — THE AI WALL, AND IT IS DELIBERATELY BLUNT. Owner, verbatim: *"If the AI is triggered by a user, the AI ITSELF should not be able to see the data."* Today the Analyst's tools run UNSCOPED — `harness/tools.run_semantic_query` takes no subject and no filter, so every question it answers is answered over the whole tenant. A restricted account asking "what were our top customers?" would therefore receive, in prose, precisely the rows the grid's permanent filter exists to withhold. Prose is not a lesser leak than a table; it is a harder one to notice. So the wall is REFUSAL, not narrowing: an account with ANY declared restriction — or any non-admin on a migrated record — does not reach the model at all. An AI that never runs cannot leak what it never read. Fail-closed on an unreadable record for the same reason. ⭢ NAMED DEBT, owner-mandated (R2, doc §Debt, log 2026-08-02). The RIGHT answer is to thread the subject's permanent filter into every semantic query the Analyst runs — `filter_sql` already compiles that tree to SQL, so the shape exists; what does not exist is a per-subject query path through `harness/semantic.py`, which takes `team_id` and nothing else. Until that lands this refusal is the whole enforcement, and it must NOT be relaxed into "narrow the prompt" or "tell the model what it may not say" — a prompt is not a permission boundary. """ import core.perms as perms import core.perm_scope as perm_scope if not isinstance(user, dict): return False # no identity resolved -> no Analyst if perms.is_admin(user): return True if perm_scope.is_migrated(user): # Migrated and not an admin: the account is governed by a wall the Analyst cannot honour. return False # Legacy record: restricted the old way is still restricted. return (perms.allowed_modules(user) is None and perms.scope_team_id(user) is None and not perms.scope_agent(user)) def ask(question, history=None, chat_fn=None, model=DEFAULT_MODEL, max_iters=MAX_ITERS, telemetry_kind="analyst", user=None): """Run one Analyst turn. Returns {answer, artifacts, tool_trace, iterations, exhausted, usage}. chat_fn(messages, tools, model) -> (assistant_message_dict, usage_dict). `user` is the SUBJECT asking. Passing it engages the R2 wall (`may_ask`); omitting it keeps the unscoped behaviour that `harness/evals.py` and `harness/routines.py` rely on — those run as the SYSTEM, not on behalf of a person, and gating them on a user record they do not have would break the eval ladder to no security benefit. Every call that IS on behalf of somebody passes it, and `app.py`'s chat entry point does. """ if user is not None and not may_ask(user): return {"answer": ANALYST_DENIED, "artifacts": [], "tool_trace": [], "iterations": 0, "exhausted": False, "denied": True, "usage": {"prompt_tokens": 0, "completion_tokens": 0}} chat = chat_fn or _live_chat tools = T.openai_tools() messages = [{"role": "system", "content": system_prompt()}] messages += list(history or []) # The model has no clock: anchor relative windows ("last week", "this month") to today. # Lives in the USER turn so the system prompt stays byte-stable (prompt-cache friendly). import datetime as _dt messages.append({"role": "user", "content": f"(today is {_dt.date.today().isoformat()})\n{question}"}) trace, artifacts, usage_total = [], [], {"prompt_tokens": 0, "completion_tokens": 0} for i in range(max_iters): msg, usage = chat(messages, tools, model) for k in usage_total: usage_total[k] += usage.get(k, 0) or 0 tool_calls = msg.get("tool_calls") or [] # Free-tier models INTERMITTENTLY answer a data question without calling any tool # (~1 in 4 observed on gpt-oss-120b — the "no chart" defect). Deterministic fix: if # the VERY FIRST turn used no tools, redo it once with tool_choice='required'. Meta # questions just get a harmless schema lookup before answering. if i == 0 and not tool_calls and chat is _live_chat: msg, usage = _live_chat(messages, tools, model, tool_choice="required") for k in usage_total: usage_total[k] += usage.get(k, 0) or 0 tool_calls = msg.get("tool_calls") or [] messages.append({"role": "assistant", "content": msg.get("content") or "", **({"tool_calls": tool_calls} if tool_calls else {})}) if not tool_calls: answer = (msg.get("content") or "").strip() if not answer: # A model occasionally returns NO content and NO tool calls (observed live # 2026-07-16: the reply bubble was simply empty and the next turn hallucinated). # Never surface silence — report the failure honestly instead. answer = ("(the model returned an empty answer — nothing was computed; " "please rephrase or ask again)") import harness.telemetry as _tel # Wave 19 (R4): ATTRIBUTED. The token count was previously logged against nobody, so # "which customer is spending model budget, and who inside it?" had no answer even in # principle. Read defensively off `user` because this function is deliberately # callable WITHOUT one — `harness/evals.py` and `harness/routines.py` run as the # system, and telemetry omits both keys for them rather than inventing a subject. _tel.log(telemetry_kind, question=question[:200], user=(user or {}).get("username") if isinstance(user, dict) else None, tenant=(user or {}).get("tenant") if isinstance(user, dict) else None, model=model or (available_providers() or [{}])[0].get("name", "auto"), iterations=i + 1, tools=[t["tool"] for t in trace], tokens=usage_total["prompt_tokens"] + usage_total["completion_tokens"]) return {"answer": answer, "artifacts": artifacts, "tool_trace": trace, "iterations": i + 1, "exhausted": False, "usage": usage_total} for tc in tool_calls: name = tc["function"]["name"] args = tc["function"].get("arguments") or "{}" res = T.dispatch(name, args) trace.append({"tool": name, "args": args if isinstance(args, str) else json.dumps(args), "ok": res["ok"], "error": res.get("error")}) if res["ok"] and isinstance(res["data"], dict) and ( "chart" in res["data"] or "kpi" in res["data"] or "table" in res["data"] or "dashboard" in res["data"]): artifacts.append(res["data"]) messages.append({"role": "tool", "tool_call_id": tc.get("id", name), "content": json.dumps(res, default=str)[:8000]}) # budget exhausted — report honestly (never as success) return {"answer": "(stopped: tool-call budget exhausted before a final answer — " "narrow the question or raise max_iters)", "artifacts": artifacts, "tool_trace": trace, "iterations": max_iters, "exhausted": True, "usage": usage_total}