simbiot-harness/harness/compact.py

"""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. Архив <home>/archive/events-<first>-<last>-<ts>.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:<id> в мосте (под тем же
    # подписанным хешем) — «граница здесь была, унесена сжатием».
    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}