From 2b266caab5023b55c5ce4fac8af80f5a235832a0 Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Fri, 16 Oct 2026 20:10:26 +0000 Subject: [PATCH] =?UTF-8?q?=D1=81=D0=B6=D0=B0=D1=82=D0=B8=D0=B5=20=D1=81?= =?UTF-8?q?=D1=82=D0=BE=D1=80=D0=B0=20=D1=81=20=D0=BF=D0=BE=D0=B4=D0=BF?= =?UTF-8?q?=D0=B8=D1=81=D1=8C=D1=8E=20+=20=C2=AB=D1=81=D0=BD=D0=B0=D1=87?= =?UTF-8?q?=D0=B0=D0=BB=D0=B0=20=D0=BE=D1=82=D0=B2=D0=B5=D1=82,=20=D0=BF?= =?UTF-8?q?=D0=BE=D1=82=D0=BE=D0=BC=20=D1=80=D0=B0=D0=B1=D0=BE=D1=82=D0=B0?= =?UTF-8?q?=C2=BB=20=D0=BD=D0=B0=20=D0=BC=D0=B8=D0=BD-=D1=83=D1=80=D0=BE?= =?UTF-8?q?=D0=B2=D0=BD=D0=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit harness/compact.py + cli.py compact [--dry-run|--vacuum|--verify-archives]: шум сек-уровня (CycleTicked sec, ReflexObserved без алерта/команд/каналов) старше часа → архив jsonl.gz (0600, строки целиком со sig) → мост цепочки (подписи унесённых строк, на которые ссылаются остающиеся, + маркеры boundary: для legacy-границ) → подписанный StoreCompacted с sha256 архива/строк/моста → удаление пачками. audit.verify_events принимает мост только при совпадении хеша с подписанным событием; проекции подгружают мост лениво, если хвост перекрыл сжатый диапазон (rebuild с нуля). Репетиция на копии прода: 1389 → 200 МБ, verify 484 с/3.4 ГБ → 29 с/420 МБ, архив 246 МБ — все 2.47 млн подписей проверены; rebuild_projections OK. Найдено репетицией: legacy-граница (id 466633) и коллизия id сжатия в одну секунду — оба покрыты тестом. scripts/compact-store.sh — процедура с VACUUM под остановленными писателями. мин-уровень: если до ответа нужна работа >30 с — сначала короткий tg-reply, итог — cli.py tg-reply --followup (в чат входящего: группа → группа; идемпотентно по тексту). p90 «старт LLM → ответ» был 223 с. hour: лимит 300 вызовов оставлен — реальный пик 205 (p90 169), 100 оборвал бы 8 из 19 тиков. Тесты: test_compact 25 PASS, test_followup 10 PASS; полный набор — без регрессий. Co-Authored-By: Claude Opus 5.5 --- cli.py | 62 +++++++++- harness/audit.py | 48 +++++++- harness/compact.py | 247 +++++++++++++++++++++++++++++++++++++++ harness/es.py | 11 +- harness/levels.py | 5 + harness/tg_inbox.py | 31 +++++ scripts/compact-store.sh | 19 +++ tests/test_compact.py | 148 +++++++++++++++++++++++ tests/test_followup.py | 73 ++++++++++++ 9 files changed, 637 insertions(+), 7 deletions(-) create mode 100644 harness/compact.py create mode 100755 scripts/compact-store.sh create mode 100644 tests/test_compact.py create mode 100644 tests/test_followup.py diff --git a/cli.py b/cli.py index ee9cc55..63c39a8 100644 --- a/cli.py +++ b/cli.py @@ -55,6 +55,9 @@ def main(): p_tgr = sub.add_parser("tg-reply", help="ответить на входящее TG-сообщение") p_tgr.add_argument("msg_id", help="id входящего (m-NNN), для батча — несколько через запятую") p_tgr.add_argument("text") + p_tgr.add_argument("--followup", action="store_true", + help="досылка итога в чат УЖЕ отвеченного входящего (сначала ответ — потом работа); " + "повтор с тем же текстом — SKIP") p_tgr.add_argument("-c", "--close", action="store_true", help="батч-режим: один сводный ответ закрывает все указанные входящие " "(t-260930064924)") @@ -190,6 +193,15 @@ def main(): p_mcs = sub.add_parser("mcp-stats", help="статистика MCP-вызовов по журналу mcp_calls.log (t-261015091108)") p_mcs.add_argument("--tail", type=int, default=10, help="последние N вызовов (0 — только сводка)") + p_cmp2 = sub.add_parser("compact", help="сжатие стора: шум сек-уровня → подписанный архив + мост цепочки") + p_cmp2.add_argument("--dry-run", action="store_true", help="только посчитать кандидатов") + p_cmp2.add_argument("--older-than-hours", type=float, default=1.0, + help="трогать только события старше N часов (1)") + p_cmp2.add_argument("--vacuum", action="store_true", + help="VACUUM стора (эксклюзивный лок: писатели должны стоять)") + p_cmp2.add_argument("--verify-archives", action="store_true", + help="проверить все архивы: sha256 против StoreCompacted + подпись каждой строки") + p_dec = sub.add_parser("decide", help="decision-модель nimble (Ollama /v1/systemone): статус, вопрос, статистика") p_dec.add_argument("-q", "--question", help="вопрос/инструкция (без -o — noul: P(да))") p_dec.add_argument("-o", "--option", action="append", default=[], @@ -290,7 +302,16 @@ def main(): # срез лестницы по запросу из TG — видимостью управляет сам харнес args.text = "📊 " + tg_inbox.ladder_snapshot() ids = [x.strip() for x in args.msg_id.split(",") if x.strip()] - if len(ids) > 1 or (args.close and len(ids) == 1): + if args.followup: + import hashlib + head = ids[0] # батч — один чат: досылаем к голове + aid = f"followup:{head}:{hashlib.sha1(args.text.encode()).hexdigest()[:10]}" + ok = commands.deliver_artifact( + aid, kind="message", owner="cli:tg-followup", caption=args.text, + send=lambda: tg_inbox.send_followup(head, args.text)) + print(f"OK: итог дослан в чат {head}" if ok + else "SKIP: уже дослано (или чат недоступен)") + elif len(ids) > 1 or (args.close and len(ids) == 1): if not args.close: print("ERROR: несколько id без -c/--close — укажите флаг для сводного батч-ответа") sys.exit(1) @@ -555,7 +576,9 @@ def main(): print(json.dumps(rep, ensure_ascii=False)) else: print(f"событий в аудите: {rep['total']} | подписано криптоключом: " - f"{rep['checked']} | genesis: {rep['genesis']} | legacy: {rep['legacy']}") + f"{rep['checked']} | genesis: {rep['genesis']} | legacy: {rep['legacy']}" + + (f" | сжатий: {rep['compactions']} (мост сверен; архивы — " + "cli.py compact --verify-archives)" if rep.get("compactions") else "")) if rep["ok"]: print("OK: подписи и тампер-цепочка валидны" + (f" (публичный ключ {audit.key_id()})" if rep["checked"] else @@ -594,6 +617,41 @@ def main(): es.init_db() for t in commands.list_tasks(): print(f"{t['id']} [{t['size']:2}] {t['status']:7} {t['title'][:70]}") + elif args.cmd == "compact": + from harness import compact + es.init_db() + if args.verify_archives: + ok_all = True + for e in es.get_events(0, 10**9): + if e["event_type"] != "StoreCompacted": + continue + d = json.loads(e["data"]) + path = os.path.join(compact.archive_dir(), d["archive"]) + if not os.path.exists(path): + print(f"{d['compaction']}: АРХИВ НЕ НАЙДЕН {path}") + ok_all = False + continue + r = compact.verify_archive(path, d.get("archive_sha256")) + ok_all &= r["ok"] and r["rows"] == d.get("rows") + print(f"{d['compaction']}: {'OK' if r['ok'] and r['rows'] == d.get('rows') else 'ПРОВАЛ'} " + f"строк {r['rows']}/{d.get('rows')}, подписи {r['signed_ok']}, genesis " + f"{r['genesis']}, legacy {r['legacy']}, sha256 {'совпал' if r['sha_ok'] else 'НЕ СОВПАЛ'}" + + (f", первая битая id={r['first_bad']}" if r["first_bad"] else "")) + sys.exit(0 if ok_all else 2) + if args.vacuum: + r = compact.vacuum() + print(f"VACUUM: {r['before'] / 1e6:.0f} МБ → {r['after'] / 1e6:.0f} МБ") + return + if args.dry_run: + p = compact.plan(args.older_than_hours) + print(f"кандидатов {p['candidates']} из {p['total']} (id {p['first_id']}..{p['last_id']}, " + f"граница id≤{p['upto_id']}); стор {p['db_bytes'] / 1e6:.0f} МБ") + return + r = compact.compact(args.older_than_hours) + if r.get("compaction"): + print(f"OK {r['compaction']}: в архив {r['rows']}, удалено {r['removed']}, " + f"мост {r['bridge_count']}; архив {r['archive_path']}") + elif args.cmd == "decide": from harness import decision if args.stats: diff --git a/harness/audit.py b/harness/audit.py index 9097097..1a5fab9 100644 --- a/harness/audit.py +++ b/harness/audit.py @@ -161,7 +161,7 @@ def sign_event_row(aggregate, aggregate_id, event_type, data_str, timestamp, return {"sig": sign_bytes(payload), "key_id": kid, "prev_sig": prev_sig or ""} -def verify_rows(rows: list, pub_b64: str = None) -> dict: +def verify_rows(rows: list, pub_b64: str = None, bridge: set = None) -> dict: """Проверка списка событий (dict'и sqlite-строк, id-порядок). - genesis: — ретро-маркер строк ДО внедрения подписи (доверенная @@ -181,7 +181,10 @@ def verify_rows(rows: list, pub_b64: str = None) -> dict: checked = genesis = legacy = link_checks = 0 first_bad = None last_signed_sig = None # sig последней проверенной подписанной строки - earlier_sigs = set() # sig всех ранее проверенных подписанных строк + # sig всех ранее проверенных подписанных строк + мост сжатия (подписи + # строк, унесённых в архив harness/compact.py; хеш моста сверен с + # подписанным StoreCompacted в verify_events ДО вызова) + earlier_sigs = set(bridge or ()) for r in rows: sig = r.get("sig") if sig and sig.startswith("genesis:"): @@ -202,7 +205,10 @@ def verify_rows(rows: list, pub_b64: str = None) -> dict: # Допускаем ссылку на любую существующую прошлую подписанную # строку (гонка писателей), НЕ только на соседнюю: подделать # prev нельзя без ключа — подпись payload'а его связывает. - if prev != last_signed_sig and prev not in earlier_sigs: + # boundary: из моста сжатия: перед строкой была неподписанная + # граница (prev пуст), унесённая в архив — подписанный маркер + if (prev != last_signed_sig and prev not in earlier_sigs + and not (prev is None and f"boundary:{r['id']}" in earlier_sigs)): first_bad = r["id"] if first_bad is None else first_bad break payload = payload_bytes(r["aggregate"], r["aggregate_id"], @@ -229,12 +235,46 @@ def verify_events(db_path: str = None, after_id: int = 0, limit: int = 10**9, rows = [dict(r) for r in db.execute( "SELECT * FROM events WHERE id > ? ORDER BY id LIMIT ?", (after_id, limit))] + bridge, bad_bridge, compactions = _load_bridge(db) db.close() - rep = verify_rows(rows, pub_b64) + if bad_bridge: + return {"checked": 0, "genesis": 0, "legacy": 0, "link_checks": 0, + "first_bad": f"bridge:{bad_bridge}", "ok": False, "total": len(rows), + "compactions": compactions} + rep = verify_rows(rows, pub_b64, bridge=bridge) rep["total"] = len(rows) + rep["compactions"] = compactions return rep +def _load_bridge(db): + """Мост сжатия: (подписи, сбойное сжатие | None, число сжатий). + + Каждая группа compaction_bridge обязана совпасть по sha256 с + bridge_sha256 своего StoreCompacted (событие подписано и само проходит + цепочку) — иначе мост подменён и ссылкам через него не верим.""" + try: + groups = {} + for sig, comp in db.execute("SELECT sig, compaction FROM compaction_bridge"): + groups.setdefault(comp, []).append(sig) + except sqlite3.OperationalError: + groups = {} # стор без сжатий + signed = {} + for data, sig in db.execute( + "SELECT data, sig FROM events WHERE event_type='StoreCompacted'"): + if not sig or sig.startswith("genesis:"): + continue # неподписанное «сжатие» — не основание доверять мосту + d = json.loads(data) + signed[d.get("compaction")] = d + for comp in set(groups) | {c for c, d in signed.items() if d.get("bridge_count")}: + sigs = groups.get(comp, []) + d = signed.get(comp) + h = hashlib.sha256("\n".join(sorted(sigs)).encode()).hexdigest() + if not d or d.get("bridge_sha256") != h: + return set(), comp, len(signed) + return {s for sigs in groups.values() for s in sigs}, None, len(signed) + + # ────────────────────────── конверты команд (шина CLI) ────────────────────────── def sign_command(command: dict) -> dict: diff --git a/harness/compact.py b/harness/compact.py new file mode 100644 index 0000000..f0162b7 --- /dev/null +++ b/harness/compact.py @@ -0,0 +1,247 @@ +"""Simbiot Harness — сжатие стора с подписью (аудит 2026-10-16). + +Сек-уровень до фикса писал CycleTicked + ReflexObserved каждую секунду: +~3 млн подписанных событий шума, 98% стора (1.37 ГБ), полный `cli.py +verify` — 8 минут и 3.4 ГБ памяти. Сжатие уносит этот шум в архив, не +ломая криптоаудит: + +1. Кандидаты — только шум сек-уровня старше --older-than-hours (1 ч): + CycleTicked level=sec и ReflexObserved без алерта/команд/каналов. + Всё остальное (задачи, сообщения, трейсы, heartbeat'ы с событием) — в сторе. +2. Архив /archive/events---.jsonl.gz (0600): строки + целиком, со sig/key_id/prev_sig — каждую можно проверить публичным + ключом (verify_archive). sha256 архива и потока строк фиксируются. +3. Мост: подписи удалённых строк, на которые ссылается prev_sig оставшихся + (цепочка иначе рвётся), — в таблицу compaction_bridge. Их sha256 (по + отсортированному списку) входит в подписанное событие StoreCompacted — + подменить мост без ключа харнеса нельзя; audit.verify_events проверяет + хеш моста против подписанного события и только тогда принимает ссылки. +4. Удаление пачками по BATCH строк короткими транзакциями — демоны + пишут дальше (busy_timeout 30 с с запасом). VACUUM — отдельно + (--vacuum), под остановленными писателями: он берёт эксклюзивный лок. + +Повторный запуск безопасен: новые кандидаты → новое сжатие со своим +мостом; события append-only, критерий детерминирован. +""" +import gzip +import hashlib +import json +import os +import secrets +import sqlite3 +import time +from datetime import datetime, timedelta, timezone + +from . import es +from . import paths as _paths + +BATCH = 20_000 + +NOISE_SQL = ( + "((event_type='CycleTicked' AND json_extract(data,'$.level')='sec') OR " + " (event_type='ReflexObserved'" + " AND json_extract(data,'$.alert') IS NULL" + " AND json_extract(data,'$.commands') IS NULL" + " AND json_extract(data,'$.portal_channel') IS NULL" + " AND json_extract(data,'$.agent_channel') IS NULL))" +) + +BRIDGE_DDL = ("CREATE TABLE IF NOT EXISTS compaction_bridge (" + "sig TEXT PRIMARY KEY, compaction TEXT NOT NULL)") + + +def archive_dir() -> str: + d = (os.environ.get("SIMBIOT_ARCHIVE_DIR", "").strip() + or os.path.join(_paths.home(), "archive")) + os.makedirs(d, mode=0o700, exist_ok=True) + return d + + +def bridge_hash(sigs) -> str: + """sha256 моста: отсортированные подписи через \\n — каноника для события.""" + return hashlib.sha256("\n".join(sorted(sigs)).encode()).hexdigest() + + +def _connect(db_path=None): + db = sqlite3.connect(db_path or es.DB_PATH, timeout=30) + db.execute("PRAGMA busy_timeout=30000") + db.row_factory = sqlite3.Row + return db + + +def _cutoff_id(db, older_than_hours: float) -> int: + """Последний id старше порога, но никогда не голова стора (на неё ссылается + prev_sig следующей записи).""" + cutoff = (datetime.now(timezone.utc) - timedelta(hours=older_than_hours)).isoformat() + head = db.execute("SELECT MAX(id) FROM events").fetchone()[0] or 0 + # id растёт вместе со временем: бинпоиск по PK вместо скана по timestamp + lo, hi = 0, head + while lo < hi: + mid = (lo + hi + 1) // 2 + row = db.execute("SELECT timestamp FROM events WHERE id >= ? ORDER BY id LIMIT 1", + (mid,)).fetchone() + if row and row[0] < cutoff: + lo = mid + else: + hi = mid - 1 + return min(lo, head - 1) + + +def plan(older_than_hours: float = 1.0, db_path=None) -> dict: + """Сухой прогон: сколько уйдёт в архив.""" + db = _connect(db_path) + upto = _cutoff_id(db, older_than_hours) + n, first, last = db.execute( + f"SELECT COUNT(*), MIN(id), MAX(id) FROM events WHERE id <= ? AND {NOISE_SQL}", + (upto,)).fetchone() + total = db.execute("SELECT COUNT(*) FROM events").fetchone()[0] + db.close() + return {"upto_id": upto, "candidates": n, "first_id": first, "last_id": last, + "total": total, "db_bytes": os.path.getsize(db_path or es.DB_PATH)} + + +def compact(older_than_hours: float = 1.0, log=print) -> dict: + """Архив → мост + подписанный StoreCompacted → удаление пачками. + + Работает со стором es.DB_PATH (SIMBIOT_DB / SIMBIOT_HOME) — тем же, куда + подписанный StoreCompacted пишет es.append_event.""" + db = _connect() + db.execute(BRIDGE_DDL) + db.commit() + upto = _cutoff_id(db, older_than_hours) + + # prev_sig всех строк, что ОСТАНУТСЯ: на какие подписи будут ссылки + kept_prev = {r[0] for r in db.execute( + f"SELECT prev_sig FROM events WHERE prev_sig IS NOT NULL AND prev_sig != '' " + f"AND NOT (id <= ? AND {NOISE_SQL})", (upto,))} + log(f"[compact] граница id≤{upto}; ссылок prev_sig у остающихся: {len(kept_prev)}") + + stamp = datetime.now(timezone.utc).strftime("%y%m%d%H%M%S") + comp_id = f"c-{stamp}-{secrets.token_hex(3)}" # уникален и в пределах секунды + tmp_path = os.path.join(archive_dir(), f".{comp_id}.jsonl.gz.tmp") + rows_h = hashlib.sha256() + bridge, n, first, last = [], 0, None, None + cols = None + with gzip.open(tmp_path, "wt", encoding="utf-8", compresslevel=6) as f: + cur = db.execute(f"SELECT * FROM events WHERE id <= ? AND {NOISE_SQL} ORDER BY id", + (upto,)) + for r in cur: + if cols is None: + cols = r.keys() + line = json.dumps({k: r[k] for k in cols}, ensure_ascii=False, + sort_keys=True, separators=(",", ":")) + f.write(line + "\n") + rows_h.update(line.encode() + b"\n") + n += 1 + first = r["id"] if first is None else first + last = r["id"] + if r["sig"] and r["sig"] in kept_prev: + bridge.append(r["sig"]) + if not n: + os.unlink(tmp_path) + db.close() + log("[compact] кандидатов нет") + return {"compaction": None, "removed": 0} + os.chmod(tmp_path, 0o600) + with open(tmp_path, "rb") as fh: + os.fsync(fh.fileno()) + arch = os.path.join(archive_dir(), f"events-{first}-{last}-{comp_id}.jsonl.gz") + os.replace(tmp_path, arch) + arch_h = hashlib.sha256() + with open(arch, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + arch_h.update(chunk) + log(f"[compact] архив {arch}: {n} строк, {os.path.getsize(arch) / 1e6:.1f} МБ; мост {len(bridge)}") + + # границы: подписанная строка с пустым prev_sig писалась после НЕподписанной + # (legacy) — легальная граница цепочки. Если та неподписанная уходит в + # архив, перед строкой окажется подписанная и link-check упадёт (кейс + # прод-репетиции, id 466633). Маркер boundary: в мосте (под тем же + # подписанным хешем) — «граница здесь была, унесена сжатием». + for (rid,) in db.execute( + f"SELECT id FROM events WHERE sig IS NOT NULL AND sig NOT LIKE 'genesis:%' " + f"AND (prev_sig IS NULL OR prev_sig = '') AND NOT (id <= ? AND {NOISE_SQL})", + (upto,)).fetchall(): + pred = db.execute( + f"SELECT id <= ? AND {NOISE_SQL} FROM events WHERE id < ? ORDER BY id DESC LIMIT 1", + (upto, rid)).fetchone() + if pred and pred[0]: + bridge.append(f"boundary:{rid}") + log(f"[compact] мост с маркерами границ: {len(bridge)}") + + # мост ДО события: событие фиксирует его хеш; сбой события → мост убираем + db.executemany("INSERT OR IGNORE INTO compaction_bridge (sig, compaction) VALUES (?, ?)", + [(s, comp_id) for s in bridge]) + db.commit() + info = { + "compaction": comp_id, "archive": os.path.basename(arch), + "archive_sha256": arch_h.hexdigest(), "rows_sha256": rows_h.hexdigest(), + "rows": n, "first_id": first, "last_id": last, "upto_id": upto, + "criteria": "sec-noise: CycleTicked level=sec, ReflexObserved без alert/commands/каналов", + "bridge_count": len(bridge), "bridge_sha256": bridge_hash(bridge), + } + res = es.append_event("harness", f"compaction:{comp_id}", "StoreCompacted", info) + if not res: + db.execute("DELETE FROM compaction_bridge WHERE compaction = ?", (comp_id,)) + db.commit() + db.close() + raise RuntimeError("StoreCompacted не записан — сжатие отменено, архив оставлен") + + removed = 0 + for start in range(first, last + 1, BATCH): + cur = db.execute(f"DELETE FROM events WHERE id BETWEEN ? AND ? AND {NOISE_SQL}", + (start, min(start + BATCH - 1, last))) + db.commit() + removed += cur.rowcount + time.sleep(0.02) # окно писателям между пачками + db.close() + log(f"[compact] удалено {removed} строк; VACUUM — отдельно (--vacuum, писатели остановлены)") + return {**info, "removed": removed, "archive_path": arch} + + +def vacuum(db_path=None) -> dict: + """VACUUM + checkpoint WAL: эксклюзивный лок — писатели должны стоять.""" + path = db_path or es.DB_PATH + before = os.path.getsize(path) + db = sqlite3.connect(path, timeout=60) + db.execute("PRAGMA wal_checkpoint(TRUNCATE)") + db.execute("VACUUM") + db.execute("PRAGMA wal_checkpoint(TRUNCATE)") + db.close() + return {"before": before, "after": os.path.getsize(path)} + + +def verify_archive(path: str, expect_sha256: str = None, pub_b64: str = None) -> dict: + """Архив сжатия: sha256 файла (против StoreCompacted) + подпись КАЖДОЙ строки.""" + from . import audit + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(1 << 20), b""): + h.update(chunk) + pub_b64 = pub_b64 or audit.public_key_b64() + rows = signed = genesis = legacy = 0 + first_bad = None + with gzip.open(path, "rt", encoding="utf-8") as f: + for line in f: + r = json.loads(line) + rows += 1 + sig = r.get("sig") + if not sig: + legacy += 1 + continue + if sig.startswith("genesis:"): + genesis += 1 + if sig != f"genesis:{r['id']}" and first_bad is None: + first_bad = r["id"] + continue + payload = audit.payload_bytes(r["aggregate"], r["aggregate_id"], r["event_type"], + r["data"], r["timestamp"], r["version"], + r.get("prev_sig") or None, r.get("key_id")) + if audit.verify_bytes(pub_b64, payload, sig): + signed += 1 + elif first_bad is None: + first_bad = r["id"] + sha_ok = expect_sha256 is None or h.hexdigest() == expect_sha256 + return {"rows": rows, "signed_ok": signed, "genesis": genesis, "legacy": legacy, + "first_bad": first_bad, "sha256": h.hexdigest(), "sha_ok": sha_ok, + "ok": first_bad is None and sha_ok} diff --git a/harness/es.py b/harness/es.py index 1287c08..8fdbbca 100644 --- a/harness/es.py +++ b/harness/es.py @@ -770,7 +770,16 @@ def run_projections(): # останавливаются — подделанное событие не попадает в read model) try: from . import audit as _audit - rep = _audit.verify_rows([dict(r) for r in rows]) + drows = [dict(r) for r in rows] + rep = _audit.verify_rows(drows) + if not rep["ok"]: + # хвост мог перекрыть сжатый диапазон (проекция отстала или + # rebuild с нуля): ссылки на унесённые в архив строки — через + # мост, сверенный с подписанным StoreCompacted (harness/compact.py). + # Лениво: обычный посекундный путь мост не грузит. + bridge, bad, _n = _audit._load_bridge(db) + if bridge and not bad: + rep = _audit.verify_rows(drows, bridge=bridge) if not rep["ok"]: raise RuntimeError( f"audit: подпись события id={rep['first_bad']} не прошла — " diff --git a/harness/levels.py b/harness/levels.py index 2be1dcf..8a45da3 100644 --- a/harness/levels.py +++ b/harness/levels.py @@ -73,6 +73,11 @@ LEVELS = { "ОДНО сообщение не должно остаться без ответа на следующий тик. " "Тон: кратко, по-русски, тёпло; " "можно выполнить просьбу инструментами и ответить результатом. " + "СНАЧАЛА ОТВЕТ, ПОТОМ РАБОТА: если до ответа нужна работа дольше " + "~30 с (поиск, ssh, правки, несколько шагов) — ПЕРВЫМ действием тика " + "короткий tg-reply («понял, делаю: …, отпишусь»), затем работа, итог — " + "cli.py tg-reply --followup \"итог\" (тот же чат, повтор — SKIP). " + "Простой ответ — сразу одним tg-reply, без досылки. " "Быстрые мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. " "НЕ создавай задачи из служебных/пустых сообщений («list», «тест», " "«привет», односложные реакции) — они засоряют доску клонами " diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index 41ba771..915a1ab 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -428,6 +428,37 @@ def send_reply(msg_id: str, text: str, timeout: int = 35): return msg_id +def send_followup(msg_id: str, text: str, timeout: int = 35) -> bool: + """Досылка результата в ЧАТ входящего, на которое уже ответили (аудит 2026-10-16). + + Правило мин-уровня «сначала короткий ответ, потом работа»: первый + tg-reply закрывает входящее (ReplySent), итог работы уходит сюда — в тот + же чат (группа → группа, личка → личка, портал/агент — своим каналом), + а не в личку по умолчанию, как tg-notify. Идемпотентность — снаружи + (commands.deliver_artifact по ключу followup::<хеш текста>). + """ + model = es.run_projections() + m = (model.get("inbox") or {}).get(msg_id) + if m is None: + raise ValueError(f"входящее {msg_id} не найдено") + chat = str(m.get("chat_id") or "") + if chat.startswith("portal:agent:"): + from . import agent_channel + ok = agent_channel.push_agent(chat.split(":", 2)[-1], text, "help") + elif chat.startswith("portal"): + from . import portal_channel + ok = portal_channel.push(m["chat_id"], text, [m.get("update_id") or msg_id]) + elif chat.startswith("tui:"): + ok = True # TUI читает стор: события FollowupSent достаточно + else: + r = send_message(m["chat_id"], text, reply_to=None, timeout=timeout) + ok = bool((r or {}).get("ok")) + if ok: + es.append_event("inbox", msg_id, "FollowupSent", { + "to_chat": m.get("chat_id"), "text": text[:2000]}) + return ok + + def send_batch_reply(msg_ids: list, text: str, timeout: int = 35): """Сводный ответ на БАТЧ входящих (t-260930064924): одна Telegram-отправка закрывает несколько сообщений одного разговора. diff --git a/scripts/compact-store.sh b/scripts/compact-store.sh new file mode 100755 index 0000000..2a1500d --- /dev/null +++ b/scripts/compact-store.sh @@ -0,0 +1,19 @@ +#!/bin/bash +# Сжатие боевого стора с подписью (harness/compact.py, аудит 2026-10-16): +# 1) шум сек-уровня → подписанный архив + мост цепочки, удаление пачками (демоны живы); +# 2) VACUUM — под остановленными писателями: ждём тихое окно (нет LLM-воркера), +# стоп лестницы и TG-поллера (Telegram копит апдейты — ничего не теряется), старт. +# Использование: scripts/compact-store.sh [older_than_hours=1] +set -euo pipefail +cd "$(dirname "$0")/.." +export SIMBIOT_HOME="${SIMBIOT_HOME:-/var/lib/simbiot-harness}" +PY=./venv/bin/python +$PY cli.py compact --older-than-hours "${1:-1}" +for i in $(seq 1 180); do + pgrep -f '/harness/_llm_worker\.py$' >/dev/null || { sleep 3; pgrep -f '/harness/_llm_worker\.py$' >/dev/null || break; } + sleep 10 +done +systemctl stop simbiot-harness simbiot-harness-tg +trap 'systemctl start simbiot-harness simbiot-harness-tg' EXIT +$PY cli.py compact --vacuum +$PY cli.py verify | tail -2 diff --git a/tests/test_compact.py b/tests/test_compact.py new file mode 100644 index 0000000..ecd4cc1 --- /dev/null +++ b/tests/test_compact.py @@ -0,0 +1,148 @@ +# -*- coding: utf-8 -*- +"""Сжатие стора с подписью (harness/compact.py) — hermetic: свой стор, свой ключ. + +Запуск: ./venv/bin/python tests/test_compact.py + +Проверяет: шум сек-уровня уходит в архив, значимое остаётся (алерт/команды +сек-уровня, задачи, сообщения, голова стора); полный verify до и после — +OK, мост сверен с подписанным StoreCompacted; подмена моста (убрать/ +добавить подпись, снести таблицу) → FAIL; архив: sha256 и подпись каждой +строки, правка строки → FAIL; повторное сжатие; VACUUM ужимает файл; +проекции после сжатия живы. +""" +import gzip +import json +import os +import sqlite3 +import sys +import tempfile + +sys.path.insert(0, "/root/simbiot-harness") +_tmp = tempfile.mkdtemp(prefix="simbiot_compact_") +os.environ["SIMBIOT_DB"] = os.path.join(_tmp, "harness.db") +os.environ["SIMBIOT_AUDIT_KEY_DIR"] = os.path.join(_tmp, "audit") +os.environ["SIMBIOT_ARCHIVE_DIR"] = os.path.join(_tmp, "archive") + +from harness import audit, compact, es + +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 fill(seq0, n): + for i in range(seq0, seq0 + n): + es.append_tick("sec", i) + es.append_event("cycle", f"reflex:{i}", "ReflexObserved", {"level": "sec", "seq": i, "load1": 0.5}) + if i % 50 == 0: + es.append_event("cycle", f"reflex-alert:{i}", "ReflexObserved", + {"level": "sec", "seq": i, "alert": "load 20 >= 12"}) + if i % 40 == 0: + es.append_event("tg", f"m-{i}", "MessageReceived", {"update_id": i, "text": "привет"}) + es.append_tick("min", i) + + +def types(): + db = sqlite3.connect(os.environ["SIMBIOT_DB"]) + out = dict(db.execute("SELECT event_type, COUNT(*) FROM events GROUP BY 1").fetchall()) + db.close() + return out + + +fill(1000, 300) +# legacy-граница: неподписанная шумовая строка (как писал старый код до +# рестарта демонов) и за ней подписанная с пустым prev — кейс id 466633 +_db = sqlite3.connect(os.environ["SIMBIOT_DB"]) +_db.execute("INSERT INTO events (aggregate,aggregate_id,event_type,data,timestamp,version) " + "VALUES ('cycle','reflex:legacy','ReflexObserved','{\"level\":\"sec\"}',?,1)", + (es.datetime.now(es.timezone.utc).isoformat(),)) +_db.commit(); _db.close() +es.append_event("tg", "m-after-legacy", "MessageReceived", {"update_id": 1, "text": "после legacy"}) +fill(2000, 60) +before = types() +v0 = audit.verify_events() +check("verify до сжатия — OK", v0["ok"] and v0["checked"] > 600, v0) +plan = compact.plan(older_than_hours=0) +check("plan: кандидаты — шум сек-уровня", plan["candidates"] >= 590, plan) + +r = compact.compact(older_than_hours=0, log=lambda *_: None) +after = types() +check("сжатие: строки ушли в архив и удалены", r["rows"] == r["removed"] >= 590, r) +check("значимое осталось: MessageReceived", after.get("MessageReceived") == before["MessageReceived"]) +check("значимое осталось: мин-тики", + sqlite3.connect(os.environ["SIMBIOT_DB"]).execute( + "SELECT COUNT(*) FROM events WHERE event_type='CycleTicked' " + "AND json_extract(data,'$.level')='min'").fetchone()[0] == 10) +check("маркер legacy-границы в мосте", + sqlite3.connect(os.environ["SIMBIOT_DB"]).execute( + "SELECT COUNT(*) FROM compaction_bridge WHERE sig LIKE 'boundary:%'").fetchone()[0] >= 1) +check("значимое осталось: ReflexObserved с алертом", + sqlite3.connect(os.environ["SIMBIOT_DB"]).execute( + "SELECT COUNT(*) FROM events WHERE event_type='ReflexObserved' " + "AND json_extract(data,'$.alert') IS NOT NULL").fetchone()[0] == 8) +check("StoreCompacted записан", after.get("StoreCompacted") == 1) +check("мост не пуст (ссылки остающихся на удалённых)", r["bridge_count"] > 0, r["bridge_count"]) + +v1 = audit.verify_events() +check("verify после сжатия — OK", v1["ok"] and v1.get("compactions") == 1, v1) +check("в цепочке меньше строк", v1["total"] < v0["total"] - 500, (v0["total"], v1["total"])) + +arch = os.path.join(os.environ["SIMBIOT_ARCHIVE_DIR"], r["archive"]) +va = compact.verify_archive(arch, r["archive_sha256"]) +check("архив: sha256 и подписи всех строк", va["ok"] and va["rows"] == r["rows"], va) +check("архив 0600", oct(os.stat(arch).st_mode & 0o777) == "0o600") + +# подделка моста +db = sqlite3.connect(os.environ["SIMBIOT_DB"]) +victim = db.execute("SELECT sig FROM compaction_bridge LIMIT 1").fetchone()[0] +db.execute("DELETE FROM compaction_bridge WHERE sig=?", (victim,)); db.commit() +check("мост без подписи → FAIL", not audit.verify_events()["ok"]) +db.execute("INSERT INTO compaction_bridge VALUES (?, ?)", (victim, r["compaction"])) +db.execute("INSERT INTO compaction_bridge VALUES ('forged', ?)", (r["compaction"],)); db.commit() +check("мост с лишней подписью → FAIL", not audit.verify_events()["ok"]) +db.execute("DELETE FROM compaction_bridge WHERE sig='forged'"); db.commit() +check("мост восстановлен → OK", audit.verify_events()["ok"]) +saved = db.execute("SELECT sig, compaction FROM compaction_bridge").fetchall() +db.execute("DROP TABLE compaction_bridge"); db.commit() +check("таблица моста снесена → FAIL", not audit.verify_events()["ok"]) +db.execute(compact.BRIDGE_DDL) +db.executemany("INSERT INTO compaction_bridge VALUES (?, ?)", saved); db.commit() +db.close() +check("мост на месте → OK", audit.verify_events()["ok"]) + +# подделка архива +bad = os.path.join(_tmp, "bad.jsonl.gz") +with gzip.open(arch, "rt", encoding="utf-8") as f: + lines = f.read().splitlines() +row = json.loads(lines[5]); row["data"] = row["data"].replace("0.5", "9.9") +lines[5] = json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":")) +with gzip.open(bad, "wt", encoding="utf-8") as f: + f.write("\n".join(lines) + "\n") +vb = compact.verify_archive(bad, r["archive_sha256"]) +check("правка строки архива → подпись и sha256 не сходятся", + not vb["ok"] and vb["first_bad"] == row["id"] and not vb["sha_ok"], vb) + +# проекции живы, повторное сжатие +es.run_projections() +fill(5000, 120) +r2 = compact.compact(older_than_hours=0, log=lambda *_: None) +v2 = audit.verify_events() +check("второе сжатие — verify OK, сжатий 2", r2["removed"] > 200 and v2["ok"] and v2["compactions"] == 2, v2) +check("проекции после сжатий", isinstance(es.run_projections(), dict)) +# бывшая голова (шум, которую второй проход не тронул) — законный кандидат +r3 = compact.compact(older_than_hours=0, log=lambda *_: None) +check("третий проход: только бывшая голова", r3["removed"] <= 1, r3) +v3 = audit.verify_events() +check("verify после третьего сжатия — OK", v3["ok"], v3) + +vac = compact.vacuum() +check("VACUUM ужал файл", vac["after"] < vac["before"], vac) +check("verify после VACUUM — OK", audit.verify_events()["ok"]) + +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_followup.py b/tests/test_followup.py new file mode 100644 index 0000000..ba3fc9a --- /dev/null +++ b/tests/test_followup.py @@ -0,0 +1,73 @@ +# -*- coding: utf-8 -*- +"""tg-reply --followup: «сначала ответ — потом работа» (аудит 2026-10-16). Hermetic. + +Запуск: ./venv/bin/python tests/test_followup.py + +Проверяет: досылка уходит в ЧАТ входящего (группа → группа, личка → +личка), не в личку по умолчанию; работает после ReplySent (обычный +tg-reply уже запрещён); повтор с тем же текстом — SKIP (ledger), другой +текст — уходит; FollowupSent в сторе; неизвестный id — ValueError; +мандат мин-уровня содержит правило. +""" +import os +import sys +import tempfile + +sys.path.insert(0, "/root/simbiot-harness") +os.environ["SIMBIOT_DB"] = os.path.join(tempfile.mkdtemp(prefix="simbiot_followup_"), "harness.db") + +from harness import commands, es, 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) + + +SENT = [] +tg_inbox.send_message = lambda chat_id, text, reply_to=None, timeout=35: ( + SENT.append((chat_id, text)) or {"ok": True}) +tg_inbox._api = lambda *a, **k: {"ok": True} + +for mid, chat in (("m-1", 7569223131), ("m-2", -1004402779881)): + es.append_event("inbox", mid, "MessageReceived", {"update_id": int(mid[2:]), "chat_id": chat, + "text": "сделай", "kind": "text"}) +tg_inbox.send_reply("m-2", "понял, делаю") +check("первый ответ ушёл в группу", SENT[-1] == (-1004402779881, "понял, делаю"), SENT) +try: + tg_inbox.send_reply("m-2", "итог") + check("повторный tg-reply запрещён", False) +except ValueError: + check("повторный tg-reply запрещён", True) + + +def followup(mid, text): + import hashlib + aid = f"followup:{mid}:{hashlib.sha1(text.encode()).hexdigest()[:10]}" + return commands.deliver_artifact(aid, kind="message", owner="test", caption=text, + send=lambda: tg_inbox.send_followup(mid, text)) + + +check("досылка после ответа — отправлена", followup("m-2", "готово: X") is True) +check("досылка — в чат входящего (группа)", SENT[-1] == (-1004402779881, "готово: X"), SENT[-1]) +n = len(SENT) +check("повтор того же итога — SKIP", followup("m-2", "готово: X") is False and len(SENT) == n) +check("другой итог — уходит", followup("m-2", "ещё: Y") is True and len(SENT) == n + 1) +check("личка → личка", followup("m-1", "итог лички") and SENT[-1][0] == 7569223131) +evs = [e for e in es.get_events(0, 1000) if e["event_type"] == "FollowupSent"] +check("FollowupSent в сторе", len(evs) == 3, len(evs)) +try: + tg_inbox.send_followup("m-404", "x") + check("неизвестный id — ValueError", False) +except ValueError: + check("неизвестный id — ValueError", True) +check("мандат мин-уровня: правило «сначала ответ»", + "СНАЧАЛА ОТВЕТ" in LEVELS["min"].mandate and "--followup" in LEVELS["min"].mandate) + +print(f"\n{'OK' if not fails else 'FAILED'}: {len(fails)} fail(s)") +sys.exit(1 if fails else 0)