simbiot-harness/tests/test_reply_memory.py

# -*- 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)