"""Simbiot Harness — динамическая память Коди: глубина D0–D3 и подгрузка по смыслу. Третья ось куба 4×4×4 (круг доверия × тик × глубина). Раньше память тика — 12 последних реплик чата + свои сообщения за 48 ч + 1 запрос к базе знаний кооператива: на «чем ты занимался последний месяц» Коди отвечал по базе кооператива и наугад — поиска по СОБСТВЕННОЙ истории не было. Индекс (/memory/index.sqlite) — инкрементально из стора: conv — сообщение + ответ Коди (ReplySent; батч — ответ головы); out — собственные рассылки бота (OutboundMessage, не ответы); task — задача и её итог (TaskCompleted); digest — содержательные дайджесты час/день (заглушки — нет). Каждая запись хранит КРУГ ПРИВАТНОСТИ: сказанное в чате видно всем его участникам → круг записи = самый недоверенный участник чата (личка Alex — 1, «Правление» — 2); задача — её круг; дайджест — 1 (внутреннее). Поиск отдаёт только записи с кругом ≥ круга тика: тик с правлением не поднимет личку владельца. Эмбеддинги — nomic-embed-text (search_document/query). Глубина (choose_depth): D0 — только текущее (односложные «ок/спасибо»); D1 — разговор (как было); D2 — + эпизоды по смыслу (8 записей); D3 — + архив (15 записей + последние дайджесты). Правила: уровень (час — D2, день — D3), слова о прошлом («помнишь», «вчера», «за месяц», «как договаривались»…), ответ на старое сообщение бота, задачи в батче; потолок — круг (trust.max_depth). Решение пишется (DepthChosen) — сырьё для головы ядра «depth». """ import json import os import re import sqlite3 import threading from datetime import datetime, timedelta, timezone from . import es from . import paths as _paths try: import numpy as np except ImportError: np = None KINDS = ("conv", "out", "task", "digest") BUDGET = {2: (8, 5000), 3: (15, 12000)} # глубина → (записей, символов) SYNC_BATCH = 400 MIN_SCORE = float(os.environ.get("SIMBIOT_MEMORY_MIN_SCORE", "0.45")) # ниже — шум _LOCK = threading.Lock() _LAST = {} # уровень → последнее решение о глубине (для DepthChosen) PAST_RE = re.compile( r"помни|вспомн|вчера|позавчера|на прошлой|прошл(ый|ую|ой|ом|ые)\s+(раз|недел|месяц|встреч)|" r"как\s+договаривал|мы\s+же|мы\s+обсуждал|ранее|раньше|в\s+прошлый|" r"за\s+(последн\w+\s+)?(месяц|недел|дн)|что\s+(ты\s+)?(делал|сделал|было)|" r"история|итог[иа]?\b|на\s+чём\s+останов|напомни", re.IGNORECASE) ARCHIVE_RE = re.compile(r"за\s+(последн\w+\s+)?(месяц|недел|квартал|год)|" r"чем\s+(ты\s+)?занимал|что\s+(ты\s+)?делал|отч[её]т|итоги|" r"стратег|план\s+на", re.IGNORECASE) TRIVIAL_RE = re.compile(r"^\W*(ок|окей|ok|ага|да|нет|спасибо|спс|понял|принято|супер|" r"класс|норм|👍|🙏|❤️|🔥)\W*$", re.IGNORECASE) # ────────────────────────── хранилище ────────────────────────── def enabled() -> bool: """SIMBIOT_MEMORY=on (env > .env инстанса); дефолт off — тесты герметичны.""" from . import decision return decision._cfg("SIMBIOT_MEMORY", "off").lower() == "on" def _dir() -> str: d = (os.environ.get("SIMBIOT_MEMORY_DIR", "").strip() or os.path.join(_paths.home(), "memory")) os.makedirs(d, mode=0o700, exist_ok=True) return d def _db(): db = sqlite3.connect(os.path.join(_dir(), "index.sqlite"), timeout=30) db.execute("CREATE TABLE IF NOT EXISTS items (id TEXT PRIMARY KEY, kind TEXT, ts TEXT, " "chat_id TEXT, circle INTEGER, who TEXT, text TEXT, emb BLOB)") db.execute("CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT)") return db def _store_ro(): return sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", uri=True, timeout=30) def _embed(texts, prefix): from . import kodi_core return kodi_core.embed(texts, prefix) # ────────────────────────── круг приватности записи ────────────────────────── def chat_circle(chat_id, model: dict, senders: dict) -> int: """Круг чата = самый недоверенный известный участник; личка — круг собеседника.""" from . import trust known = senders.get(str(chat_id)) or set() if not known: try: cid = int(chat_id) if cid > 0: return trust.circle_of(f"tg:{cid}", model) except (TypeError, ValueError): pass return trust.circle_of(trust.subject_of({"chat_id": chat_id}), model) return max(trust.circle_of(s, model) for s in known) # ────────────────────────── синхронизация ────────────────────────── def sync(limit: int = SYNC_BATCH, log=None) -> int: """Новые события стора → записи индекса с эмбеддингами. Возвращает число записей.""" return _sync_once(limit, log)[1] def sync_all(log=None) -> int: """Догнать стор целиком (бэкфилл): пачками, пока есть события.""" total = 0 while True: nrows, nitems = _sync_once(SYNC_BATCH, log) total += nitems if nrows == 0: return total def _sync_once(limit: int, log=None) -> tuple: """(событий обработано, записей добавлено).""" from . import trust with _LOCK: db = _db() try: last = int((db.execute("SELECT v FROM meta WHERE k='last'").fetchone() or ["0"])[0]) src = _store_ro() try: rows = src.execute( "SELECT id, aggregate_id, event_type, timestamp, data FROM events WHERE id > ? " "AND event_type IN ('MessageReceived','ReplySent','OutboundMessage'," "'TaskSubmitted','TaskCompleted','DigestWritten') ORDER BY id LIMIT ?", (last, limit)).fetchall() if not rows: return 0, 0 model = es.run_projections() senders = json.loads((db.execute("SELECT v FROM meta WHERE k='senders'") .fetchone() or ["{}"])[0]) senders = {k: set(v) for k, v in senders.items()} pending = [] # (id, kind, ts, chat, circle, who, text) for eid, aid, et, ts, d in rows: d = json.loads(d) if et == "MessageReceived": if d.get("chat_id") is not None: senders.setdefault(str(d["chat_id"]), set()).add(trust.subject_of(d)) elif et == "ReplySent": rec = _conv_item(src, db, aid, d, ts, model, senders, pending) if rec: pending.append(rec) # исходящее с тем же текстом — это ответ, не рассылка db.execute("DELETE FROM items WHERE kind='out' AND chat_id=? " "AND substr(text,1,200)=?", (str(rec[3]), (d.get("text") or "")[:200])) pending[:] = [p for p in pending if not ( p[1] == "out" and p[3] == rec[3] and p[6][:200] == (d.get("text") or "")[:200])] elif et == "OutboundMessage": t = (d.get("text") or "").strip() # ответ на входящее (ReplySent мог прийти раньше) — это conv, не рассылка is_reply = any(p[1] == "conv" and p[3] == str(d.get("chat_id")) and p[6].split("\nКоди: ", 1)[-1][:200] == t[:200] for p in pending) or db.execute( "SELECT 1 FROM items WHERE kind='conv' AND chat_id=? AND " "instr(text, ?) > 0 LIMIT 1", (str(d.get("chat_id")), "\nКоди: " + t[:200])).fetchone() if t and not t.startswith("(сводный") and not is_reply: pending.append((f"out:{aid}", "out", ts, str(d.get("chat_id")), chat_circle(d.get("chat_id"), model, senders), "Коди", t[:1500])) elif et == "TaskCompleted": t = (model.get("tasks") or {}).get(aid) or {} ans = (d.get("answer") or d.get("result") or "") if t.get("title") or ans: pending.append((f"task:{aid}", "task", ts, "", trust.norm(t.get("circle")), "задача", f"Задача: {t.get('title') or aid}\n" f"Итог: {str(ans)[:1200]}")) elif et == "DigestWritten": txt = (d.get("text") or "").strip() if (d.get("level") in ("hour", "day") and not d.get("stub") and txt and not txt.startswith("[gate:quiet]") and "LLM не запускался" not in txt): pending.append((f"digest:{d.get('level')}:{d.get('period')}", "digest", ts, "", 1, f"дайджест {d.get('level')}", txt[:1500])) finally: src.close() if pending: vecs = _embed([p[6] for p in pending], "search_document") db.executemany("INSERT OR REPLACE INTO items VALUES (?,?,?,?,?,?,?,?)", [(*p, np.asarray(v, np.float32).tobytes()) for p, v in zip(pending, vecs)]) db.execute("INSERT OR REPLACE INTO meta VALUES ('last', ?)", (str(rows[-1][0]),)) db.execute("INSERT OR REPLACE INTO meta VALUES ('senders', ?)", (json.dumps({k: sorted(v) for k, v in senders.items()}),)) db.commit() if log and pending: log(f"memory: +{len(pending)} записей") return len(rows), len(pending) finally: db.close() def _conv_item(src, db, aid, d, ts, model, senders, pending=()): row = src.execute("SELECT data FROM events WHERE aggregate='inbox' AND aggregate_id=? " "AND event_type='MessageReceived' LIMIT 1", (aid,)).fetchone() if not row: return None m = json.loads(row[0]) text = (m.get("text") or "").strip() if not text: return None reply = d.get("text") or "" if reply.startswith("(сводный"): head = ((d.get("batch") or {}).get("head")) hp = next((p[6] for p in pending if p[0] == f"conv:{head}"), None) # голова в этой пачке hr = (hp,) if hp else db.execute("SELECT text FROM items WHERE id=?", (f"conv:{head}",)).fetchone() reply = (hr[0].split("\nКоди: ", 1)[-1] if hr else "(общий ответ на несколько сообщений)") who = m.get("first_name") or m.get("username") or str(m.get("user_id")) return (f"conv:{aid}", "conv", ts, str(m.get("chat_id")), chat_circle(m.get("chat_id"), model, senders), who, f"{who}: {text[:700]}\nКоди: {reply[:700]}") # ────────────────────────── поиск ────────────────────────── def search(query: str, circle: int, k: int = 8, kinds=KINDS, exclude_chat=None, exclude_since: str = "") -> list: """Записи по смыслу, видимые кругу тика (круг записи ≥ круга тика).""" if np is None or not query.strip(): return [] db = _db() try: rows = db.execute( f"SELECT id, kind, ts, chat_id, circle, who, text, emb FROM items WHERE circle >= ? " f"AND kind IN ({','.join('?' * len(kinds))})", (int(circle), *kinds)).fetchall() finally: db.close() if exclude_chat is not None and exclude_since: # свежий разговор этого чата уже в контексте (D1) — не дублируем rows = [r for r in rows if not (r[3] == str(exclude_chat) and r[2] >= exclude_since)] if not rows: return [] q = np.asarray(_embed([query[:2000]], "search_query")[0], np.float32) M = np.stack([np.frombuffer(r[7], np.float32) for r in rows]) sims = M @ q order = np.argsort(-sims)[:k] return [{"id": rows[i][0], "kind": rows[i][1], "ts": rows[i][2], "circle": rows[i][4], "text": rows[i][6], "score": round(float(sims[i]), 3)} for i in order] def stats() -> dict: db = _db() try: by = dict(db.execute("SELECT kind, COUNT(*) FROM items GROUP BY kind").fetchall()) circ = dict(db.execute("SELECT circle, COUNT(*) FROM items GROUP BY circle").fetchall()) last = (db.execute("SELECT v FROM meta WHERE k='last'").fetchone() or ["0"])[0] finally: db.close() return {"items": by, "by_circle": circ, "last_event": int(last)} # ────────────────────────── глубина ────────────────────────── def choose_depth(level: str, inbox: list, batch: list, circle: int) -> tuple: """(глубина 0..3, причины). Правила + потолок круга; голова ядра — в тени.""" from . import trust reasons = [] base = {"min": 1, "hour": 2, "day": 3}.get(level, 1) d = base reasons.append(f"уровень {level} → D{base}") texts = [(m.get("text") or "") for m in inbox or []] if inbox and not batch and all(TRIVIAL_RE.match(t.strip() or "x") for t in texts) \ and not any(m.get("reply_to_full") for m in inbox): d = 0 reasons = ["односложные реплики → D0"] joined = "\n".join(texts + [f"{t.get('title') or ''} {t.get('details') or ''}" for t in batch or []]) if PAST_RE.search(joined): d = max(d, 2) reasons.append("отсылка к прошлому → D2") if ARCHIVE_RE.search(joined): d = max(d, 3) reasons.append("обзор периода → D3") old = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat() for m in inbox or []: if m.get("reply_to_full") and (m.get("reply_to_ts") or "") < old: d = max(d, 2) reasons.append("ответ на давнее сообщение бота → D2") if batch: d = max(d, 2) reasons.append("задачи в батче → D2 (прошлые похожие)") # голова ядра «depth» (harness/kodi_core.py): shadow — прогноз в причинах; # on — уверенное «нужна глубина» поднимает D0/D1 до D2 student = None try: from . import kodi_core student = kodi_core.depth_predict({"query": joined[:1500], "level": level, "circle": circle, "reasons": " ".join(reasons)}) except Exception: student = None if student: p, sure = student reasons.append(f"ученик depth: p={p:.2f}{' уверен' if sure else ''}") if sure and p >= .5 and d < 2 and kodi_core.mode(kodi_core.DEPTH_HEAD) == "on": d = 2 reasons.append("ученик → D2") cap = trust.max_depth(circle) if d > cap: reasons.append(f"потолок круга {circle} → D{cap}") d = cap _LAST[level] = {"depth": d, "reasons": reasons, "circle": circle, "query": joined[:500]} return d, reasons def context_block(depth: int, level: str, inbox: list, batch: list, circle: int) -> str: """Текст памяти для контекста тика на глубине ≥ 2 (пусто — для D0/D1).""" if depth < 2 or not enabled(): return "" k, budget = BUDGET[min(depth, 3)] query = "\n".join([(m.get("text") or "") + " " + (m.get("reply_to_full") or "")[:300] for m in inbox or []] + [f"{t.get('title') or ''}\n{(t.get('details') or '')[:400]}" for t in batch or []]) if not query.strip(): query = {"hour": "итоги последнего часа, задачи, решения", "day": "итоги дня, решения, задачи, договорённости"}.get(level, "недавние дела") chat = inbox[0].get("chat_id") if inbox else None since = (datetime.now(timezone.utc) - timedelta(hours=48)).isoformat() try: hits = search(query, circle, k=k, exclude_chat=chat, exclude_since=since) except Exception as e: return f"ПАМЯТЬ: поиск недоступен ({str(e)[:80]}) — отвечай из контекста тика." lines, used = [], 0 for h in hits: if h["score"] < MIN_SCORE: continue line = f"- [{h['kind']} {h['ts'][:10]}] {h['text'][:900]}" if used + len(line) > budget: break lines.append(line) used += len(line) if _LAST.get(level): _LAST[level]["hits"] = len(lines) if not lines: return "" head = (f"ПАМЯТЬ — НАЙДЕНО ПО СМЫСЛУ (глубина D{depth}; записи, видимые кругу {circle}; " "это прошлое — опирайся, но проверяй актуальность):") return head + "\n" + "\n".join(lines) def last_choice(level: str) -> dict: return dict(_LAST.get(level) or {})