"""AI REVIEW — let a model decide which stage a card moves to (wave 23, owner ruling R4/R14). ⭐ WHAT THIS IS. A review stage holds a record until somebody decides where it goes next. R4 made that somebody optionally a MODEL: the engine hands over the record's own values, the review's prompt, and the list of stages the review is allowed to send a card to, and gets back ONE of those stage labels plus a one-line reason. Every decision is written to the same `reviews` audit log a human click writes to, tagged `by: "ai"` with the provider and model that made it. ⛔ FAIL-CLOSED IN EVERY DIRECTION, and this is the whole safety story. No key configured, a network failure, a slow answer, a malformed answer, or an answer naming a stage the review does not offer — all return `("", {...})`, and the caller leaves the card exactly where a human would have found it. The feature can be absent, broken or wrong and the worst outcome is a person doing the work. Nothing here can move a card somewhere the review does not already permit. ⭐ CHEAP FIRST (owner R14, verbatim: *"Claude is a bit too expensive"*). The ladder is `groq → cerebras → openrouter → anthropic`; the first CONFIGURED provider wins, and Anthropic is last rather than absent — it is the quality backstop, not the default. `AIOS_AI_REVIEW_PROVIDER` pins one; `AIOS_AI_MODEL` overrides the model. ⚠ WHY RAW HTTP RATHER THAN THE `anthropic` SDK, stated because it is a deliberate deviation from the /claude-api skill's default and not an oversight. Three of the four legs are OpenAI-chat-shaped endpoints with no shared SDK, so a ladder built on the SDK would be one SDK leg beside three hand-rolled ones — two implementations of the same call, the seam this repo keeps closing. And `aios-web/api/requirements.txt` is PINNED to what the verify battery proves ([[pin-deps-space- rebuilds]]): adding a dependency there rebuilds the container, which is a real deploy risk to take for one leg of an optional feature. `requests` is already a dependency and `harness/analyst.py` established per-provider raw HTTP as the house pattern. Booked as a DEBT line so the integrator can overturn it deliberately rather than by drift. The Messages shape below is the current one: `x-api-key` + `anthropic-version: 2023-06-01`, and `stop_reason: "refusal"` is checked BEFORE reading `content` — a refusal answers HTTP 200 with an empty content list, so code that indexes `content[0]` unconditionally breaks on it. """ from __future__ import annotations import json import os import re import requests #: The ladder. Order IS the policy (R14) — cheapest capable first, Anthropic last. PROVIDERS = [ {"name": "groq", "env": "GROQ_API_KEY", "shape": "openai", "url": "https://api.groq.com/openai/v1/chat/completions", "model": "llama-3.3-70b-versatile"}, {"name": "cerebras", "env": "CEREBRAS_API_KEY", "shape": "openai", "url": "https://api.cerebras.ai/v1/chat/completions", "model": "gpt-oss-120b"}, {"name": "openrouter", "env": "OPENROUTER_API_KEY", "shape": "openai", "url": "https://openrouter.ai/api/v1/chat/completions", "model": "openai/gpt-4o-mini"}, # ⚠ haiku-class DELIBERATELY, not the default Opus tier: R14 put Anthropic on this ladder as # the backstop for a one-line classification, and this is the cheapest current Claude that # does it well. A bigger model here would be spending the owner's money to pick between two # labels it already has in front of it. {"name": "anthropic", "env": "ANTHROPIC_API_KEY", "shape": "anthropic", "url": "https://api.anthropic.com/v1/messages", "model": "claude-haiku-4-5"}, ] ANTHROPIC_VERSION = "2023-06-01" TIMEOUT_SECONDS = float(os.environ.get("AIOS_AI_REVIEW_TIMEOUT") or 20) MAX_FIELD_CHARS = 200 # per value handed to the model MAX_FIELDS = 30 # columns handed to the model MAX_REASON = 200 def ladder(): """The providers that are actually usable here, in order. Empty = the feature is off.""" pin = (os.environ.get("AIOS_AI_REVIEW_PROVIDER") or "").strip().lower() live = [p for p in PROVIDERS if (os.environ.get(p["env"]) or "").strip()] if pin: live = [p for p in live if p["name"] == pin] return live def configured(): return bool(ladder()) def _record_text(row, fields): """The record, as the model sees it. Values are truncated and the column set is bounded — an automation table can carry a 32 KB JSON blob per row (C7) and a review decision does not need it. Machine bookkeeping columns are dropped: a stage cell naming the stage the card is sitting at would be the model reading its own question back.""" keys = [k for k in (fields or list((row or {}).keys())) if not str(k).startswith("stage_")][:MAX_FIELDS] lines = [] for k in keys: v = (row or {}).get(k) if v is None or str(v).strip() == "": continue lines.append(f"{k}: {str(v)[:MAX_FIELD_CHARS]}") return "\n".join(lines) or "(this record has no filled-in values)" def _instruction(prompt, options, label): return ( f"You are deciding what happens to one record waiting at a review step called " f"{label!r} in a workflow.\n\n" f"The person who built this workflow told you: {prompt}\n\n" f"Choose EXACTLY ONE of these next steps, by its exact name:\n" + "\n".join(f"- {o}" for o in options) + "\n\nAnswer with one line of JSON and nothing else:\n" '{"choice": "", "reason": ""}\n' "If the record does not give you enough to decide, answer " '{"choice": "", "reason": "why not"} and a person will decide instead.' ) def _parse(text, options): """The model's line → `(choice, reason)`. A choice that is not one of the offered stages is DISCARDED, not fuzzy-matched: the offered list is a permission boundary, and a near-miss resolved by string distance is how a card ends up somewhere nobody authorised.""" raw = str(text or "").strip() obj = None m = re.search(r"\{.*\}", raw, re.S) if m: try: obj = json.loads(m.group(0)) except (ValueError, TypeError): obj = None if not isinstance(obj, dict): return "", "" choice = str(obj.get("choice") or "").strip() reason = str(obj.get("reason") or "").strip()[:MAX_REASON] for opt in options: if choice.lower() == str(opt).lower(): return str(opt), reason # the OFFERED spelling wins, never the model's return "", reason def _call_openai(p, model, system, user, timeout): r = requests.post(p["url"], timeout=timeout, headers={"Authorization": f"Bearer {os.environ[p['env']].strip()}", "Content-Type": "application/json"}, json={"model": model, "max_tokens": 300, "temperature": 0, "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}]}) if r.status_code >= 400: return "", f"{p['name']} answered {r.status_code}" body = r.json() choices = body.get("choices") or [] if not choices: return "", f"{p['name']} returned no choices" return str(((choices[0] or {}).get("message") or {}).get("content") or ""), "" def _call_anthropic(p, model, system, user, timeout): r = requests.post(p["url"], timeout=timeout, headers={"x-api-key": os.environ[p["env"]].strip(), "anthropic-version": ANTHROPIC_VERSION, "content-type": "application/json"}, json={"model": model, "max_tokens": 300, "system": system, "messages": [{"role": "user", "content": user}]}) if r.status_code >= 400: return "", f"anthropic answered {r.status_code}" body = r.json() # ⛔ stop_reason FIRST. A safety refusal is a successful 200 with an EMPTY content list, so # reading content[0] before this check turns a refusal into an IndexError inside a run. if body.get("stop_reason") == "refusal": return "", "anthropic declined to answer this record" parts = [b.get("text") or "" for b in (body.get("content") or []) if isinstance(b, dict) and b.get("type") == "text"] if not parts: return "", "anthropic returned no text" return "".join(parts), "" def decide(*, prompt, options, row, fields=(), label="Review", timeout=None): """Pick this record's next stage. Returns `(choice, meta)`. `choice` is "" whenever a person should decide — which is every failure mode there is. `meta` carries `provider`, `model`, `reason` on success, and `problem` on refusal to answer. """ opts = [str(o) for o in (options or []) if str(o).strip()] if not opts: return "", {"problem": "the review offers no next stages"} if not str(prompt or "").strip(): return "", {"problem": "the review has no prompt for the model to follow"} live = ladder() if not live: return "", {"problem": "no AI provider is configured on this deployment"} system = _instruction(prompt, opts, label) user = "Here is the record:\n\n" + _record_text(row, fields) tmo = float(timeout or TIMEOUT_SECONDS) override = (os.environ.get("AIOS_AI_MODEL") or "").strip() problems = [] for p in live: model = override or p["model"] try: text, err = (_call_anthropic if p["shape"] == "anthropic" else _call_openai)( p, model, system, user, tmo) except Exception as e: # noqa: BLE001 text, err = "", f"{p['name']} failed: {type(e).__name__}" if err: problems.append(err) continue # ladder: a dead provider degrades to the next one choice, reason = _parse(text, opts) if not choice: # The provider ANSWERED and declined (or answered unusably). That is a decision about # this record, not a fault in the provider, so it does NOT fall through to a more # expensive one — the card goes to a human, which is what the model just asked for. return "", {"provider": p["name"], "model": model, "problem": reason or "the model did not choose one of the stages"} return choice, {"provider": p["name"], "model": model, "reason": reason} return "", {"problem": "; ".join(problems)[:300] or "no provider answered"}