память разговора: ответ на сообщение бота — с полным его текстом, свои рассылки в памяти тика

Кейс Alex 10:10: «Ок делай» реплаем на итог рефлексии. Telegram отдаёт в
reply начало, харнес хранил 300 символов — тик видел «✅ Готово: Рефлексия…»
без самого предложения; два мин-тика (10:11–10:17) искали по коду «что за
переключатель» и упирались в бюджет, ответ — через 8 минут «посмотрю».
- журнал исходящих: send_message пишет OutboundMessage (chat:message_id →
  полный текст до 4000) — любой ответ/доставка/уведомление;
- receive_message: reply на бота → reply_to_full по message_id; для
  старых сообщений — по началу текста среди DeliverySent/ReplySent;
- контекст тика: «↳ ЭТО ОТВЕТ НА ТВОЁ СООБЩЕНИЕ (полный текст)»;
- память тика: «ТВОИ НЕДАВНИЕ СООБЩЕНИЯ» — собственные рассылки бота
  (итоги задач, уведомления, предложения) за 48 ч, без дублей с ответами;
- kodi_core.status: прогноз rule-ветки без p_fit (коммит часа a163e88,
  проверен: precision 0.74 / recall 0.52 на 93 задачах) не валит статус.
Тест: test_reply_memory 9 PASS; полный набор — без регрессий.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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,26 +827,41 @@ 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:
+        p = nov = None
-            return None
+        if model is not None:
-        t = {"title": task.get("title"), "details": task.get("details"),
+            t = {"title": task.get("title"), "details": task.get("details"),
-             "est": task.get("est_minutes"), "await": task.get("await_delivery")}
+                 "est": task.get("est_minutes"), "await": task.get("await_delivery")}
-        X, em = task_featurize([t])
+            X, em = task_featurize([t])
-        p = float(predict(model, X)[0])
+            p = float(predict(model, X)[0])
-        nov = float(novelty(model, em)[0])
+            nov = float(novelty(model, em)[0])
     except Exception as e:
         return {"error": str(e)[:200]}
-    fam = float(model["nov_q"][0])
+    ext = len(_EXT_RE.findall(text))
-    r = "student" if (nov < fam and abs(p - .5) >= CONF) else "rule"
+    if p is not None and nov is not None:
-    rec = {"head": TASK_HEAD, "task": task.get("id"), "p_fit": round(p, 4),
+        fam = float(model["nov_q"][0])
-           "novelty": round(nov, 4), "route": r, "mode": m,
+        if nov < fam and abs(p - .5) >= CONF:      # ученик уверен — он и решает
-           "act": bool(m == "on" and r == "student" and p < .5)}
+            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
 
@@ -1007,8 +1027,14 @@ def status() -> dict:
     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]
+    # прогноз «влезет» = p_fit ≥ 0.5; у rule-ветки без чемпиона p_fit=None —
+    # её прогноз: rule:ext → «не влезет», rule → «влезет»
+    def _says_fit(p):
+        if p.get("p_fit") is not None:
+            return p["p_fit"] >= .5
+        return p.get("route") != "rule:ext"
     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)
+    out[TASK_HEAD]["outcome_acc_7d"] = (round(sum(_says_fit(p) == (y >= .5) for p, y in judged)
                                               / len(judged), 3) if judged else None)
     out[TASK_HEAD]["judged_7d"] = len(judged)
     return out

harness/scheduler.py

@@ -126,6 +126,37 @@ def _recent_docs(model: dict, chat_id, limit=3):
     return out
 
 
