task_fit: rule-ветка «внешние системы → не влезет в мин-тик» (рефлексия t-261017093718)

93 размеченных S-задачи: ext>=1 precision 0.74 / recall 0.52; ML-ученик
logloss 0.849 хуже базы 0.707 — чемпион не промоутится. Без чемпиона /
вне зоны уверенности ученика решает детерминированное правило: портал/
сервер/деплой/UI/git в тексте S-задачи → S→M до мин-клейма (route=rule:ext,
act только в SIMBIOT_CORE_TASK_FIT=on; дефолт shadow — прогноз пишется,
батч не трогается). Уверенный ученик (student) приоритетнее правила.

harness/kodi_core.py

@@ -747,6 +747,11 @@ def backfill(log=print, limit: int = 2000) -> int:
 # захвата; 0 — повторные захваты на min / закрыта другим уровнем / эскалация.
 # На истории: 98 S-задач, 45 влезли, 53 нет. В on — уверенное «не влезет»
 # сразу S→M (минуя пустые мин-попытки); правило эскалации — страховка.
+# Rule-ветка (рефлексия t-261017093718, 93 метки): ученик не уверен или
+# чемпиона нет → решает правило «внешние системы (портал/сервер/деплой/
+# UI/git) в тексте → не влезет» (precision 0.74 / recall 0.52; ML-ученик
+# logloss 0.849 хуже базы 0.707 — не промоутится). Режим дефолт shadow:
+# правило пишется в CorePrediction (route=rule:ext), включается SIMBIOT_CORE_TASK_FIT=on.
 
 TASK_HEAD = "task_fit"
 TASK_FLAGS = ("details_len", "code_paths", "external_systems", "est_minutes", "await_delivery")
@@ -822,14 +827,20 @@ def task_shadow(task: dict, level: str):
 
     Возвращает {"p_fit", "route", ...} или None. В on уверенное «не влезет»
     → вызывающий поднимает S→M до клейма. CorePrediction — один раз на задачу.
+
+    Rule-ветка (рефлексия t-261017093718): без чемпиона/вне зоны уверенности
+    ученика решает детерминированное правило «внешние системы → не влезет»
+    (портал/сервер/деплой/UI/git в тексте задачи; история S-задач:
+    precision 0.74, recall 0.52 — см. комментарий головы task_fit).
     """
     m = mode(TASK_HEAD)
     if m == "off" or level != "min" or task.get("size") != "S":
         return None
+    text = f"{task.get('title') or ''}\n{task.get('details') or ''}"
     try:
         model = load_champion(TASK_HEAD)
-        if model is None:
-            return None
+        p = nov = None
+        if model is not 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])
@@ -837,11 +848,20 @@ def task_shadow(task: dict, level: str):
             nov = float(novelty(model, em)[0])
     except Exception as e:
         return {"error": str(e)[:200]}
+    ext = len(_EXT_RE.findall(text))
+    if p is not None and nov is not None:
         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)}
+        if nov < fam and abs(p - .5) >= CONF:      # ученик уверен — он и решает
+            r, act = "student", bool(m == "on" and p < .5)
+        elif ext:                                   # серая зона + правило
+            p, r, act = 0.2, "rule:ext", bool(m == "on")
+        else:
+            r, act = "rule", False
+    else:                                           # чемпиона нет — только правило
+        r, act = ("rule:ext", bool(m == "on")) if ext else ("rule", False)
+    rec = {"head": TASK_HEAD, "task": task.get("id"), "p_fit": round(p, 4) if p is not None else None,
+           "novelty": round(nov, 4) if nov is not None else None, "route": r, "mode": m,
+           "ext_hits": ext, "act": act}
     _once(f"pred:{TASK_HEAD}:{task.get('id')}", "CorePrediction", rec)
     return rec
 

harness/scheduler.py

@@ -581,10 +581,13 @@ def _core_task_fit(batch: list) -> list:
             except Exception:
                 rec = None
         if rec and rec.get("act"):
+            why = (f"ядро Коди: не влезет в мин-тик, rule:ext (внешние системы, "
+                   f"ext_hits={rec.get('ext_hits')})") if rec.get("route") == "rule:ext" \
+                else f"ядро Коди: не влезет в мин-тик (p_fit={rec.get('p_fit')})"
             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']})")
+                "to": "M", "from": "S", "reason": why + " — сразу к часу"})
+            _log(f"core: {t['id']} S→M до мин-попытки ({rec.get('route')}, "
+                 f"p_fit={rec.get('p_fit')}, ext={rec.get('ext_hits')})")
             continue
         keep.append(t)
     return keep

tests/test_kodi_core_tasks.py

@@ -165,5 +165,47 @@ kodi_core._forget("task_fit")
 st = kodi_core.status()
 check("статус: обе головы", set(st) == {"addressing", "task_fit"} and st["task_fit"]["champion"], st.keys())
 
+# ── rule-ветка (рефлексия t-261017093718): внешние системы → не влезет ──
+# чемпиона нет (пустой core-dir, как снятый) → решает правило; route=rule:ext
+os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core-empty")
+kodi_core._forget("task_fit")
+ext_t = task("S: подключить Google OAuth к порталу cil.by")
+plain_t = task("S: лёгкая сводка расходов")
+model = es.run_projections()
+out = scheduler._core_task_fit([model["tasks"][ext_t], model["tasks"][plain_t]])
+t3 = es.run_projections()["tasks"]
+check("rule без чемпиона: shadow — батч не тронут", [t["id"] for t in out] == [ext_t, plain_t])
+pred = [json.loads(e["data"]) for e in es.get_events(0, 10**6)
+        if e["event_type"] == "CorePrediction" and json.loads(e["data"])["task"] == ext_t][-1]
+check("rule: ext-задача помечена rule:ext без p_fit",
+      pred["route"] == "rule:ext" and pred.get("ext_hits", 0) >= 2 and pred["p_fit"] is None
+      and not pred["act"], pred)
+os.environ["SIMBIOT_CORE_TASK_FIT"] = "on"
+ext_t2 = task("S: баг портала /project/1 — фича-реквесты показываются открытыми")
+model = es.run_projections()
+out = scheduler._core_task_fit([model["tasks"][ext_t2]])
+t4 = es.run_projections()["tasks"]
+check("rule on: ext-задача сразу S→M, из мин-батча убрана",
+      out == [] and t4[ext_t2]["size"] == "M", (out, t4[ext_t2]["size"]))
+plain_t2 = task("S: лёгкая новость без внешних систем")
+model = es.run_projections()
+out = scheduler._core_task_fit([model["tasks"][plain_t2]])
+t5 = es.run_projections()["tasks"]
+check("rule on: без внешних систем остаётся S в мин-батче",
+      [t["id"] for t in out] == [plain_t2] and t5[plain_t2]["size"] == "S")
+os.environ.pop("SIMBIOT_CORE_TASK_FIT")
+os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core")
+kodi_core._forget("task_fit")
+# при уверенном ученике (чемпион есть) rule не мешает его решению
+conf_ext = task("тяжёлая: уверенный ученик против правила")
+model = es.run_projections()
+out = scheduler._core_task_fit([model["tasks"][conf_ext]])
+pred_conf = [json.loads(e["data"]) for e in es.get_events(0, 10**6)
+             if e["event_type"] == "CorePrediction" and json.loads(e["data"])["task"] == conf_ext][-1]
+check("уверенный ученик: route=student, rule не перебивает",
+      pred_conf["route"] == "student" and pred_conf["p_fit"] is not None
+      and [t["id"] for t in out] == [conf_ext], pred_conf)
+
+
 print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)")
 sys.exit(1 if fails else 0)