diff --git a/README.md b/README.md index 8cdaf4b..733d12d 100644 --- a/README.md +++ b/README.md @@ -332,6 +332,43 @@ event store через `run_projections()` — консистентно с sched `/etc/simbiot/dashboard.env` (env-файл юнита, режим 600); без токена — 401. UFW: порт 8765/tcp открыт (токен-защищён). +## Decision-модель (Ollama /v1/systemone: облачная clef:27b → локальная nimble) + +Decision-модели из бенча `symbiot/systemone-bench`. Основная — `clef:27b` +в Ollama Cloud (`https://ollama.com/v1/systemone`, ключ +`SIMBIOT_DECISION_API_KEY` в `.env` инстанса, ~0.6–0.9 с на запрос). +Запасная — локальная nimble (qwen3.5-9B Q8_0, `ollama pull nimble`, +Ollama ≥ 0.35.1, CPU 10–30 с): отвечает, если облако упало/квота/сеть. +У каждого бэкенда свой предохранитель (5 мин после сбоя). Не генерирует +текст — возвращает распределение по вариантам (`choice`) или P(да) +(`noul`); детерминирована, кредитов роутера не тратит. Схема как в +`ga/router42.py` бенча: детерминированная механика — быстрый путь, +nimble — арбитр там, где механика слепа; недоступна — тихий фолбэк. + +| Потребитель | Было | С моделью | Режим по умолчанию | +|---|---|---|---| +| `tasks` — порядок батча тика | чистый FIFO | из 9 старейших первой — самая ценная для AUM, остальные FIFO | on | +| `topic` — тема инициативной реплики | round-robin | выбор из 9 наименее использованных с учётом доски | on | +| `gate` — серая зона гейта (только warn/info) | LLM-тик всегда | P(нужен LLM) < 0.35 → тихий такт | **shadow** (только пишет) | +| `agent` — `cli.py decide -q … -o k=v` | — | выбор 2–9 вариантов для LLM-агента тика | on | + +Конфиг (env > `.env` инстанса): `SIMBIOT_DECISION=off|shadow|on` (дефолт +off — тесты герметичны), точечно `SIMBIOT_DECISION_`, +`SIMBIOT_DECISION_API_KEY` (есть → основной = облако clef:27b, нет → +локальная nimble без запасного), `SIMBIOT_DECISION_URL`, +`SIMBIOT_DECISION_MODEL`, `SIMBIOT_DECISION_TIMEOUT` (облако 15 с / +локально 90 с), `SIMBIOT_DECISION_FALLBACK=on|off`, +`SIMBIOT_DECISION_FALLBACK_URL|MODEL|TIMEOUT` (локальная nimble, 90 с), `SIMBIOT_DECISION_GATE_MIN_P` (0.35), +`SIMBIOT_DECISION_KEEP_ALIVE` (2h). Глобальный shadow — потолок для всех. + +Ограничения из бенча: шортлист строго ≤ 9 (точность 100% на 6 → 50% на +10), `confidence` — не вероятность правильности. Кеш по хешу +запроса (ответы детерминированы — та же очередь между тиками бесплатна), +только min/hour/day (не sec). Каждый реальный вызов → событие +`DecisionMade` (с `backend`/`model`) (policy trace для GA-сшивки с +исходом). Статус/статистика: `cli.py decide`, `cli.py decide --stats`. +Тест: `./venv/bin/python tests/test_decision.py` (фейковый эндпоинт). + ## Модели 9router (router.cil.by), отдельный ключ `sk-simbiot` в `.env`, комбо `auto`: diff --git a/cli.py b/cli.py index f7a2ecc..ee9cc55 100644 --- a/cli.py +++ b/cli.py @@ -190,6 +190,14 @@ def main(): p_mcs = sub.add_parser("mcp-stats", help="статистика MCP-вызовов по журналу mcp_calls.log (t-261015091108)") p_mcs.add_argument("--tail", type=int, default=10, help="последние N вызовов (0 — только сводка)") + p_dec = sub.add_parser("decide", help="decision-модель nimble (Ollama /v1/systemone): статус, вопрос, статистика") + p_dec.add_argument("-q", "--question", help="вопрос/инструкция (без -o — noul: P(да))") + p_dec.add_argument("-o", "--option", action="append", default=[], + help="вариант ключ=описание (2–9 штук → choice)") + p_dec.add_argument("-s", "--state", default="", help="ситуация/контекст для модели") + p_dec.add_argument("--stats", type=int, nargs="?", const=200, metavar="N", + help="сводка последних N событий DecisionMade") + args = ap.parse_args() if args.cmd == "init": @@ -586,6 +594,75 @@ def main(): es.init_db() for t in commands.list_tasks(): print(f"{t['id']} [{t['size']:2}] {t['status']:7} {t['title'][:70]}") + elif args.cmd == "decide": + from harness import decision + if args.stats: + import collections + es.init_db() + total = es.count_events() + evs = [e for e in es.get_events(max(0, total - 20000), 20000) + if e["event_type"] == "DecisionMade"][-args.stats:] + by = collections.defaultdict(lambda: {"n": 0, "applied": 0, "err": 0, "ms": []}) + for e in evs: + d = json.loads(e["data"]) + b = by[d.get("purpose", "?")] + b["n"] += 1 + b["applied"] += bool(d.get("applied")) + b["err"] += bool(d.get("error")) + if d.get("ms") and not d.get("error"): + b["ms"].append(d["ms"]) + if not by: + print("(решений ещё не было)") + for p, b in sorted(by.items()): + ms = sorted(b["ms"]) + med = ms[len(ms) // 2] if ms else 0 + print(f"{p:8} решений {b['n']:4} применено {b['applied']:4} " + f"ошибок {b['err']:3} медиана {med} мс") + return + if args.question: + opts = {} + for o in args.option: + k, _, v = o.partition("=") + opts[k.strip()] = v.strip() or k.strip() + os.environ.setdefault("SIMBIOT_DECISION_AGENT", "on") + if decision.mode("agent") == "off": + print("decision-модель выключена (SIMBIOT_DECISION=off) — включи on/shadow") + sys.exit(2) + if len(opts) > decision.MAX_OPTIONS: + print(f"ОШИБКА: вариантов {len(opts)} > {decision.MAX_OPTIONS} — " + "сократи шортлист (точность падает до 50% на 10)") + sys.exit(2) + state = args.state or decision.AUM_GOAL + if opts: + r = decision.choose("agent", opts, state, args.question) + if not r: + print("нет решения (модель недоступна или ответ вне вариантов)") + sys.exit(1) + print(f"ВЫБОР: {r['choice']}") + for k, v in sorted(r["probabilities"].items(), key=lambda kv: -kv[1]): + print(f" {v:6.1%} {k}") + else: + r = decision.yes_prob("agent", args.question, state) + if not r: + print("нет решения (модель недоступна)") + sys.exit(1) + print(f"P(да) = {r['p']:.1%}") + return + st = decision.status() + print(f"режим: {st['mode']} ({', '.join(f'{k}={v}' for k, v in st['modes'].items())})") + for b in st["backends"]: + where = "облако" if b["cloud"] else "локально" + print(f"{b['name']}: {b['model']} ({where}) {b['url']}") + if not b.get("reachable"): + print(f" НЕДОСТУПЕН — {b.get('error')}") + continue + line = f" доступен; модель {'есть' if b.get('installed') else 'НЕ НАЙДЕНА'}" + if "loaded" in b: + line += f"; в памяти: {'да' if b['loaded'] else 'нет (холодный старт ~20 с)'}" + if b["breaker"].get("until", 0) > __import__("time").time(): + line += f"; ПРЕДОХРАНИТЕЛЬ: {b['breaker'].get('error')}" + print(line) + elif args.cmd == "events": es.init_db() if args.after is None: diff --git a/docs/KEYS-REGISTRY.md b/docs/KEYS-REGISTRY.md index e349287..e479752 100644 --- a/docs/KEYS-REGISTRY.md +++ b/docs/KEYS-REGISTRY.md @@ -19,6 +19,10 @@ Push теперь идёт через admin-токен 9f29709… (значен Диагностика: `scripts/deploy_channel_probe.py`. Правка: менять только pushurl, fetch остаётся на узком токене. Деплой лендинга: docs/DEPLOY-LANDING.md. +Ollama Cloud для decision-модели харнеса (clef:27b, /v1/systemone): +`SIMBIOT_DECISION_API_KEY` в `/var/lib/simbiot-harness/env/.env` (600), +подхватывается `harness/decision.py`; бэкап до правки — `.env.pre-decision`. + ## 2. Наследие агента Гермес (живое, сервисы active) | Путь | Что | Права | Примечание | diff --git a/harness/decision.py b/harness/decision.py new file mode 100644 index 0000000..a89d096 --- /dev/null +++ b/harness/decision.py @@ -0,0 +1,391 @@ +"""Simbiot Harness — decision-модель (Ollama /v1/systemone: облачная clef → локальная nimble). + +Схема из systemone-bench (ga/router42.py, последняя версия): детерминированная +механика остаётся быстрым путём, decision-модель — арбитр там, где механика +слепа (FIFO по задачам, round-robin по темам, порог гейта). Недоступна +модель / таймаут / мусор в ответе → None, вызывающий берёт свой фолбэк. +Тик никогда не падает из-за модели. + +Контракт эндпоинта (README бенча, пробы 2026-10-04): +- choice: criteria — объект {ключ: описание}, 2–26 опций; ответ — ключ + + probabilities. Точность 100% на шортлисте 6 → 50% на 10, поэтому здесь + жёсткий кап MAX_OPTIONS = 9 (вызывающий режет шортлист сам); +- noul: P(«да») в [0, 1]; +- confidence — концентрация распределения, НЕ вероятность правильности. + +Бэкенды по порядку (у каждого свой предохранитель): +1) основной — SIMBIOT_DECISION_URL/MODEL/API_KEY. С ключом Ollama Cloud: + https://ollama.com/v1/systemone, clef:27b — ~0.6–0.8 с на запрос; +2) запасной — локальная nimble (SIMBIOT_DECISION_FALLBACK_URL/MODEL, + SIMBIOT_DECISION_FALLBACK=off — выключить): CPU, 10–30 с, холодный + старт ~20 с, зато без сети и квот. +Без API_KEY и URL основной = локальная nimble, запасного нет (как раньше). + +Ответы детерминированы (один и тот же запрос → те же вероятности), поэтому +кеш по хешу вопроса+состояния валиден: одинаковая очередь в соседних тиках +не стоит повторного прогона. + +Режим — SIMBIOT_DECISION (env процесса сильнее .env инстанса): + off (дефолт) — модель не зовётся; on — решения применяются; + shadow — модель зовётся и решение пишется событием, но не применяется. +Точечный оверрайд потребителя: SIMBIOT_DECISION_=off|shadow|on +(например SIMBIOT_DECISION_GATE=on). Гейт по умолчанию — shadow: пропуск +LLM-тика по мнению модели включается только явно. + +Каждый реальный вызов пишет DecisionMade (purpose, опции, выбор, +вероятности, латентность, applied) — policy trace для будущей GA-сшивки +с исходом (skills/systemone-bench, §4 «Интеграция в тик харнеса»). +""" +import hashlib +import json +import os +import threading +import time +import urllib.error +import urllib.request + +from . import paths as _paths + +MAX_OPTIONS = 9 # шортлист stage2 строго ≤ 9 (бенч: 100% на 6 → 50% на 10) +CACHE_TTL = 3600.0 # ответы детерминированы — час держим +CACHE_MAX = 256 +BREAKER_COOLDOWN = 300.0 # после сбоя молчим 5 минут, не тормозим тики таймаутами + +_LOCK = threading.Lock() +_CACHE = {} # sha256(вопросы+состояние) → (ts, answers, model) +_BREAKERS = {} # имя бэкенда → {"until", "error"} + +LOCAL_URL = "http://127.0.0.1:11434/v1/systemone" +CLOUD_URL = "https://ollama.com/v1/systemone" + +# тексты ниже уходят в модель — на английском: nimble (qwen3.5) калибровался +# в бенче на английских инструкциях +AUM_GOAL = ("Goal of the harness: maximize long-term AUM of the cooperative " + "(income and resources of all projects over a year+ horizon). " + "Prefer work that adds >=1% to AUM in a year over short wins that " + "eat the long-term base.") + + +# ────────────────────────── конфиг ────────────────────────── + +def _cfg(key: str, default: str = "") -> str: + """env процесса > .env инстанса (как runtime._env_or_file, без импорта runtime).""" + val = os.environ.get(key, "").strip() + if val: + return val + try: + with open(_paths.env_path(), encoding="utf-8") as f: + for line in f: + if line.startswith(key + "="): + return line.split("=", 1)[1].strip().strip('"').strip("'") or default + except OSError: + pass + return default + + +def _float(key: str, default: float) -> float: + try: + return float(_cfg(key, str(default))) + except ValueError: + return default + + +def backends() -> list: + """[{name, url, model, key, timeout}] в порядке опроса.""" + key = _cfg("SIMBIOT_DECISION_API_KEY") + if key: + primary = {"name": "primary", "key": key, + "url": _cfg("SIMBIOT_DECISION_URL", CLOUD_URL), + "model": _cfg("SIMBIOT_DECISION_MODEL", "clef:27b"), + "timeout": _float("SIMBIOT_DECISION_TIMEOUT", 15.0)} + else: + primary = {"name": "primary", "key": "", + "url": _cfg("SIMBIOT_DECISION_URL", LOCAL_URL), + "model": _cfg("SIMBIOT_DECISION_MODEL", "nimble"), + "timeout": _float("SIMBIOT_DECISION_TIMEOUT", 90.0)} + out = [primary] + fb_default = "on" if primary["key"] else "off" # запасной — только за облаком + if _cfg("SIMBIOT_DECISION_FALLBACK", fb_default).lower() == "on": + out.append({"name": "fallback", "key": "", + "url": _cfg("SIMBIOT_DECISION_FALLBACK_URL", LOCAL_URL), + "model": _cfg("SIMBIOT_DECISION_FALLBACK_MODEL", "nimble"), + "timeout": _float("SIMBIOT_DECISION_FALLBACK_TIMEOUT", 90.0)}) + return out + + +def url() -> str: + return backends()[0]["url"] + + +def model_name() -> str: + return backends()[0]["model"] + + +_PURPOSE_DEFAULTS = {"gate": "shadow"} + + +def mode(purpose: str = "") -> str: + """off | shadow | on для потребителя purpose (глобальный режим — потолок).""" + glob = _cfg("SIMBIOT_DECISION", "off").lower() + if glob not in ("on", "shadow"): + return "off" + if purpose: + local = _cfg("SIMBIOT_DECISION_" + purpose.upper(), + _PURPOSE_DEFAULTS.get(purpose, glob)).lower() + if local not in ("on", "shadow"): + return "off" + if glob == "shadow": + return "shadow" + return local + return glob + + +# ────────────────────────── транспорт ────────────────────────── + +def _post(be: dict, body: dict) -> dict: + raw = json.dumps(body, ensure_ascii=False).encode() + headers = {"Content-Type": "application/json"} + if be["key"]: + headers["Authorization"] = "Bearer " + be["key"] + req = urllib.request.Request(be["url"], data=raw, method="POST", headers=headers) + try: + with urllib.request.urlopen(req, timeout=be["timeout"]) as r: + return json.loads(r.read().decode()) + except urllib.error.HTTPError as e: + # тело ошибки systemone информативно («model not found», «not a decision model») + raise ValueError(f"HTTP {e.code}: {e.read().decode(errors='replace')[:160]}") + + +def ask(questions: dict, state: str): + """Сырой вызов: {qid: question} → (answers, meta) | (None, meta). + + Бэкенды по порядку (backends()); сбой бэкенда взводит ЕГО предохранитель + на BREAKER_COOLDOWN и сразу пробует следующий. meta: {"cached", "ms", + "error", "model", "backend"}; error — сводка сбоев, если все упали. + """ + state = state[:24000] + key = hashlib.sha256(json.dumps({"s": state, "q": questions}, sort_keys=True, + ensure_ascii=False).encode()).hexdigest() + now = time.time() + with _LOCK: + hit = _CACHE.get(key) + if hit and now - hit[0] < CACHE_TTL: + return hit[1], {"cached": True, "ms": 0, "error": "", "model": hit[2], + "backend": "cache"} + errors = [] + t0 = time.time() + for be in backends(): + with _LOCK: + br = _BREAKERS.get(be["name"]) or {} + if time.time() < br.get("until", 0.0): + errors.append(f"{be['name']}: breaker ({br.get('error')})") + continue + body = {"model": be["model"], "state": state, "questions": questions} + if not be["key"]: + body["keep_alive"] = _cfg("SIMBIOT_DECISION_KEEP_ALIVE", "2h") + t1 = time.time() + try: + d = _post(be, body) + answers = d.get("answers") + if not isinstance(answers, dict): + raise ValueError(f"нет answers: {str(d)[:120]}") + except (urllib.error.URLError, OSError, ValueError, TimeoutError) as e: + err = f"{e.__class__.__name__}: {str(e)[:160]}" + with _LOCK: + _BREAKERS[be["name"]] = {"until": time.time() + BREAKER_COOLDOWN, + "error": err} + errors.append(f"{be['name']} {be['model']}: {err} " + f"({int((time.time() - t1) * 1000)} мс)") + continue + with _LOCK: + if len(_CACHE) >= CACHE_MAX: + for k in sorted(_CACHE, key=lambda k: _CACHE[k][0])[:CACHE_MAX // 4]: + _CACHE.pop(k, None) + _CACHE[key] = (time.time(), answers, be["model"]) + _BREAKERS.pop(be["name"], None) + return answers, {"cached": False, "ms": int((time.time() - t1) * 1000), + "error": "; ".join(errors), "model": be["model"], + "backend": be["name"]} + return None, {"cached": False, "ms": int((time.time() - t0) * 1000), + "error": "; ".join(errors) or "нет бэкендов", "model": None, + "backend": None} + + +def _record(purpose: str, kind: str, payload: dict, meta: dict, applied: bool): + """DecisionMade в стор — только на реальный вызов (кеш-хиты не спамят стор).""" + if meta.get("cached"): + return + try: + from . import es + rid = hashlib.sha1(json.dumps(payload, sort_keys=True, ensure_ascii=False) + .encode()).hexdigest()[:12] + es.append_event("decision", f"{purpose}:{int(time.time())}:{rid}", "DecisionMade", { + "purpose": purpose, "kind": kind, "model": meta.get("model"), + "backend": meta.get("backend"), + "ms": meta.get("ms"), "error": meta.get("error") or None, + "applied": applied, **payload, + }) + except Exception: + pass # трасса — best effort, решение важнее записи + + +# ────────────────────────── примитивы ────────────────────────── + +def choose(purpose: str, options: dict, state: str, instructions: str): + """Выбор одного ключа из options {ключ: описание} (2..MAX_OPTIONS). + + Возвращает {"choice", "probabilities", "confidence", "applied"} или None + (режим off, <2 опций, модель недоступна, ответ вне опций). applied=False + в shadow — вызывающий обязан тогда остаться на своём фолбэке. + """ + m = mode(purpose) + if m == "off" or len(options) < 2: + return None + opts = dict(list(options.items())[:MAX_OPTIONS]) + crit = {k: (v or k)[:400] for k, v in opts.items()} + answers, meta = ask({"c": {"type": "choice", "instructions": instructions, + "criteria": crit}}, state) + a = (answers or {}).get("c") or {} + pick = a.get("choice") + if pick not in opts: + _record(purpose, "choice", {"options": list(opts), "choice": None}, meta, False) + return None + probs = {k: round(float(v), 4) for k, v in (a.get("probabilities") or {}).items()} + applied = m == "on" + _record(purpose, "choice", {"options": list(opts), "choice": pick, + "probabilities": probs}, meta, applied) + return {"choice": pick, "probabilities": probs, + "confidence": round(float(a.get("confidence") or 0), 4), + "applied": applied} + + +def yes_prob(purpose: str, question: str, state: str): + """noul: P(«да») на вопрос. {"p", "applied"} или None.""" + m = mode(purpose) + if m == "off": + return None + answers, meta = ask({"n": {"type": "noul", "instructions": question}}, state) + p = ((answers or {}).get("n") or {}).get("noul") + if not isinstance(p, (int, float)): + _record(purpose, "noul", {"question": question[:300], "p": None}, meta, False) + return None + applied = m == "on" + _record(purpose, "noul", {"question": question[:300], "p": round(float(p), 4)}, + meta, applied) + return {"p": float(p), "applied": applied} + + +# ────────────────────────── потребители харнеса ────────────────────────── + +def _policy_line(model: dict) -> str: + pol = (model or {}).get("policy") or {} + aum = pol.get("aum_target") + if isinstance(aum, dict): + aum = json.dumps(aum, ensure_ascii=False) + return f"Current AUM policy: {str(aum)[:300]}" if aum else "" + + +def prioritize_tasks(level_name: str, tasks: list, model: dict): + """Какая из ready-задач (шортлист ≤ 9, старейшие) даст больше AUM — её первой. + + tasks — уже отсортированы FIFO. Возвращает id выбранной или None + (тогда остаётся FIFO). Состояние без времени и счётчиков — чтобы + одинаковая очередь попадала в кеш между тиками. + """ + short = tasks[:MAX_OPTIONS] + if len(short) < 2: + return None + opts = {} + for t in short: + desc = f"[{t.get('size')}] {t.get('title') or ''}" + if t.get("details"): + desc += " — " + t["details"][:120] + opts[t["id"]] = desc + state = "\n".join(x for x in ( + AUM_GOAL, _policy_line(model), + f"Harness level '{level_name}' picks the task to execute FIRST this tick. " + "Older tasks are listed first; a task blocking money, clients or " + "other people outweighs internal polish.") if x) + r = choose("tasks", opts, state, + "Which task should be executed first to maximize long-term AUM?") + return r["choice"] if r and r["applied"] else None + + +def gate_wants_llm(level_name: str, signals: list): + """Серая зона гейта (только warn/info-сигналы): стоит ли тратить кредиты роутера. + + Возвращает (allow: bool | None, p). None — решения нет/shadow: гейт + остаётся на пороговой механике. Порог SIMBIOT_DECISION_GATE_MIN_P (0.35). + """ + if not signals: + return None, None + lines = [f"- {s[0]}: {s[1]} — {s[2]}" for s in signals[:10]] + state = (AUM_GOAL + "\nA full LLM tick costs router credits (direct minus to AUM) " + f"and takes minutes. Level '{level_name}' sees these non-critical " + "signals:\n" + "\n".join(lines)) + r = yes_prob("gate", "Do these signals need attention of an expensive LLM " + "agent right now, rather than waiting for the next scheduled tick?", + state) + if not r: + return None, None + try: + thr = float(_cfg("SIMBIOT_DECISION_GATE_MIN_P", "0.35")) + except ValueError: + thr = 0.35 + if not r["applied"]: + return None, r["p"] + return r["p"] >= thr, r["p"] + + +def pick_topic(topics: list, counts: dict, model: dict): + """Тема инициативной реплики: шортлист ≤ 9 наименее использованных → choice. + + Возвращает id темы или None (фолбэк — round-robin). + """ + ranked = sorted(topics, key=lambda t: counts.get(t["id"], 0))[:MAX_OPTIONS] + if len(ranked) < 2: + return None + opts = {t["id"]: t["text"][:300] for t in ranked} + tasks = (model or {}).get("tasks") or {} + active = [t.get("title") or "" for t in tasks.values() + if t.get("status") in ("ready", "running")][:8] + state = "\n".join(x for x in ( + AUM_GOAL, _policy_line(model), + "The harness proactively writes one thought to the founder's chat. " + "Topics used less often are listed first.", + ("Work in progress: " + "; ".join(a[:80] for a in active)) if active else "", + ) if x) + r = choose("topic", opts, state, + "Which topic is the most useful to raise with the founder now " + "to grow long-term AUM?") + return r["choice"] if r and r["applied"] else None + + +def status() -> dict: + """Срез для CLI/дашборда: режимы и по каждому бэкенду — доступность, + модель установлена/в памяти (облаку /api/ps не нужен), предохранитель.""" + out = {"mode": mode(), "modes": {p: mode(p) for p in ("tasks", "gate", "topic")}, + "cache": len(_CACHE), "backends": []} + for be in backends(): + b = {"name": be["name"], "url": be["url"], "model": be["model"], + "cloud": bool(be["key"]), "breaker": dict(_BREAKERS.get(be["name"]) or {})} + base = be["url"].split("/v1/")[0] + hdr = {"Authorization": "Bearer " + be["key"]} if be["key"] else {} + want = be["model"] if ":" in be["model"] else be["model"] + ":latest" + try: + req = urllib.request.Request(base + "/api/tags", headers=hdr) + with urllib.request.urlopen(req, timeout=8) as r: + tags = json.loads(r.read().decode()) + names = {m.get("name", "") for m in tags.get("models") or []} + b["installed"] = be["model"] in names or want in names + if not be["key"]: + with urllib.request.urlopen(base + "/api/ps", timeout=5) as r: + ps = json.loads(r.read().decode()) + b["loaded"] = any(m.get("name") in (be["model"], want) + for m in ps.get("models") or []) + b["reachable"] = True + except Exception as e: + b["reachable"] = False + b["error"] = f"{e.__class__.__name__}: {str(e)[:120]}" + out["backends"].append(b) + return out diff --git a/harness/levels.py b/harness/levels.py index 2e40b3e..2be1dcf 100644 --- a/harness/levels.py +++ b/harness/levels.py @@ -127,9 +127,13 @@ LEVELS = { "cli.py escalate \"причина\" -f hour -s critical. " "Конфиг уровней — масштабированная контекст-схема по тактам (согласована " "Алексом, e-261001082710): min 256k → hour 512k → day 1M ctx_tokens. " - "Пачку задач бери СТРОГО по очереди — от старейшей к свежим, как " - "отдаёт очередь (анти-старение); не выбирай «поинтереснее» и " - "не пересортировывай пачку. " + "Пачку задач бери СТРОГО в порядке списка — очередь уже упорядочена " + "(FIFO + первой — задача, которую decision-модель сочла самой " + "ценной для AUM); не выбирай «поинтереснее» и не пересортировывай. " + "Выбор из 2–9 вариантов (как ответить, какой подход взять) — " + "дёшево и детерминированно: cli.py decide -q \"вопрос\" -o ключ=описание " + "-o ... -s \"ситуация\" (локальная nimble, без кредитов роутера; " + "без -o — P(да)). " "Файлы/уведомления в чат — только идемпотентно: cli.py tg-deliver-file " "<путь> -c \"подпись\", cli.py tg-notify \"текст\" -k <ключ> (повтор — SKIP). " "Прямые curl/api.telegram отправки из shell запрещены — нет защиты " @@ -151,6 +155,8 @@ LEVELS = { model="auto", effort="high", ctx_tokens=1_000_000, task_classes=("S", "M", "L"), mandate=("Ты — дневной цикл харнеса Simbiot. Манифест дня: план, выбор задач " "L-класса, дневной отчёт. Думай глубоко, планируй на день вперёд. " + "Выбор из 2–9 вариантов — cli.py decide -q \"вопрос\" -o ключ=описание " + "(локальная decision-модель nimble, без кредитов роутера). " "Открытые эскалации (секция ЭСКАЛАЦИИ) — приоритет: это важное, " "поднятое с нижних уровней; закрой cli.py escalation-ack . " "Файлы в чат — только cli.py tg-deliver-file (идемпотентно, " diff --git a/harness/proactive.py b/harness/proactive.py index 2a6eca1..d99bf40 100644 --- a/harness/proactive.py +++ b/harness/proactive.py @@ -202,11 +202,20 @@ def should_speak(model: dict, now: "datetime | None" = None): def pick_topic(model: dict): - """Следующая тема: наименее использованная (round-robin по банку).""" + """Следующая тема: наименее использованная (round-robin по банку). + + При включённой decision-модели (SIMBIOT_DECISION) — выбор nimble из + шортлиста ≤ 9 наименее использованных с учётом доски и AUM-политики; + модель недоступна — round-robin, как раньше. + """ init = model.get("initiative") or {} counts = {} for s in (init.get("sent") or []): counts[s.get("topic")] = counts.get(s.get("topic"), 0) + 1 + from . import decision + pick = decision.pick_topic(TOPICS, counts, model) + if pick: + return next(t for t in TOPICS if t["id"] == pick) return min(TOPICS, key=lambda t: counts.get(t["id"], 0)) diff --git a/harness/scheduler.py b/harness/scheduler.py index 0f00e9a..d66c900 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -157,7 +157,7 @@ def find_task(level_cfg, model: dict): return None -def select_task_batch(level_cfg, model: dict): +def select_task_batch(level_cfg, model: dict, prioritize: bool = False): """Батч ready-задач доступных классов в бюджете контекста тика. Возвращает [task_id, ...] — от старейших к свежим (анти-старение @@ -171,6 +171,11 @@ def select_task_batch(level_cfg, model: dict): мандат, выжимки, эскалации и инбокс остаются в тике. Первая (старейшая) задача берётся всегда — как раньше find_task; бюджетная отсечка работает начиная со второй пачки. + + prioritize=True (боевой тик): decision-модель (harness/decision.py, + режим SIMBIOT_DECISION) выбирает из шортлиста ≤ 9 старейших ту, что + больше всего даёт AUM, — она встаёт первой; остальные в FIFO. Модель + недоступна/выключена — чистый FIFO, как раньше. """ classes = level_cfg.task_classes if not classes: @@ -178,6 +183,11 @@ def select_task_batch(level_cfg, model: dict): ready = [t for t in (model.get("tasks") or {}).values() if t.get("status") == "ready" and t.get("size") in classes] ready.sort(key=lambda t: t.get("created") or "") + if prioritize and len(ready) > 1: + from . import decision + pick = decision.prioritize_tasks(level_cfg.name, ready, model) + if pick: + ready.sort(key=lambda t: t["id"] != pick) # stable: остальные — FIFO cap = _batch_cap(level_cfg) budget_chars = int(level_cfg.ctx_tokens * CHARS_PER_TOKEN * TASK_BUDGET_SHARE) picked, used = [], 0 @@ -447,7 +457,7 @@ def build_context(level_cfg, model: dict, batch=None): lines.append(f" ПОСТАНОВКА: {t['details'][:1800]}") parts.append( f"ТЕКУЩАЯ РАБОТА — БАТЧ из {len(batch)} задач (выполняй в этом тике, " - "самую старую — ПЕРВОЙ; не растягивай: тик ограничен таймаутом и " + "по порядку списка — первая самая важная; не растягивай: тик ограничен таймаутом и " f"бюджетом шагов; суммарная оценка ~{sum(_est_minutes(t) for t in batch)}м):\n" + "\n".join(lines) + "\nКаждую завершённую закрывай ОТДЕЛЬНО сразу после выполнения: " @@ -542,7 +552,7 @@ def run_level_tick(level_cfg, seq: int, model: dict): """ from . import isolated_runner - batch_ids = select_task_batch(level_cfg, model) + batch_ids = select_task_batch(level_cfg, model, prioritize=True) batch = [model["tasks"].get(tid) for tid in batch_ids] batch = [t for t in batch if t] claimed = [] @@ -877,7 +887,16 @@ def process_level(level_cfg, model: dict): except Exception: load = None signals = collect_signals(level_cfg, model, load) - if not gate_allows_llm(level_cfg, signals): + allow = gate_allows_llm(level_cfg, signals) + decided_p = None + if allow and signals and not any(s[0] == "critical" for s in signals): + # серая зона: порог пропустил только warn/info — спросить + # decision-модель, стоит ли тратить кредиты (critical — всегда LLM) + from . import decision + verdict, decided_p = decision.gate_wants_llm(level_cfg.name, signals) + if verdict is False: + allow = False + if not allow: period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc) es.append_event("cycle", f"digest:{level_cfg.name}:{seq}", "DigestWritten", { "level": level_cfg.name, "period": period.isoformat(), @@ -889,6 +908,7 @@ def process_level(level_cfg, model: dict): es.append_event("cycle", f"gate:{level_cfg.name}:{seq}", "GateSkipped", { "level": level_cfg.name, "threshold": _gate_threshold(level_cfg), + "decision_p": decided_p, "signals": [{"severity": s[0], "what": s[1], "detail": s[2]} for s in signals][:10], }) diff --git a/skills/systemone-bench/SKILL.md b/skills/systemone-bench/SKILL.md index fc08b58..c1861c2 100644 --- a/skills/systemone-bench/SKILL.md +++ b/skills/systemone-bench/SKILL.md @@ -72,6 +72,17 @@ random search до +59%. Что уже известно (не переоткры ### 4. Интеграция в тик харнеса +Внедрено: `harness/decision.py` (README харнеса, раздел «Decision-модель +nimble») — порядок батча задач, тема инициативы, серая зона гейта (shadow), +CLI `cli.py decide -q "вопрос" -o ключ=описание ... -s "ситуация"` для +выбора из 2–9 вариантов прямо из тика. Бэкенды: Ollama Cloud `clef:27b` +(~0.7 с, ключ SIMBIOT_DECISION_API_KEY) → запасная локальная nimble на +CPU (10–30 с). Из decision-моделей облака проверены: clef:27b — работает; +beam, muse-spark-open, glm-5.3-flash и прочие LLM — «not a decision +model»; nimble в облаке нет. Батчи вопросы в один запрос (до 64). + +Исходная схема: + На тиках, где нужен выбор действия: собрать кандидатов → разбить на группы → stage1 score (top-2) → choice ≤9 → применить → записать выбор и исход событием в стор (для последующей GA-сшивки policy trace с fitness). diff --git a/tests/test_decision.py b/tests/test_decision.py new file mode 100644 index 0000000..fbd36c9 --- /dev/null +++ b/tests/test_decision.py @@ -0,0 +1,225 @@ +# -*- coding: utf-8 -*- +"""decision-модель (harness/decision.py) — hermetic, фейковый /v1/systemone. + +Запуск: ./venv/bin/python tests/test_decision.py + +Проверяет: off — эндпоинт не зовётся и очередь FIFO; on — выбор модели +встаёт первым в батче, остальные FIFO; shadow — решение пишется, но не +применяется; шортлист ≤ 9; кеш (повтор не идёт в сеть, стор не спамится); +предохранитель (мёртвый эндпоинт → None быстро, второй вызов без сети); +гейт: серая зона p<порога → False, p≥порога → True, гейт по умолчанию shadow; +тема инициативы — выбор модели, при off — round-robin. +""" +import json +import os +import sys +import tempfile +import threading +import time +from http.server import BaseHTTPRequestHandler, HTTPServer + +sys.path.insert(0, "/root/simbiot-harness") + +_tmp = tempfile.mkdtemp(prefix="simbiot_decision_") +os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db") +for _v in [k for k in os.environ if k.startswith("SIMBIOT_DECISION")] + [ + "SIMBIOT_BATCH_MAX_MIN", "SIMBIOT_BATCH_MAX_HOUR", "SIMBIOT_BATCH_MAX_DAY"]: + os.environ.pop(_v, None) + +from harness import decision, es, scheduler, proactive +from harness.levels import LEVELS + +es.init_db() +fails = [] + + +def check(name, cond, extra=""): + print(("PASS" if cond else "FAIL"), name, ("| " + str(extra)) if not cond else "") + if not cond: + fails.append(name) + + +# ── фейковый эндпоинт: choice → ключ из FAKE["pick"] (или последний), noul → FAKE["p"] +FAKE = {"pick": None, "p": 0.2, "calls": [], "delay": 0.0} + + +class H(BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + body["_auth"] = self.headers.get("Authorization") + body["_path"] = self.path + FAKE["calls"].append(body) + time.sleep(FAKE["delay"]) + ans = {} + for qid, q in body["questions"].items(): + if q["type"] == "choice": + keys = list(q["criteria"]) + pick = FAKE["pick"] if FAKE["pick"] in keys else keys[-1] + ans[qid] = {"type": "choice", "choice": pick, "confidence": 0.5, + "probabilities": {k: (0.9 if k == pick else 0.1 / (len(keys) - 1)) + for k in keys}} + else: + ans[qid] = {"type": "noul", "noul": FAKE["p"]} + raw = json.dumps({"model": body["model"], "answers": ans}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(raw))) + self.end_headers() + self.wfile.write(raw) + + +srv = HTTPServer(("127.0.0.1", 0), H) +threading.Thread(target=srv.serve_forever, daemon=True).start() +os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{srv.server_port}/v1/systemone" + + +def reset(): + FAKE.update(pick=None, p=0.2, calls=[], delay=0.0) + decision._CACHE.clear() + decision._BREAKERS.clear() + + +def _task(tid, created, title="задача", size="M"): + return {"id": tid, "title": title, "size": size, "status": "ready", + "created": created, "details": ""} + + +def _model(n): + tasks = {f"t{i}": _task(f"t{i}", f"2026-10-0{i}T00:00:00" if i < 10 + else f"2026-10-{i}T00:00:00", title=f"задача {i}") + for i in range(1, n + 1)} + return {"tasks": tasks, "policy": {"aum_target": "AUM ≥ +1%/год"}} + + +def _decisions(): + return [json.loads(e["data"]) for e in es.get_events(0, 100000) + if e["event_type"] == "DecisionMade"] + + +hour = LEVELS["hour"] + +# 1) off по умолчанию: сеть не трогаем, FIFO +reset() +check("mode off по умолчанию", decision.mode() == "off" and decision.mode("tasks") == "off") +ids = scheduler.select_task_batch(hour, _model(4), prioritize=True) +check("off: FIFO", ids[:1] == ["t1"], ids) +check("off: эндпоинт не вызывался", not FAKE["calls"]) + +# 2) on: выбор модели — первым, остальные FIFO +os.environ["SIMBIOT_DECISION"] = "on" +reset() +FAKE["pick"] = "t3" +ids = scheduler.select_task_batch(hour, _model(4), prioritize=True) +check("on: выбор модели первым", ids[0] == "t3", ids) +check("on: остальные FIFO", ids[1:] == ["t1", "t2", "t4"], ids) +check("on: без prioritize — чистый FIFO", + scheduler.select_task_batch(hour, _model(4))[0] == "t1") +d = _decisions() +check("DecisionMade записан", d and d[-1]["purpose"] == "tasks" and d[-1]["choice"] == "t3" + and d[-1]["applied"] is True, d[-1:] if d else None) + +# 3) кеш: та же очередь — без сети и без нового события +n_calls, n_ev = len(FAKE["calls"]), len(_decisions()) +scheduler.select_task_batch(hour, _model(4), prioritize=True) +check("кеш: повтор без сети", len(FAKE["calls"]) == n_calls) +check("кеш: стор не спамится", len(_decisions()) == n_ev) + +# 4) шортлист ≤ 9 старейших +reset() +scheduler.select_task_batch(hour, _model(14), prioritize=True) +crit = FAKE["calls"][-1]["questions"]["c"]["criteria"] +check("шортлист ≤ 9", len(crit) == 9, len(crit)) +check("шортлист — старейшие", list(crit) == [f"t{i}" for i in range(1, 10)], list(crit)) +check("keep_alive передан", FAKE["calls"][-1].get("keep_alive")) + +# 5) shadow: решение записано, но FIFO +os.environ["SIMBIOT_DECISION"] = "shadow" +reset() +FAKE["pick"] = "t2" +ids = scheduler.select_task_batch(hour, _model(3), prioritize=True) +check("shadow: FIFO остаётся", ids[0] == "t1", ids) +check("shadow: решение записано с applied=False", + _decisions()[-1]["choice"] == "t2" and _decisions()[-1]["applied"] is False) + +# 6) предохранитель: мёртвый эндпоинт → None, второй вызов без сети +os.environ["SIMBIOT_DECISION"] = "on" +reset() +os.environ["SIMBIOT_DECISION_URL"] = "http://127.0.0.1:9/v1/systemone" +ids = scheduler.select_task_batch(hour, _model(3), prioritize=True) +check("мёртвый эндпоинт: FIFO", ids[0] == "t1", ids) +check("предохранитель взведён", decision._BREAKERS["primary"]["until"] > time.time()) +t0 = time.time() +r = decision.choose("tasks", {"a": "x", "b": "y"}, "s", "q") +check("предохранитель: мгновенный None", r is None and time.time() - t0 < 0.5) +os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{srv.server_port}/v1/systemone" + +# 6b) облако + локальный запасной: ключ в Authorization, без keep_alive; +# облако упало → ответ запасного, бэкенд записан в DecisionMade +reset() +port = srv.server_port +os.environ["SIMBIOT_DECISION_API_KEY"] = "k-test" +os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{port}/cloud/v1/systemone" +os.environ["SIMBIOT_DECISION_FALLBACK_URL"] = f"http://127.0.0.1:{port}/local/v1/systemone" +bes = decision.backends() +check("с ключом: облако clef:27b + запасная nimble", + [b["model"] for b in bes] == ["clef:27b", "nimble"], bes) +FAKE["pick"] = "b" +r = decision.choose("agent", {"a": "x", "b": "y"}, "s-cloud", "q") +c = FAKE["calls"][-1] +check("облако: выбор", r and r["choice"] == "b", r) +check("облако: Bearer-ключ", c["_auth"] == "Bearer k-test", c["_auth"]) +check("облако: без keep_alive", "keep_alive" not in c) +check("облако: путь основного", c["_path"].startswith("/cloud/"), c["_path"]) +reset() +os.environ["SIMBIOT_DECISION_URL"] = "http://127.0.0.1:9/v1/systemone" +FAKE["pick"] = "a" +r = decision.choose("agent", {"a": "x", "b": "y"}, "s-fallback", "q") +c = FAKE["calls"][-1] if FAKE["calls"] else {} +check("облако упало → запасной ответил", r and r["choice"] == "a", r) +check("запасной: путь локального, без ключа", + c.get("_path", "").startswith("/local/") and c.get("_auth") is None, c.get("_path")) +check("запасной: предохранитель облака взведён", + decision._BREAKERS.get("primary", {}).get("until", 0) > time.time()) +dm = _decisions()[-1] +check("DecisionMade: backend=fallback, model=nimble, ошибка облака сохранена", + dm.get("backend") == "fallback" and dm.get("model") == "nimble" and dm.get("error"), dm) +for _v in ("SIMBIOT_DECISION_API_KEY", "SIMBIOT_DECISION_FALLBACK_URL"): + os.environ.pop(_v) +os.environ["SIMBIOT_DECISION_URL"] = f"http://127.0.0.1:{port}/v1/systemone" +check("без ключа: только локальная (как раньше)", + [b["name"] for b in decision.backends()] == ["primary"]) + +# 7) гейт: по умолчанию shadow (не влияет), с GATE=on — по порогу +reset() +sigs = [("warn", "escalation:e-1", "что-то не так")] +check("гейт по умолчанию shadow", decision.mode("gate") == "shadow") +v, p = decision.gate_wants_llm("hour", sigs) +check("гейт shadow: вердикта нет, p записан", v is None and abs(p - 0.2) < 1e-9, (v, p)) +os.environ["SIMBIOT_DECISION_GATE"] = "on" +reset() +FAKE["p"] = 0.1 +check("гейт on: p<порога → False", decision.gate_wants_llm("hour", sigs)[0] is False) +reset() +FAKE["p"] = 0.8 +check("гейт on: p≥порога → True", decision.gate_wants_llm("hour", sigs)[0] is True) +os.environ.pop("SIMBIOT_DECISION_GATE") + +# 8) тема инициативы: выбор модели; при off — round-robin +reset() +FAKE["pick"] = "cyberbank" +check("тема: выбор модели", proactive.pick_topic({"tasks": {}})["id"] == "cyberbank") +os.environ["SIMBIOT_DECISION"] = "off" +check("тема off: round-robin", proactive.pick_topic({"tasks": {}})["id"] == proactive.TOPICS[0]["id"]) + +# 9) глобальный shadow — потолок для точечного on +os.environ["SIMBIOT_DECISION"] = "shadow" +os.environ["SIMBIOT_DECISION_TASKS"] = "on" +check("глобальный shadow сильнее точечного on", decision.mode("tasks") == "shadow") +os.environ.pop("SIMBIOT_DECISION_TASKS") + +srv.shutdown() +print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)") +sys.exit(1 if fails else 0)