+def _own_messages(chat_ids, replied_hist, hours: int = 48, cap: int = 6) -> list:
+    """Собственные исходящие бота (не ответы на входящие) за последние часы —
+    в память тика. Раньше память разговора = только входящие + наши ответы:
+    итог задачи/уведомление/предложение, на которое Alex потом отвечает,
+    из памяти выпадали."""
+    if not chat_ids:
+        return []
+    since = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
+    replies = {(m.get("reply_text") or "")[:200] for m in replied_hist}
+    out = []
+    try:
+        db = sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", uri=True, timeout=30)
+        try:
+            rows = db.execute(
+                "SELECT timestamp, data FROM events WHERE event_type='OutboundMessage' "
+                "AND timestamp > ? ORDER BY id DESC LIMIT 200", (since,)).fetchall()
+        finally:
+            db.close()
+    except Exception:
+        return []
+    for ts, d in rows:
+        d = json.loads(d)
+        t = d.get("text") or ""
+        if d.get("chat_id") not in chat_ids or t[:200] in replies or t.startswith("(сводный"):
+            continue
+        out.append(f"- [{ts[5:16].replace('T', ' ')}] {t[:800]}")
+        if len(out) >= cap:
+            break
+    return list(reversed(out))
+
+
 def _est_minutes(t: dict) -> int:
     """Оценка длительности: из TaskSubmitted или дефолт по классу.
 
@@ -431,7 +462,13 @@ def build_context(level_cfg, model: dict, batch=None):
                                  or (m.get("reply_to_desc") or "").strip())
                             else (" [reply на сообщение без текста]"
                                   if m.get("reply_to_message_id") is not None else ""))
-                         + ": " + m.get('text','')[:1200])
+                         + ": " + m.get('text','')[:1200]
+                         # ответ на НАШЕ сообщение: полный его текст, а не 300
+                         # символов из Telegram (аудит 2026-10-17: «Ок делай»
+                         # на итог рефлексии — тик не видел, что именно делать)
+                         + (f"\n   ↳ ЭТО ОТВЕТ НА ТВОЁ СООБЩЕНИЕ (полный текст):\n   "
+                            + m["reply_to_full"][:3000].replace("\n", "\n   ")
+                            if m.get("reply_to_full") else ""))
         batch_ids = ",".join(ids)
         chat_ids = {m.get("chat_id") for m in inbox if m.get("chat_id") is not None}
         # t-261013131228: если в батче есть reply на сообщение без текста —
@@ -459,6 +496,11 @@ def build_context(level_cfg, model: dict, batch=None):
                         if m.get("replied") and not m.get("ignored")
                         and m.get("chat_id") in chat_ids
                         and (m.get("text") or "").strip()]
+        own = _own_messages(chat_ids, replied_hist)
+        if own:
+            parts.append("ТВОИ НЕДАВНИЕ СООБЩЕНИЯ В ЭТОТ ЧАТ (сам отправил: итоги задач, "
+                         "уведомления, предложения — Alex может отвечать на них):\n"
+                         + "\n".join(own))
         if replied_hist:
             replied_hist = sorted(replied_hist, key=lambda x: x.get("ts") or "")[-12:]
             hist_lines = []
@@ -581,10 +623,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",
+                "to": "M", "from": "S", "reason": why + " — сразу к часу"})
-                "reason": f"ядро Коди: не влезет в мин-тик (p_fit={rec['p_fit']}) — сразу к часу"})
+            _log(f"core: {t['id']} S→M до мин-попытки ({rec.get('route')}, "
-            _log(f"core: {t['id']} S→M до мин-попытки (p_fit={rec['p_fit']})")
+                 f"p_fit={rec.get('p_fit')}, ext={rec.get('ext_hits')})")
             continue
         keep.append(t)
     return keep

harness/tg_inbox.py

@@ -8,6 +8,7 @@
 getUpdates offset=last+1. Идемпотентность: update_id → aggregate_id.
 """
 import json
+import re
 import os
 import sys
 import time
@@ -66,14 +67,63 @@ def send_message(chat_id, text: str, reply_to=None, timeout: int = 35):
         # сообщение удалили — ответ всё равно уходит (без цитаты), не теряется
         params["allow_sending_without_reply"] = "true"
     try:
-        return _api("sendMessage", params, timeout=timeout)
+        r = _api("sendMessage", params, timeout=timeout)
     except urllib.error.HTTPError as e:
         # 400 от кривого HTML (незакрытый тег, «a < b» в тексте) — шлём как plain text.
         # Иначе ответ/доставка молча теряются: исключение глотается на верхнем уровне.
