platform/harness/analyst.py

"""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
   <m>_ly, <m>_delta, <m>_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}