From ebcfe3b9be487a5c2c616c90b6a1bf5db5a72a61 Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Sat, 17 Oct 2026 09:32:41 +0000 Subject: [PATCH] =?UTF-8?q?=D1=8F=D0=B4=D1=80=D0=BE=20=D0=9A=D0=BE=D0=B4?= =?UTF-8?q?=D0=B8:=20=D1=8D=D1=81=D0=BA=D0=B0=D0=BB=D0=B0=D1=86=D0=B8?= =?UTF-8?q?=D1=8F=20=D0=BD=D0=B5=D0=B4=D0=BE=D0=B4=D0=B5=D0=BB=D0=B0=D0=BD?= =?UTF-8?q?=D0=BD=D1=8B=D1=85=20=D0=B7=D0=B0=D0=B4=D0=B0=D1=87,=20=D0=B3?= =?UTF-8?q?=D0=BE=D0=BB=D0=BE=D0=B2=D0=B0=20task=5Ffit,=20=D1=80=D0=B5?= =?UTF-8?q?=D1=84=D0=BB=D0=B5=D0=BA=D1=81=D0=B8=D1=8F,=20=D0=BD=D0=B5?= =?UTF-8?q?=D0=B4=D0=B5=D0=BB=D1=8C=D0=BD=D1=8B=D0=B9=20=D0=BE=D1=82=D1=87?= =?UTF-8?q?=D1=91=D1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Разбор гейта (трассы с 15.10): hour — 0% пустых тиков, min — 6% (15% токенов уровня) → модель-гейт не нужен. Пустые мин-тики — зацикливание: «бюджет исчерпан» возвращал S как S, мин брал её снова с нуля (S-задача 33 захвата за 2 ч; 45 из 272 задач ≥ 2 мин-захватов). - requeue_or_escalate: после 2 захватов на уровне (SIMBIOT_TASK_ATTEMPTS_ PER_LEVEL) размер растёт выше классов уровня (S→M, к часу, 300 шагов) — и для guard «бюджет исчерпан», и для смерти воркера; счётчик попыток best effort (сбой → обычный requeue, не ломает путь падения воркера); - kodi_core — несколько голов (общие: хранение/обучение/промоут/откат, nf в модели; SIMBIOT_CORE_<ГОЛОВА>). Голова task_fit: влезет ли S в мин-тик (метки — из истории задач); shadow — прогноз перед первым мин-захватом, on — уверенное «не влезет» сразу S→M. Честно: на 93 задачах 5 фолдов 0.58–0.62 при базе 0.52, уверенных 0% — пока инертна; - рефлексия (как reflexion.py): ≥ 3 расхождений вердикта адресации с меткой человека/чата или ≥ 3 S-задач, не влезших в мин → задача часу с автодоставкой: разобрать и предложить Alex правило (код не трогать); - недельный отчёт ядра (понедельник, без LLM, идемпотентно по неделе); ночью: train_all → reflect → weekly_report; cli.py core reflect|report|--head. Тесты: test_task_escalation 8, test_kodi_core_tasks 13, test_kodi_core_ reflect 10, test_kodi_core 28 PASS; полный набор — без регрессий. Co-Authored-By: Claude Opus 5.5 --- cli.py | 13 +- harness/kodi_core.py | 425 ++++++++++++++++++++++++++------ harness/scheduler.py | 96 +++++++- tests/test_kodi_core.py | 6 +- tests/test_kodi_core_reflect.py | 89 +++++++ tests/test_kodi_core_tasks.py | 129 ++++++++++ tests/test_task_escalation.py | 89 +++++++ 7 files changed, 761 insertions(+), 86 deletions(-) create mode 100644 tests/test_kodi_core_reflect.py create mode 100644 tests/test_kodi_core_tasks.py create mode 100644 tests/test_task_escalation.py diff --git a/cli.py b/cli.py index 4012a96..bf74cb5 100644 --- a/cli.py +++ b/cli.py @@ -203,7 +203,9 @@ def main(): help="проверить все архивы: sha256 против StoreCompacted + подпись каждой строки") p_core = sub.add_parser("core", help="ядро Коди: обучаемые ученики решений (status|train|backfill|rollback)") - p_core.add_argument("action", choices=["status", "train", "backfill", "rollback", "feedback"]) + p_core.add_argument("action", choices=["status", "train", "backfill", "rollback", "feedback", + "reflect", "report"]) + p_core.add_argument("--head", default=None, help="голова: addressing | task_fit (train/rollback; без — все для train)") p_dec = sub.add_parser("decide", help="decision-модель nimble (Ollama /v1/systemone): статус, вопрос, статистика") p_dec.add_argument("-q", "--question", help="вопрос/инструкция (без -o — noul: P(да))") @@ -661,13 +663,18 @@ def main(): if args.action == "status": print(json.dumps(kodi_core.status(), ensure_ascii=False, indent=1)) elif args.action == "train": - print(json.dumps(kodi_core.train(), ensure_ascii=False, default=str)) + res = (kodi_core.train(head=args.head) if args.head else kodi_core.train_all()) + print(json.dumps(res, ensure_ascii=False, default=str)) elif args.action == "backfill": print(f"меток учителя: {kodi_core.backfill()}") elif args.action == "feedback": print(f"обратная связь: {kodi_core.harvest_feedback()}") + elif args.action == "reflect": + print(f"задачи рефлексии: {kodi_core.reflect()}") + elif args.action == "report": + print(kodi_core.weekly_report(force=True) or "SKIP: уже отправлен на этой неделе") else: - print(json.dumps(kodi_core.rollback(), ensure_ascii=False)) + print(json.dumps(kodi_core.rollback(args.head or kodi_core.HEAD), ensure_ascii=False)) elif args.cmd == "decide": from harness import decision diff --git a/harness/kodi_core.py b/harness/kodi_core.py index 2b8daf7..8dac9ea 100644 --- a/harness/kodi_core.py +++ b/harness/kodi_core.py @@ -77,21 +77,22 @@ SECOND_RE = re.compile(r"\b(?:ты|тебе|тебя|твой|твоя|твоё| LINK_RE = re.compile(r"https?://") _LOCK = threading.Lock() -_STATE = {"model": None, "loaded_sha": None} +_STATE = {} # голова → {"model", "sha"} # ────────────────────────── конфиг/пути ────────────────────────── -def mode() -> str: +def mode(head: str = HEAD) -> str: + """SIMBIOT_CORE_<ГОЛОВА>=shadow|on|off (дефолт shadow).""" from . import decision - v = decision._cfg("SIMBIOT_CORE_ADDRESSING", "shadow").lower() + v = decision._cfg("SIMBIOT_CORE_" + head.upper(), "shadow").lower() return v if v in ("on", "shadow", "off") else "shadow" def core_dir() -> str: d = (os.environ.get("SIMBIOT_CORE_DIR", "").strip() or os.path.join(_paths.home(), "core")) - os.makedirs(os.path.join(d, HEAD), mode=0o700, exist_ok=True) + os.makedirs(d, mode=0o700, exist_ok=True) return d @@ -296,16 +297,17 @@ def _prep(model, X): """Эмбеддинги — как есть (×EMB_SCALE), стандартизуются только флаги. Стандартизация каждого из 1536 измерений на сотнях примеров раздувала шумовые измерения до масштаба полезных — регрессия учила шум.""" + nf = int(model.get("nf", NF)) Z = X.astype(np.float64, copy=True) - Z[:, :-NF] *= EMB_SCALE - Z[:, -NF:] = (Z[:, -NF:] - model["mu"]) / model["sd"] + Z[:, :-nf] *= EMB_SCALE + Z[:, -nf:] = (Z[:, -nf:] - model["mu"]) / model["sd"] return Z -def train_logreg(X, y, w, l2=0.05, epochs=800, lr=0.2): +def train_logreg(X, y, w, l2=0.05, epochs=800, lr=0.2, nf=NF): """Логистическая регрессия: взвешенный log-loss + L2, градиентный спуск (Adam).""" - fl = X[:, -NF:].astype(np.float64) - model = {"mu": fl.mean(0), "sd": fl.std(0) + 1e-6} + fl = X[:, -nf:].astype(np.float64) + model = {"mu": fl.mean(0), "sd": fl.std(0) + 1e-6, "nf": nf} Z = _prep(model, X) wt = np.zeros(Z.shape[1]); b = 0.0 sw = w / w.sum() @@ -330,7 +332,6 @@ def predict(model, X): def novelty(model, em): """1 − средняя косинусная близость к 5 ближайшим из обучающей памяти.""" - import numpy as np mem = model.get("mem") if mem is None or not len(mem): return np.ones(len(em)) @@ -341,85 +342,101 @@ def novelty(model, em): def metrics(model, X, y, w): - import numpy as np p = np.clip(predict(model, X), 1e-6, 1 - 1e-6) ll = float(-(w * (y * np.log(p) + (1 - y) * np.log(1 - p))).sum() / w.sum()) acc = float((w * ((p >= .5) == (y >= .5))).sum() / w.sum()) return {"logloss": round(ll, 4), "acc": round(acc, 4), "n": int(len(y))} -def _model_path(name): - return os.path.join(core_dir(), HEAD, name) +def _model_path(name, head: str = HEAD): + d = os.path.join(core_dir(), head) + os.makedirs(d, mode=0o700, exist_ok=True) + return os.path.join(d, name) -def _champion_meta(): +def _champion_meta(head: str = HEAD): try: - with open(_model_path("champion.json")) as f: + with open(_model_path("champion.json", head)) as f: return json.load(f) except (OSError, ValueError): return {} -def load_champion(): - """Чемпион из файла (кеш по sha256) или None.""" - import numpy as np - meta = _champion_meta() +def load_champion(head: str = HEAD): + """Чемпион головы из файла (кеш по sha256) или None.""" + meta = _champion_meta(head) if not meta.get("file"): return None - if _STATE["loaded_sha"] == meta.get("sha256") and _STATE["model"] is not None: - return _STATE["model"] - path = _model_path(meta["file"]) + st = _STATE.setdefault(head, {}) + if st.get("sha") == meta.get("sha256") and st.get("model") is not None: + return st["model"] + path = _model_path(meta["file"], head) with open(path, "rb") as f: blob = f.read() if hashlib.sha256(blob).hexdigest() != meta.get("sha256"): - return None # файл подменён — ученика нет, решает учитель + return None # файл подменён — ученика нет, решает учитель/правило z = np.load(path) model = {k: z[k] for k in ("w", "mu", "sd", "mem")} model["b"] = float(z["b"]) + model["nf"] = int(z["nf"]) if "nf" in z else NF model["nov_q"] = z["nov_q"] if "nov_q" in z else np.asarray([NOV_FAMILIAR, NOV_NOVEL]) - _STATE.update(model=model, loaded_sha=meta["sha256"]) + st.update(model=model, sha=meta["sha256"]) return model -def _save(model, metrics_, n_total): +def _forget(head: str): + _STATE.pop(head, None) + + +def _save(model, metrics_, n_total, head: str = HEAD, flag_names=FLAG_NAMES): import io - import numpy as np name = f"model-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}.npz" buf = io.BytesIO() np.savez(buf, w=model["w"], b=np.float64(model["b"]), mu=model["mu"], - sd=model["sd"], mem=model["mem"], + sd=model["sd"], mem=model["mem"], nf=np.int64(model.get("nf", NF)), nov_q=np.asarray(model.get("nov_q", [NOV_FAMILIAR, NOV_NOVEL]))) blob = buf.getvalue() - path = _model_path(name) + path = _model_path(name, head) with open(path, "wb") as f: f.write(blob) os.chmod(path, 0o600) sha = hashlib.sha256(blob).hexdigest() - prev = _champion_meta() + prev = _champion_meta(head) meta = {"file": name, "sha256": sha, "metrics": metrics_, "n": n_total, - "kind": "logreg", "features": {"emb": EMB_MODEL, "flags": FLAG_NAMES}, + "kind": "logreg", "features": {"emb": EMB_MODEL, "flags": list(flag_names)}, "ts": datetime.now(timezone.utc).isoformat(), "history": ([{k: prev[k] for k in ("file", "sha256", "metrics", "ts") if k in prev}] + prev.get("history", []))[:10] if prev.get("file") else []} - tmp = _model_path(".champion.tmp") + tmp = _model_path(".champion.tmp", head) with open(tmp, "w") as f: json.dump(meta, f, ensure_ascii=False, indent=1) - os.replace(tmp, _model_path("champion.json")) + os.replace(tmp, _model_path("champion.json", head)) return meta -def train(log=print) -> dict: +def _heads(): + """Реестр голов: dataset() → [(item, y, w, src)], featurize(items) → (X, em).""" + return { + HEAD: {"dataset": lambda: [((r[0], r[1]), r[2], r[3], r[4]) for r in dataset()], + "featurize": featurize, "nf": NF, "flags": FLAG_NAMES}, + TASK_HEAD: {"dataset": task_dataset, "featurize": task_featurize, + "nf": len(TASK_FLAGS), "flags": TASK_FLAGS}, + } + + +def train(log=print, head: str = HEAD) -> dict: """Претендент против чемпиона на временном холдауте; лучше — промоут.""" - import numpy as np - rows = dataset() + spec = _heads()[head] + rows = spec["dataset"]() if len(rows) < MIN_TRAIN: - log(f"[core] {HEAD}: примеров {len(rows)} < {MIN_TRAIN} — обучение отложено") - return {"promoted": False, "n": len(rows), "reason": "few_samples"} - X, em = featurize([(r[0], r[1]) for r in rows]) - y = np.asarray([r[2] for r in rows], np.float64) - w = np.asarray([r[3] for r in rows], np.float64) + log(f"[core] {head}: примеров {len(rows)} < {MIN_TRAIN} — обучение отложено") + return {"head": head, "promoted": False, "n": len(rows), "reason": "few_samples"} + X, em = spec["featurize"]([r[0] for r in rows]) + y = np.asarray([r[1] for r in rows], np.float64) + w = np.asarray([r[2] for r in rows], np.float64) + nf = spec["nf"] cut = max(int(len(rows) * 0.8), len(rows) - max(10, len(rows) // 5)) - ch = train_logreg(X[:cut], y[:cut], w[:cut]) + ch = train_logreg(X[:cut], y[:cut], w[:cut], nf=nf) ch["mem"] = em[:cut] m_ch = metrics(ch, X[cut:], y[cut:], w[cut:]) nov_h = novelty(ch, em[cut:]) @@ -429,8 +446,9 @@ def train(log=print) -> dict: m_ch["confident_share"] = round(float(sure.mean()), 3) m_ch["confident_acc"] = (round(float(((p_h[sure] >= .5) == (y[cut:][sure] >= .5)).mean()), 3) if sure.any() else None) + m_ch["base_rate"] = round(float((y[cut:] >= .5).mean()), 3) m_ch["nov_q"] = [round(float(v), 4) for v in nov_q] - champ = load_champion() + champ = load_champion(head) m_old = None if champ is not None: try: @@ -440,38 +458,50 @@ def train(log=print) -> dict: better = m_old is None or m_ch["logloss"] < m_old["logloss"] - 1e-4 srcs = {} for r in rows: - srcs[r[4]] = srcs.get(r[4], 0) + 1 - rep = {"head": HEAD, "n": len(rows), "holdout": m_ch, "champion_holdout": m_old, + srcs[r[3]] = srcs.get(r[3], 0) + 1 + rep = {"head": head, "n": len(rows), "holdout": m_ch, "champion_holdout": m_old, "sources": srcs, "promoted": bool(better)} if better: - full = train_logreg(X, y, w) + full = train_logreg(X, y, w, nf=nf) full["mem"] = em full["nov_q"] = nov_q - meta = _save(full, m_ch, len(rows)) - _STATE.update(model=None, loaded_sha=None) - es.append_event("core", f"promote:{HEAD}:{meta['file']}", "CoreModelPromoted", { - "head": HEAD, "file": meta["file"], "sha256": meta["sha256"], + meta = _save(full, m_ch, len(rows), head, spec["flags"]) + _forget(head) + es.append_event("core", f"promote:{head}:{meta['file']}", "CoreModelPromoted", { + "head": head, "file": meta["file"], "sha256": meta["sha256"], "holdout": m_ch, "champion_holdout": m_old, "n": len(rows), "sources": srcs}) - log(f"[core] {HEAD}: n={len(rows)} претендент {m_ch} чемпион {m_old} → " + log(f"[core] {head}: n={len(rows)} претендент {m_ch} чемпион {m_old} → " f"{'ПРОМОУТ' if better else 'чемпион остаётся'}") return rep -def rollback() -> dict: - meta = _champion_meta() +def train_all(log=print) -> list: + """Ночная тренировка всех голов (дневной тик).""" + out = [] + for head in _heads(): + try: + out.append(train(log=log, head=head)) + except Exception as e: + log(f"[core] {head}: тренировка упала ({e})") + out.append({"head": head, "error": str(e)[:200]}) + return out + + +def rollback(head: str = HEAD) -> dict: + meta = _champion_meta(head) hist = meta.get("history") or [] if not hist: return {"ok": False, "reason": "истории нет"} prev = hist[0] new = {**prev, "history": hist[1:], "ts": datetime.now(timezone.utc).isoformat(), "rolled_back_from": meta.get("file")} - tmp = _model_path(".champion.tmp") + tmp = _model_path(".champion.tmp", head) with open(tmp, "w") as f: json.dump(new, f, ensure_ascii=False, indent=1) - os.replace(tmp, _model_path("champion.json")) - _STATE.update(model=None, loaded_sha=None) - es.append_event("core", f"rollback:{HEAD}:{int(time.time())}", "CoreModelPromoted", { - "head": HEAD, "file": prev.get("file"), "sha256": prev.get("sha256"), + os.replace(tmp, _model_path("champion.json", head)) + _forget(head) + es.append_event("core", f"rollback:{head}:{int(time.time())}", "CoreModelPromoted", { + "head": head, "file": prev.get("file"), "sha256": prev.get("sha256"), "rollback_from": meta.get("file")}) return {"ok": True, "file": prev.get("file")} @@ -641,25 +671,274 @@ def backfill(log=print, limit: int = 2000) -> int: return n +# ────────────────────────── голова task_fit ────────────────────────── +# Влезет ли S-задача в мин-тик с первой попытки (бюджет 30 шагов) или ей +# нужен час. Метки — сами из истории задач: 1 — закрыта на min с одного +# захвата; 0 — повторные захваты на min / закрыта другим уровнем / эскалация. +# На истории: 98 S-задач, 45 влезли, 53 нет. В on — уверенное «не влезет» +# сразу S→M (минуя пустые мин-попытки); правило эскалации — страховка. + +TASK_HEAD = "task_fit" +TASK_FLAGS = ("details_len", "code_paths", "external_systems", "est_minutes", "await_delivery") +_CODE_RE = re.compile(r"/root|/var/|/opt/|\.py\b|\.js\b|\.html\b|app\.py|static/", re.IGNORECASE) +_EXT_RE = re.compile(r"портал|cil\.by|сервер|ssh|деплой|deploy|миграц|панел|panel|кабинет|" + r"фронт|ui\b|дашборд|dashboard|репо|git", re.IGNORECASE) + + +def task_history() -> dict: + """{task_id: {title, details, size, est, await, claims:[level], done_level, ts}}.""" + T = {} + for aid, et, ts, d in _q("SELECT aggregate_id, event_type, timestamp, data FROM events " + "WHERE aggregate='task' ORDER BY id"): + d = json.loads(d) + if et == "TaskSubmitted": + T[aid] = {"id": aid, "title": d.get("title") or "", "details": d.get("details") or "", + "size": d.get("size"), "est": d.get("est_minutes"), "await": False, + "claims": [], "done_level": None, "ts": ts, "escalated": False} + elif aid in T: + t = T[aid] + if et == "TaskAwaitDelivery": + t["await"] = True + elif et == "TaskClaimed": + t["claims"].append(d.get("level")) + elif et == "TaskCompleted": + t["done_level"] = t["claims"][-1] if t["claims"] else None + elif et == "TaskPromoted" and d.get("from") != d.get("to"): + t["escalated"] = True + return T + + +def task_label(t: dict): + """1/0 — исход S-задачи на мин-уровне; None — исход ещё не ясен.""" + if t.get("size") != "S" or not t["claims"] or t["claims"][0] != "min": + return None + if t["done_level"] == "min" and t["claims"].count("min") == 1: + return 1.0 + if t["claims"].count("min") >= 2 or t["escalated"] or (t["done_level"] and t["done_level"] != "min"): + return 0.0 + return None + + +def task_dataset(): + rows = [] + for t in sorted(task_history().values(), key=lambda t: t["ts"]): + y = task_label(t) + if y is not None: + rows.append((t, y, 1.0, "outcome")) + return rows + + +def task_flags(t: dict) -> list: + text = f"{t.get('title') or ''}\n{t.get('details') or ''}" + est = t.get("est") + return [min(len(t.get("details") or "") / 1000.0, 1.0), + 1.0 if _CODE_RE.search(text) else 0.0, + 1.0 if _EXT_RE.search(text) else 0.0, + min(float(est) / 60.0, 1.0) if isinstance(est, (int, float)) and est > 0 else 0.0, + 1.0 if t.get("await") else 0.0] + + +def task_featurize(items: list): + texts = [f"задача: {t.get('title') or ''} — {(t.get('details') or '')[:600]}" for t in items] + vecs = embed(texts) + n = len(items) + em = np.stack(vecs) if n else np.zeros((0, EMB_DIM), np.float32) + fl = np.asarray([task_flags(t) for t in items], dtype=np.float32).reshape(n, len(TASK_FLAGS)) + return np.hstack([em, fl]), em + + +def task_shadow(task: dict, level: str): + """Прогноз «влезет в мин-тик» для S-задачи перед первым мин-захватом. + + Возвращает {"p_fit", "route", ...} или None. В on уверенное «не влезет» + → вызывающий поднимает S→M до клейма. CorePrediction — один раз на задачу. + """ + m = mode(TASK_HEAD) + if m == "off" or level != "min" or task.get("size") != "S": + return None + try: + model = load_champion(TASK_HEAD) + if model is None: + return None + t = {"title": task.get("title"), "details": task.get("details"), + "est": task.get("est_minutes"), "await": task.get("await_delivery")} + X, em = task_featurize([t]) + p = float(predict(model, X)[0]) + nov = float(novelty(model, em)[0]) + except Exception as e: + return {"error": str(e)[:200]} + fam = float(model["nov_q"][0]) + r = "student" if (nov < fam and abs(p - .5) >= CONF) else "rule" + rec = {"head": TASK_HEAD, "task": task.get("id"), "p_fit": round(p, 4), + "novelty": round(nov, 4), "route": r, "mode": m, + "act": bool(m == "on" and r == "student" and p < .5)} + _once(f"pred:{TASK_HEAD}:{task.get('id')}", "CorePrediction", rec) + return rec + + +# ────────────────────────── рефлексия и отчёт ────────────────────────── +# Как reflexion.py бенча: системная ошибка → не молча копить, а принести +# человеку правило. Без новых LLM-путей: досье уходит задачей часовому +# уровню (с автодоставкой итога в чат) — он разбирает случаи и предлагает +# Alex правило/навык. Порог — SIMBIOT_CORE_REFLECT_MIN (3) случаев с +# прошлой рефлексии этой темы. + +def _last(etype: str, key: str): + """Время прошлой рефлексии темы; первой — окно 7 дней (не тащить всю историю, + уже исправленную, в досье).""" + row = _q("SELECT MAX(timestamp) FROM events WHERE event_type=? " + "AND json_extract(data,'$.topic')=?", (etype, key)) + return row[0][0] or (datetime.now(timezone.utc) - timedelta(days=7)).isoformat() + + +def _reflect_cases_addressing(since: str) -> list: + """Где учитель/правило ошиблись: метка человека или чата против вердикта.""" + out = [] + for chat, msgs in group_timeline(since).items(): + for m in msgs: + if m.get("addressed") is None: + continue + labs = [l for l in m.get("labels", []) if l["source"] in ("human", "feedback")] + if labs and (labs[-1]["y"] >= .5) != bool(m["addressed"]): + out.append({"msg": m["id"], "who": m.get("first_name"), + "text": (m.get("text") or "")[:200], "rule": m.get("rule"), + "verdict": "ответил" if m["addressed"] else "промолчал", + "truth": "было боту" if labs[-1]["y"] >= .5 else "было не боту"}) + return out + + +def _reflect_cases_tasks(since: str) -> list: + """S-задачи, поданные после прошлой рефлексии, которые не влезли в мин-тик.""" + done = _reflected_tasks() + return [{"task": t["id"], "title": t["title"][:120], "min_attempts": t["claims"].count("min")} + for t in task_history().values() + if t["ts"] > since and t["id"] not in done + and (t["escalated"] or t["claims"].count("min") >= 2)] + + +def _reflected_tasks() -> set: + got = set() + for (d,) in _q("SELECT data FROM events WHERE event_type='CoreReflection' " + "AND json_extract(data,'$.topic')='task_fit'"): + got.update(json.loads(d).get("cases") or []) + return got + + +def reflect(log=print) -> list: + """Досье системных ошибок → задача часу с автодоставкой. Возвращает id задач.""" + from . import commands + limit = int(os.environ.get("SIMBIOT_CORE_REFLECT_MIN", "3")) + made = [] + since = _last("CoreReflection", HEAD) + cases = _reflect_cases_addressing(since) + if len(cases) >= limit: + lines = [f"- {c['who']}: «{c['text']}» — я {c['verdict']} (правило {c['rule']}), " + f"а {c['truth']}" for c in cases[:20]] + tid, _ = commands.submit_task( + "Рефлексия адресации: где я ошибся в группе — предложи правило", + "Ядро Коди собрало случаи, где решение «отвечать/молчать» разошлось с " + "твоей отметкой или реакцией чата:\n" + "\n".join(lines) + + "\n\nРазбери, что их объединяет, и предложи Alex 1–3 конкретных правила " + "(текстом в чат: формулировка + пример). Ничего в коде не меняй — правила " + "внедряются после одобрения.", minutes=30, source="core:reflect") + es.append_event("task", tid, "TaskAwaitDelivery", {"reply_to": None, "reason": "рефлексия ядра"}) + es.append_event("core", f"reflect:{HEAD}:{tid}", "CoreReflection", + {"topic": HEAD, "task": tid, "cases": [c["msg"] for c in cases]}) + made.append(tid) + tcases = _reflect_cases_tasks(_last("CoreReflection", TASK_HEAD)) + if len(tcases) >= limit: + lines = [f"- {c['title']} ({c['min_attempts']} попыток на мин-уровне)" for c in tcases[:20]] + tid, _ = commands.submit_task( + "Рефлексия задач: какие S-задачи не влезают в минутный тик — предложи правило", + "Эти задачи были поданы как S, но не уложились в мин-тик (30 шагов) и " + "ушли выше:\n" + "\n".join(lines) + + "\n\nНайди общее (тип работы, нужный доступ, чужой код…) и предложи Alex " + "правило постановки/оценки размера (текстом в чат). Код не меняй.", + minutes=30, source="core:reflect") + es.append_event("task", tid, "TaskAwaitDelivery", {"reply_to": None, "reason": "рефлексия ядра"}) + es.append_event("core", f"reflect:{TASK_HEAD}:{tid}", "CoreReflection", + {"topic": TASK_HEAD, "task": tid, "cases": [c["task"] for c in tcases]}) + made.append(tid) + if made: + log(f"[core] рефлексия: задачи {made}") + return made + + +def weekly_report(force: bool = False) -> "str | None": + """Недельный отчёт ядра Alex — детерминированный текст, без LLM. + Раз в ISO-неделю (идемпотентно по ключу доставки), в понедельник.""" + now = datetime.now(timezone.utc) + if not force and now.weekday() != 0: + return None + st = status() + a, t = st[HEAD], st[TASK_HEAD] + week = now.strftime("%G-W%V") + since = (now - timedelta(days=7)).isoformat() + prom = [json.loads(d) for (d,) in _q("SELECT data FROM events WHERE event_type=" + "'CoreModelPromoted' AND timestamp > ?", (since,))] + refl = _q("SELECT COUNT(*) FROM events WHERE event_type='CoreReflection' AND timestamp > ?", + (since,))[0][0] + fb = _q("SELECT COUNT(*) FROM events WHERE event_type='CoreFeedback' AND timestamp > ?", + (since,))[0][0] + + def pct(v): + return "—" if v is None else f"{v * 100:.0f}%" + am, tm = a.get("champion_metrics") or {}, t.get("champion_metrics") or {} + text = "\n".join([ + f"🧠 Ядро Коди — неделя {week}", + "", + f"Адресация в группах ({a['mode']}): ученик на {a.get('champion_n') or 0} примерах, " + f"совпадение с учителем на свежих {pct(am.get('acc'))}; за неделю прогнозов " + f"{a['predictions_7d']}, решил бы сам {pct(a.get('would_be_local'))}, " + f"согласие с учителем {pct(a.get('agree_with_teacher_7d'))}.", + f"Метки: от тебя {a['labels'].get('human', 0)}, от реакции чата за неделю {fb}; " + f"вопросов тебе {a['questions_7d']}.", + f"Задачи ({t['mode']}): влезет ли S в мин-тик — точность на свежих " + f"{pct(tm.get('acc'))} при базовой {pct(tm.get('base_rate'))}; сверено с исходом " + f"{t.get('judged_7d', 0)}, верно {pct(t.get('outcome_acc_7d'))}.", + f"Смен чемпиона за неделю: {len(prom)}; рефлексий (правил на разбор): {refl}.", + "", + "Включить ученика вместо учителя: SIMBIOT_CORE_ADDRESSING=on / SIMBIOT_CORE_TASK_FIT=on.", + ]) + from . import commands, tg_inbox + ok = commands.deliver_artifact(f"msg:core-report:{week}", kind="message", + owner="core:weekly", caption=text, + send=lambda: bool(tg_inbox.notify_chat(text))) + return text if ok else None + + def status() -> dict: - meta = _champion_meta() day = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat() preds = [json.loads(d) for (d,) in _q( "SELECT data FROM events WHERE event_type='CorePrediction' AND timestamp > ?", (day,))] - agree = [p["agree"] for p in preds if p.get("agree") is not None] + out = {} + for head in _heads(): + meta = _champion_meta(head) + hp = [p for p in preds if p.get("head", HEAD) == head] + out[head] = {"mode": mode(head), "champion": meta.get("file"), + "champion_metrics": meta.get("metrics"), "champion_n": meta.get("n"), + "champion_ts": meta.get("ts"), + "predictions_7d": len(hp), + "would_be_local": (round(sum(p["route"] == "student" for p in hp) / len(hp), 3) + if hp else None)} + a = out[HEAD] + ap = [p for p in preds if p.get("head", HEAD) == HEAD] + agree = [p["agree"] for p in ap if p.get("agree") is not None] srcs = {} for (d,) in _q("SELECT data FROM events WHERE event_type IN ('CoreLabel','CoreFeedback')"): - s = json.loads(d).get("source") or "feedback" - srcs[s] = srcs.get(s, 0) + 1 - return { - "head": HEAD, "mode": mode(), "champion": meta.get("file"), - "champion_metrics": meta.get("metrics"), "champion_n": meta.get("n"), - "champion_ts": meta.get("ts"), "labels": srcs, - "shadow_7d": {"predictions": len(preds), - "agree_with_teacher": (round(sum(agree) / len(agree), 3) if agree else None), - "would_be_local": (round(sum(p["route"] == "student" for p in preds) - / len(preds), 3) if preds else None), - "would_ask_human": sum(p["route"] == "human" for p in preds)}, - "questions_7d": _q("SELECT COUNT(*) FROM events WHERE event_type='CoreQuestionAsked' " - "AND timestamp > ?", (day,))[0][0], - } + s_ = json.loads(d).get("source") or "feedback" + srcs[s_] = srcs.get(s_, 0) + 1 + a.update(labels=srcs, + agree_with_teacher_7d=(round(sum(agree) / len(agree), 3) if agree else None), + would_ask_human_7d=sum(p["route"] == "human" for p in ap), + questions_7d=_q("SELECT COUNT(*) FROM events WHERE event_type='CoreQuestionAsked' " + "AND timestamp > ?", (day,))[0][0]) + # task_fit: сверка теневых прогнозов с фактическим исходом + hist = task_history() + tp = [p for p in preds if p.get("head") == TASK_HEAD] + judged = [(p, task_label(hist[p["task"]])) for p in tp if p.get("task") in hist] + judged = [(p, y) for p, y in judged if y is not None] + out[TASK_HEAD]["outcome_acc_7d"] = (round(sum((p["p_fit"] >= .5) == (y >= .5) for p, y in judged) + / len(judged), 3) if judged else None) + out[TASK_HEAD]["judged_7d"] = len(judged) + return out diff --git a/harness/scheduler.py b/harness/scheduler.py index aca0a1c..08c9444 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -531,7 +531,79 @@ def _notify_busy(level_cfg, task): ) -def _guard_budget_stub_requeue(claimed, text: str): +# Эскалация недоделанных (аудит 2026-10-17): «бюджет исчерпан»/смерть воркера +# возвращали задачу с тем же размером — мин-уровень брал её снова и снова +# с нуля (рекорд: S-задача 33 захвата за 2 ч; 45 из 272 задач ≥ 2 захватов +# на min; пустые мин-тики по 0.45M токенов — те же grep по кругу). После +# SIMBIOT_TASK_ATTEMPTS_PER_LEVEL (2) захватов на уровне задача растёт до +# размера, которого этот уровень не берёт (S→M: к часу, бюджет 300 шагов). +_SIZES = ("S", "M", "L") + + +def _attempts_at(task_id: str, level: str) -> int: + """Число захватов задачи уровнем. Best effort: сбой счётчика не должен + ломать возврат задачи в очередь (он идёт и из обработки падения воркера) + — тогда 0, т.е. обычный requeue, как до эскалации.""" + try: + db = sqlite3.connect(es.DB_PATH, timeout=30) + try: + return db.execute( + "SELECT COUNT(*) FROM events WHERE aggregate='task' AND aggregate_id=? " + "AND event_type='TaskClaimed' AND json_extract(data,'$.level')=?", + (task_id, level)).fetchone()[0] + finally: + db.close() + except Exception: + return 0 + + +def _core_task_fit(batch: list) -> list: + """Ядро Коди, голова task_fit: S-задача перед ПЕРВЫМ мин-захватом — + прогноз «влезет ли в мин-тик». shadow — только CorePrediction; on — + уверенное «не влезет» сразу S→M (к часу), минуя пустые мин-попытки.""" + try: + from . import kodi_core + except Exception: + return batch + keep = [] + for t in batch: + rec = None + if t.get("size") == "S" and _attempts_at(t["id"], "min") == 0: + try: + rec = kodi_core.task_shadow(t, "min") + except Exception: + rec = None + if rec and rec.get("act"): + es.append_event("task", t["id"], "TaskPromoted", { + "to": "M", "from": "S", + "reason": f"ядро Коди: не влезет в мин-тик (p_fit={rec['p_fit']}) — сразу к часу"}) + _log(f"core: {t['id']} S→M до мин-попытки (p_fit={rec['p_fit']})") + continue + keep.append(t) + return keep + + +def requeue_or_escalate(task_id: str, level_cfg, size: str, reason: str) -> str: + """Вернуть недоделанную задачу в очередь; после N попыток на уровне — + поднять размер выше классов уровня. Возвращает новый размер.""" + try: + limit = int(os.environ.get("SIMBIOT_TASK_ATTEMPTS_PER_LEVEL", "2")) + except ValueError: + limit = 2 + new = size + if size in _SIZES and _attempts_at(task_id, level_cfg.name) >= limit: + bigger = [z for z in _SIZES[_SIZES.index(size) + 1:] if z not in level_cfg.task_classes] + if bigger: + new = bigger[0] + reason = (f"эскалация: {limit}+ попыток на уровне {level_cfg.name} не хватило " + f"бюджета — {size}→{new}, возьмёт уровень выше; " + reason) + es.append_event("task", task_id, "TaskPromoted", {"to": new, "from": size, "reason": reason}) + if new != size: + _log(f"task {task_id}: {size}→{new} после попыток на {level_cfg.name}") + return new + + +def _guard_budget_stub_requeue(claimed, text: str, level_cfg=None, batch=None): """GUARD (e-261001074527-f0c1b4): финальный текст тика — отписка воркера «бюджет исчерпан» (стоп-сообщение, не результат) → задачи пачки без TaskCompleted возвращаются в ready. @@ -549,6 +621,13 @@ def _guard_budget_stub_requeue(claimed, text: str): reason = ("guard: финал тика «бюджет исчерпан» — работа не завершена, " "задача в ready (e-261001074527)") for tid in (claimed or []): + if level_cfg is not None: + if not _task_still_active(tid): + _log(f"guard: requeue {tid} пропущен — уже не running") + continue + size = next((t.get("size", "S") for t in (batch or []) if t.get("id") == tid), "S") + requeue_or_escalate(tid, level_cfg, size, reason) + continue try: _requeue(tid, reason=reason) except ValueError as e: @@ -591,6 +670,8 @@ def run_level_tick(level_cfg, seq: int, model: dict): 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] + if level_cfg.name == "min": + batch = _core_task_fit(batch) claimed = [] stale = [] # заняты другим уровнем после снапшота модели (гонка клейма) for t in batch: @@ -679,10 +760,9 @@ def run_level_tick(level_cfg, seq: int, model: dict): if not _task_still_active(tid): continue size = next((t.get("size", "M") for t in batch if t.get("id") == tid), "M") - es.append_event("task", tid, "TaskPromoted", - {"to": size, "from": size, - "reason": f"тик-воркер умер ({str(e)[:120]}) — " - f"claim возвращён в очередь без потери"}) + requeue_or_escalate(tid, level_cfg, size, + f"тик-воркер умер ({str(e)[:120]}) — " + f"claim возвращён в очередь без потери") raise if text is None: # воркер упал (таймаут/краш) — НИ одна задача не выполнена с его @@ -725,7 +805,7 @@ def run_level_tick(level_cfg, seq: int, model: dict): # 5 L-задач. Такой тик работы не завершал — задачи возвращаются в ready. from .commands import looks_like_budget_stub if (text or "") and looks_like_budget_stub(text): - return _guard_budget_stub_requeue(claimed, text) + return _guard_budget_stub_requeue(claimed, text, level_cfg, batch) # guard двойного закрытия (r2 e2e cron-migrate 2026-10-03): воркер мог # ЗАКРЫТЬ задачу сам (cli complete) ещё до батч-фазы — его TaskCompleted # содержит целевой ответ; повторный append поверх перезаписал бы answer @@ -912,7 +992,9 @@ def process_level(level_cfg, model: dict, seq: int = None): # ядро Коди: ночная тренировка учеников (чемпион/претендент) try: from . import kodi_core - kodi_core.train(log=_log) + kodi_core.train_all(log=_log) + kodi_core.reflect(log=_log) + kodi_core.weekly_report() except Exception as e: _log(f"core: ночная тренировка не удалась ({e})") try: diff --git a/tests/test_kodi_core.py b/tests/test_kodi_core.py index 56908fe..36d64e8 100644 --- a/tests/test_kodi_core.py +++ b/tests/test_kodi_core.py @@ -96,7 +96,7 @@ check("метка человека сильнее учителя (y=1, вес 5) # ── 2. обучение и промоут ── rep = kodi_core.train(log=lambda *_: None) check("первое обучение → промоут", rep["promoted"] and rep["holdout"]["acc"] >= 0.9, rep) -st = kodi_core.status() +st = kodi_core.status()["addressing"] check("чемпион записан", st["champion"] and st["champion_n"] == 61, st) prom = [json.loads(e["data"]) for e in es.get_events(0, 10**6) if e["event_type"] == "CoreModelPromoted"] check("CoreModelPromoted с sha256", prom and len(prom[-1]["sha256"]) == 64) @@ -118,10 +118,10 @@ meta = kodi_core._champion_meta() path = kodi_core._model_path(meta["file"]) blob = open(path, "rb").read() open(path, "wb").write(blob[:-1] + bytes([blob[-1] ^ 1])) -kodi_core._STATE.update(model=None, loaded_sha=None) +kodi_core._forget("addressing") check("файл модели подменён → ученика нет", kodi_core.load_champion() is None) open(path, "wb").write(blob) -kodi_core._STATE.update(model=None, loaded_sha=None) +kodi_core._forget("addressing") # ── 3. роутер ── check("роутер: знакомо+уверенно → ученик", kodi_core.route(0.95, 0.1, 0.9) == "student") diff --git a/tests/test_kodi_core_reflect.py b/tests/test_kodi_core_reflect.py new file mode 100644 index 0000000..a9648a6 --- /dev/null +++ b/tests/test_kodi_core_reflect.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- +"""Ядро Коди: рефлексия и недельный отчёт — hermetic. + +Запуск: ./venv/bin/python tests/test_kodi_core_reflect.py + +Проверяет: < порога ошибок — рефлексии нет; ≥ 3 расхождений вердикта с +меткой человека/чата → задача часу (M, автодоставка, случаи в постановке) ++ CoreReflection; повтор — только по новым случаям; задачи: ≥ 3 S-задач +не влезли в мин → своя рефлексия, разобранные не повторяются; недельный +отчёт: не в понедельник — нет, force — текст с метриками, повтор недели +— SKIP (идемпотентная доставка). +""" +import json +import os +import sys +import tempfile + +sys.path.insert(0, "/root/simbiot-harness") +_tmp = tempfile.mkdtemp(prefix="simbiot_core_refl_") +os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db") +os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core") +os.environ.pop("SIMBIOT_CORE_REFLECT_MIN", None) + +from harness import es, kodi_core, tg_inbox + +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) + + +SENT = [] +tg_inbox.notify_chat = lambda text: SENT.append(text) or {"ok": True} +G = -1004402779881 +N = [0] + + +def wrong(verdict_addressed: bool, truth: int, src="feedback"): + N[0] += 1 + mid = f"m-{N[0]}" + es.append_event("inbox", mid, "MessageReceived", {"update_id": N[0], "chat_id": G, + "chat_type": "supergroup", "first_name": "Alex", "text": f"реплика {N[0]}"}) + es.append_event("inbox", mid, "MessageTriaged" if verdict_addressed else "MessageIgnored", + {"addressed": verdict_addressed, "reason": "not_addressed", "rule": "decision", "p": .6}) + es.append_event("core", f"lab:{mid}", "CoreLabel" if src == "human" else "CoreFeedback", + {"head": "addressing", "msg": mid, "label": truth, "source": src}) + + +def tasks_with(event_type): + return [json.loads(e["data"]) for e in es.get_events(0, 10**6) if e["event_type"] == event_type] + + +wrong(True, 0); wrong(False, 1) +check("2 ошибки < порога — рефлексии нет", kodi_core.reflect(log=lambda *_: None) == []) +wrong(True, 0, "human") +made = kodi_core.reflect(log=lambda *_: None) +sub = {t["id"]: t for t in tasks_with("TaskSubmitted")} +check("3 ошибки → задача рефлексии", len(made) == 1 and "Рефлексия адресации" in sub[made[0]]["title"], made) +check("задача — к часу (M) с автодоставкой", + sub[made[0]]["size"] == "M" and es.run_projections()["tasks"][made[0]].get("await_delivery")) +check("случаи в постановке", "я ответил" in sub[made[0]]["details"] and "было не боту" in sub[made[0]]["details"]) +check("повтор без новых случаев — нет", kodi_core.reflect(log=lambda *_: None) == []) + +# задачи: 3 S-задачи не влезли +for i in range(3): + tid = f"t-{i}" + es.append_event("task", tid, "TaskSubmitted", {"id": tid, "title": f"найди плашку {i}", "size": "S"}) + es.append_event("task", tid, "TaskClaimed", {"level": "min"}) + es.append_event("task", tid, "TaskClaimed", {"level": "min"}) +made2 = kodi_core.reflect(log=lambda *_: None) +sub = {t["id"]: t for t in tasks_with("TaskSubmitted")} +check("задачи: 3 не влезли → рефлексия задач", len(made2) == 1 and "Рефлексия задач" in sub[made2[0]]["title"], made2) +check("задачи: разобранные не повторяются", kodi_core.reflect(log=lambda *_: None) == []) + +# недельный отчёт +from datetime import datetime, timezone +if datetime.now(timezone.utc).weekday() != 0: + check("не понедельник — отчёта нет", kodi_core.weekly_report() is None and not SENT) +txt = kodi_core.weekly_report(force=True) +check("отчёт: текст с головами и метриками", txt and "Адресация в группах" in txt and "Задачи" in txt + and "рефлексий" in txt and SENT, txt) +check("отчёт: повтор недели — SKIP", kodi_core.weekly_report(force=True) is None and len(SENT) == 1) + +print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)") +sys.exit(1 if fails else 0) diff --git a/tests/test_kodi_core_tasks.py b/tests/test_kodi_core_tasks.py new file mode 100644 index 0000000..9eebab5 --- /dev/null +++ b/tests/test_kodi_core_tasks.py @@ -0,0 +1,129 @@ +# -*- coding: utf-8 -*- +"""Ядро Коди, голова task_fit: «влезет ли S-задача в мин-тик» — hermetic. + +Запуск: ./venv/bin/python tests/test_kodi_core_tasks.py + +Проверяет: метки из истории (1 — закрыта на min с одного захвата; 0 — +повторы/эскалация/другой уровень; None — исход не ясен, M-задачи); +обучение и чемпион головы; shadow — прогноз пишется, батч не меняется; +on — уверенное «не влезет» → S→M до мин-захвата, «влезет» остаётся; +повторно (после мин-попытки) прогноз не делается; статус по головам. +""" +import hashlib +import json +import os +import sys +import tempfile + +sys.path.insert(0, "/root/simbiot-harness") +_tmp = tempfile.mkdtemp(prefix="simbiot_core_tasks_") +os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db") +os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core") +for _v in ("SIMBIOT_CORE_TASK_FIT", "SIMBIOT_DECISION"): + os.environ.pop(_v, None) + +import numpy as np + +from harness import es, kodi_core, scheduler + +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) + + +A = np.zeros(768, np.float32); A[0] = 1 +B = np.zeros(768, np.float32); B[1] = 1 + + +def fake_embed(texts): + out = [] + for t in texts: + rng = np.random.default_rng(int(hashlib.sha1(t.encode()).hexdigest(), 16) % (2 ** 32)) + v = (A if "тяжёлая" in t else B) + 0.02 * rng.standard_normal(768).astype(np.float32) + out.append(v / np.linalg.norm(v)) + return out + + +kodi_core.embed = fake_embed +N = [0] + + +def task(title, size="S", claims=(), done=None, promote=None): + N[0] += 1 + tid = f"t-{N[0]:04d}" + es.append_event("task", tid, "TaskSubmitted", {"id": tid, "title": title, "size": size, + "created": f"2026-10-01T00:{N[0] % 60:02d}:00"}) + for lv in claims: + es.append_event("task", tid, "TaskClaimed", {"level": lv}) + if promote and lv == "min": + es.append_event("task", tid, "TaskPromoted", {"from": "S", "to": "S", "reason": "guard"}) + if promote == "esc": + es.append_event("task", tid, "TaskPromoted", {"from": "S", "to": "M", "reason": "эскалация"}) + if done: + es.append_event("task", tid, "TaskCompleted", {"result": "ok"}) + return tid + + +# ── метки ── +fit = task("лёгкая: проверь бэкап", claims=["min"], done=True) +loop = task("тяжёлая: найди плашку в портале", claims=["min", "min", "min"], done=True, promote="guard") +esc = task("тяжёлая: миграция", claims=["min", "min", "hour"], done=True, promote="esc") +open_ = task("лёгкая: в работе", claims=["min"]) +m_ = task("M-задача", size="M", claims=["hour"], done=True) +H = kodi_core.task_history() +check("метка 1: закрыта на min с одного захвата", kodi_core.task_label(H[fit]) == 1.0) +check("метка 0: три захвата на min", kodi_core.task_label(H[loop]) == 0.0) +check("метка 0: эскалация к часу", kodi_core.task_label(H[esc]) == 0.0) +check("исход не ясен — без метки", kodi_core.task_label(H[open_]) is None) +check("M-задача — не для этой головы", kodi_core.task_label(H[m_]) is None) + +# ── обучение ── +for i in range(30): + task(f"лёгкая: мелочь {i}", claims=["min"], done=True) + task(f"тяжёлая: исследование {i}", claims=["min", "min"], done=True, promote="guard") +rep = kodi_core.train(log=lambda *_: None, head="task_fit") +check("обучение task_fit → чемпион", rep["promoted"] and rep["holdout"]["acc"] >= .9, rep) +check("модели голов — в разных каталогах", + os.path.isfile(os.path.join(os.environ["SIMBIOT_CORE_DIR"], "task_fit", "champion.json"))) + +# ── shadow: прогноз есть, батч не трогаем ── +heavy = task("тяжёлая: новая про портал") +light = task("лёгкая: новая проверка") +model = es.run_projections() +batch = [model["tasks"][heavy], model["tasks"][light]] +out = scheduler._core_task_fit(list(batch)) +check("shadow: батч не изменён", [t["id"] for t in out] == [heavy, light]) +preds = {json.loads(e["data"])["task"]: json.loads(e["data"]) for e in es.get_events(0, 10**6) + if e["event_type"] == "CorePrediction"} +check("shadow: прогнозы записаны (тяжёлая < 0.5 < лёгкая)", + preds[heavy]["p_fit"] < .5 < preds[light]["p_fit"] and not preds[heavy]["act"], preds) + +# ── on: уверенное «не влезет» → S→M до клейма ── +os.environ["SIMBIOT_CORE_TASK_FIT"] = "on" +heavy2 = task("тяжёлая: ещё одна") +light2 = task("лёгкая: ещё одна") +model = es.run_projections() +fam = kodi_core.load_champion("task_fit")["nov_q"] +kodi_core.load_champion("task_fit")["nov_q"] = np.asarray([1.0, 1.0]) # всё «знакомо» — проверяем связку +out = scheduler._core_task_fit([model["tasks"][heavy2], model["tasks"][light2]]) +t2 = es.run_projections()["tasks"] +check("on: тяжёлая убрана из мин-батча и стала M", [t["id"] for t in out] == [light2] + and t2[heavy2]["size"] == "M", ([t["id"] for t in out], t2[heavy2]["size"])) +check("on: лёгкая осталась S", t2[light2]["size"] == "S") +es.append_event("task", light2, "TaskClaimed", {"level": "min"}) +n_before = sum(1 for e in es.get_events(0, 10**6) if e["event_type"] == "CorePrediction") +scheduler._core_task_fit([es.run_projections()["tasks"][light2]]) +check("после мин-попытки прогноз не повторяется", + sum(1 for e in es.get_events(0, 10**6) if e["event_type"] == "CorePrediction") == n_before) +os.environ.pop("SIMBIOT_CORE_TASK_FIT") + +st = kodi_core.status() +check("статус: обе головы", set(st) == {"addressing", "task_fit"} and st["task_fit"]["champion"], st.keys()) + +print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)") +sys.exit(1 if fails else 0) diff --git a/tests/test_task_escalation.py b/tests/test_task_escalation.py new file mode 100644 index 0000000..8761bb8 --- /dev/null +++ b/tests/test_task_escalation.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- +"""Эскалация недоделанных задач по уровням (аудит 2026-10-17). Hermetic. + +Запуск: ./venv/bin/python tests/test_task_escalation.py + +Кейс: «бюджет исчерпан» возвращал S-задачу как S — мин-уровень брал её +по кругу с нуля (33 захвата одной задачи за 2 ч). Проверяет: 1-я неудача +на min — остаётся S; 2-я — S→M, мин её больше не берёт, час берёт; +guard «бюджет исчерпан» идёт тем же путём; закрытую внутри тика задачу +guard не трогает; на часу (берёт S/M/L) размер не растёт — расти некуда; +порог настраивается env. +""" +import os +import sys +import tempfile + +sys.path.insert(0, "/root/simbiot-harness") +os.environ["SIMBIOT_DB"] = os.path.join(tempfile.mkdtemp(prefix="simbiot_esc_"), "harness.db") +os.environ.pop("SIMBIOT_TASK_ATTEMPTS_PER_LEVEL", None) + +from harness import es, scheduler +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) + + +def submit(tid, size="S"): + es.append_event("task", tid, "TaskSubmitted", {"id": tid, "title": f"задача {tid}", + "size": size, "created": "2026-10-17T08:00:00"}) + + +def claim(tid, lvl="min"): + assert es.claim_task_if_ready(tid, {"level": lvl, "session_id": f"s-{lvl}-1"}) + + +def task(tid): + return es.run_projections()["tasks"][tid] + + +MIN, HOUR = LEVELS["min"], LEVELS["hour"] +submit("t-1") +claim("t-1") +scheduler.requeue_or_escalate("t-1", MIN, "S", "бюджет исчерпан") +check("1-я неудача на min — остаётся S, ready", task("t-1")["size"] == "S" + and task("t-1")["status"] == "ready") +claim("t-1") +scheduler.requeue_or_escalate("t-1", MIN, "S", "бюджет исчерпан") +t = task("t-1") +check("2-я неудача на min — S→M", t["size"] == "M" and t["status"] == "ready", t) +check("мин больше не берёт", "t-1" not in scheduler.select_task_batch(MIN, es.run_projections())) +check("час берёт", "t-1" in scheduler.select_task_batch(HOUR, es.run_projections())) + +# guard «бюджет исчерпан» с уровнем — тот же путь +submit("t-2") +for _ in range(2): + claim("t-2") + scheduler._guard_budget_stub_requeue(["t-2"], "бюджет исчерпан", MIN, + [{"id": "t-2", "size": task("t-2")["size"]}]) +check("guard: после 2 попыток S→M", task("t-2")["size"] == "M", task("t-2")) + +# закрытую внутри тика guard не воскрешает +submit("t-3") +claim("t-3") +es.append_event("task", "t-3", "TaskCompleted", {"result": "сделано"}) +scheduler._guard_budget_stub_requeue(["t-3"], "бюджет исчерпан", MIN, [{"id": "t-3", "size": "S"}]) +check("guard: закрытая задача не возвращается", task("t-3")["status"] == "done", task("t-3")["status"]) + +# на часу размер не растёт: час берёт S/M/L +submit("t-4", "M") +for _ in range(3): + claim("t-4", "hour") + scheduler.requeue_or_escalate("t-4", HOUR, "M", "бюджет исчерпан") +check("час: размер не растёт (выше брать некому)", task("t-4")["size"] == "M") + +os.environ["SIMBIOT_TASK_ATTEMPTS_PER_LEVEL"] = "1" +submit("t-5") +claim("t-5") +scheduler.requeue_or_escalate("t-5", MIN, "S", "x") +check("порог из env (1 попытка)", task("t-5")["size"] == "M") + +print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)") +sys.exit(1 if fails else 0)