-        if e.code == 400:
+        if e.code != 400:
-            params.pop("parse_mode", None)
+            raise
-            return _api("sendMessage", params, timeout=timeout)
+        params.pop("parse_mode", None)
-        raise
+        r = _api("sendMessage", params, timeout=timeout)
+    _record_outbound(chat_id, r, text, reply_to)
+    return r
+
+
+def _record_outbound(chat_id, r, text: str, reply_to=None):
+    """Журнал исходящих (аудит 2026-10-17): message_id → полный текст. Alex
+    ответил «Ок делай» на итог рефлексии, а Telegram отдаёт в reply только
+    начало, харнес хранил 300 символов — тик не видел самого предложения.
+    Best effort: журнал не мешает отправке."""
+    try:
+        res = (r or {}).get("result") if isinstance(r, dict) else None
+        if not (isinstance(res, dict) and res.get("message_id")):
+            return
+        es.append_event("outbound", f"{chat_id}:{res['message_id']}", "OutboundMessage", {
+            "chat_id": chat_id, "message_id": res["message_id"], "text": (text or "")[:4000],
+            "reply_to": reply_to})
+    except Exception:
+        pass
+
+
+def _strip_html(s: str) -> str:
+    return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", s or "")).strip()
+
+
+def resolve_bot_message(chat_id, message_id, prefix: str = "") -> str:
+    """Полный текст НАШЕГО сообщения, на которое ответили: журнал исходящих по
+    (chat, message_id); для старых — по совпадению начала с доставками/ответами."""
+    import sqlite3
+    try:
+        db = sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", uri=True, timeout=30)
+        try:
+            row = db.execute("SELECT data FROM events WHERE aggregate='outbound' AND aggregate_id=? "
+                             "LIMIT 1", (f"{chat_id}:{message_id}",)).fetchone()
+            if row:
+                return json.loads(row[0]).get("text") or ""
+            want = _strip_html(prefix)[:150]
+            if len(want) < 20:
+                return ""
+            for (d,) in db.execute(
+                    "SELECT data FROM events WHERE event_type IN ('DeliverySent','ReplySent',"
+                    "'FollowupSent','OutboundMessage') ORDER BY id DESC LIMIT 3000"):
+                t = json.loads(d).get("text") or ""
+                if _strip_html(t).startswith(want):
+                    return t
+        finally:
+            db.close()
+    except Exception:
+        pass
+    return ""
 
 
 def set_reaction(chat_id, message_id: int, emoji: str = "👀", is_big: bool = False):
@@ -404,6 +454,11 @@ def receive_message(update: dict):
         data["reply_to_is_bot"] = bool(rfrom.get("is_bot"))
         data["reply_to_first_name"] = rfrom.get("first_name")
     data["mentions_bot"] = _mentions_bot(msg)
+    if data.get("reply_to_is_bot") and data.get("reply_to_message_id"):
+        full = resolve_bot_message(chat.get("id"), data["reply_to_message_id"],
+                                   data.get("reply_to_text") or "")
+        if full:
+            data["reply_to_full"] = full[:4000]
     es.append_event("inbox", f"m-{update['update_id']}", "MessageReceived", data)
     return f"m-{update['update_id']}"
 

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)

tests/test_reply_memory.py

@@ -0,0 +1,116 @@
+# -*- coding: utf-8 -*-
+"""Память разговора: ответ на сообщение бота видит его полностью (аудит 2026-10-17). Hermetic.
+
+Запуск: ./venv/bin/python tests/test_reply_memory.py
+
+Кейс: Alex ответил «Ок делай» на итог рефлексии; Telegram отдал в reply
+начало, харнес хранил 300 символов — тик не видел, что делать. Проверяет:
+send_message пишет OutboundMessage (message_id → полный текст); ответ на
+бота получает reply_to_full по message_id; для старых (до журнала) — по
+началу текста среди DeliverySent; контекст тика показывает полный текст и
+хвост за 300 символами; собственные рассылки бота — в памяти тика, ответы
+на входящие не дублируются; статус ядра не падает на прогнозе правила
+без p_fit.
+"""
+import os
+import re
+import sys
+import tempfile
+
+sys.path.insert(0, "/root/simbiot-harness")
+_tmp = tempfile.mkdtemp(prefix="simbiot_replymem_")
+os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db")
+os.environ["SIMBIOT_CORE_DIR"] = os.path.join(_tmp, "core")
+
+from harness import es, scheduler, tg_inbox
+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)
+
+
+P = 7569223131
+NEXT = [4964]
+
+
+def fake_api(method, params, timeout=35):
+    if method == "sendMessage":
+        NEXT[0] += 1
+        return {"ok": True, "result": {"message_id": NEXT[0], "chat": {"id": params["chat_id"]}}}
+    if method == "getMe":
+        return {"ok": True, "result": {"id": 7245454391, "username": "juno42bot"}}
+    return {"ok": True, "result": True}
+
+
+tg_inbox._api = fake_api
+tg_inbox._load_env = lambda: None
+
+PROPOSAL = ("✅ Готово: «Рефлексия задач»\n\nПравило найдено. " + "Обоснование. " * 40 +
+            "\n\nПРЕДЛОЖЕНИЕ АЛЕКСУ: включить — в .env добавить SIMBIOT_CORE_TASK_FIT=on.")
+assert len(PROPOSAL) > 600
+tg_inbox.send_message(P, PROPOSAL)
+ob = [e for e in es.get_events(0, 10**6) if e["event_type"] == "OutboundMessage"]
+check("исходящее записано с message_id", ob and '"message_id": 4965' in ob[-1]["data"])
+
+
+def recv(uid, text, reply_mid=None, reply_text=""):
+    msg = {"message_id": 5000 + uid, "date": 1, "chat": {"id": P, "type": "private"},
+           "from": {"id": P, "first_name": "Alex"}, "text": text}
+    if reply_mid:
+        msg["reply_to_message"] = {"message_id": reply_mid, "text": reply_text,
+                                   "from": {"id": 7245454391, "is_bot": True, "first_name": "Cody"}}
+    return tg_inbox.receive_message({"update_id": uid, "message": msg})
+
+
+mid = recv(1, "Ок делай", 4965, PROPOSAL[:300])
+m = es.run_projections()["inbox"][mid]
+check("reply на бота: полный текст по message_id", m.get("reply_to_full") == PROPOSAL)
+
+# старая доставка (до журнала): только DeliverySent
+OLD = "✅ Готово: «Старая задача»\n\n" + "Подробности. " * 50 + "\nИТОГ: сделать X."
+es.append_event("delivery", "t-old", "DeliverySent", {"task_id": "t-old", "text": OLD})
+mid2 = recv(2, "давай", 4000, OLD[:300])
+check("старое сообщение: полный текст по началу", es.run_projections()["inbox"][mid2].get("reply_to_full") == OLD)
+mid3 = recv(3, "?", 3999, "коротко")
+check("слишком короткое начало — не угадываем", not es.run_projections()["inbox"][mid3].get("reply_to_full"))
+
+# контекст тика
+ctx = scheduler.build_context(LEVELS["min"], es.run_projections(), batch=[])
+check("контекст: блок «ответ на твоё сообщение»", "ЭТО ОТВЕТ НА ТВОЁ СООБЩЕНИЕ" in ctx)
+check("контекст: хвост за 300 символами виден", "SIMBIOT_CORE_TASK_FIT=on" in ctx)
+
+# память: собственные рассылки, без дублей с ответами
+tg_inbox.notify_chat = tg_inbox.notify_chat  # реальная, через send_message
+tg_inbox._default_chat_id = lambda: P
+tg_inbox.send_message(P, "🧠 Ядро Коди — неделя: отчёт")
+tg_inbox.send_reply(mid3, "Это ответ на вопрос")
+ctx2 = scheduler.build_context(LEVELS["min"], es.run_projections(), batch=[])
+own = ""
+if "ТВОИ НЕДАВНИЕ СООБЩЕНИЯ" in ctx2:  # секция — до следующей известной секции
+    tail = ctx2.split("ТВОИ НЕДАВНИЕ СООБЩЕНИЯ", 1)[1]
+    ends = [tail.find(h) for h in ("НЕДАВНИЙ РАЗГОВОР", "ВХОДЯЩИЕ СООБЩЕНИЯ") if tail.find(h) >= 0]
+    own = tail[:min(ends)] if ends else tail
+check("память: рассылка бота в контексте", "Ядро Коди — неделя" in own, ctx2[:600])
+check("память: ответ на входящее не дублируется", "Это ответ на вопрос" not in own)
+
+# статус ядра с прогнозом правила без p_fit
+from harness import kodi_core
+es.append_event("core", "pred:task_fit:t-x", "CorePrediction",
+                {"head": "task_fit", "task": "t-x", "p_fit": None, "route": "rule:ext", "act": False})
+es.append_event("task", "t-x", "TaskSubmitted", {"id": "t-x", "title": "портал", "size": "S"})
+es.append_event("task", "t-x", "TaskClaimed", {"level": "min"})
+es.append_event("task", "t-x", "TaskClaimed", {"level": "min"})
+try:
+    st = kodi_core.status()["task_fit"]
+    check("статус ядра: прогноз правила без p_fit не валит", st["judged_7d"] == 1 and st["outcome_acc_7d"] == 1.0, st)
+except Exception as e:
+    check("статус ядра: прогноз правила без p_fit не валит", False, e)
+
+print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)")
+sys.exit(1 if fails else 0)