сжатие стора с подписью + «сначала ответ, потом работа» на мин-уровне
harness/compact.py + cli.py compact [--dry-run|--vacuum|--verify-archives]: шум сек-уровня (CycleTicked sec, ReflexObserved без алерта/команд/каналов) старше часа → архив jsonl.gz (0600, строки целиком со sig) → мост цепочки (подписи унесённых строк, на которые ссылаются остающиеся, + маркеры boundary:<id> для 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 <id> --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 <noreply@anthropic.com>
This commit is contained in:
Simbiot Harness
parent
b257c32558
commit
2b266caab5
9 changed files with 637 additions and 7 deletions
62
cli.py
62
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:
|
||||
|
|
|
|||
|
|
@ -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:<id> — ретро-маркер строк ДО внедрения подписи (доверенная
|
||||
|
|
@ -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:<id> из моста сжатия: перед строкой была неподписанная
|
||||
# граница (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:
|
||||
|
|
|
|||
247
harness/compact.py
Normal file
247
harness/compact.py
Normal file
|
|
@ -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. Архив <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}
|
||||
|
|
@ -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']} не прошла — "
|
||||
|
|
|
|||
|
|
@ -73,6 +73,11 @@ LEVELS = {
|
|||
"ОДНО сообщение не должно остаться без ответа на следующий тик. "
|
||||
"Тон: кратко, по-русски, тёпло; "
|
||||
"можно выполнить просьбу инструментами и ответить результатом. "
|
||||
"СНАЧАЛА ОТВЕТ, ПОТОМ РАБОТА: если до ответа нужна работа дольше "
|
||||
"~30 с (поиск, ssh, правки, несколько шагов) — ПЕРВЫМ действием тика "
|
||||
"короткий tg-reply («понял, делаю: …, отпишусь»), затем работа, итог — "
|
||||
"cli.py tg-reply <id> --followup \"итог\" (тот же чат, повтор — SKIP). "
|
||||
"Простой ответ — сразу одним tg-reply, без досылки. "
|
||||
"Быстрые мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. "
|
||||
"НЕ создавай задачи из служебных/пустых сообщений («list», «тест», "
|
||||
"«привет», односложные реакции) — они засоряют доску клонами "
|
||||
|
|
|
|||
|
|
@ -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:<id>:<хеш текста>).
|
||||
"""
|
||||
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-отправка
|
||||
закрывает несколько сообщений одного разговора.
|
||||
|
|
|
|||
19
scripts/compact-store.sh
Executable file
19
scripts/compact-store.sh
Executable file
|
|
@ -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
|
||||
148
tests/test_compact.py
Normal file
148
tests/test_compact.py
Normal file
|
|
@ -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)
|
||||
73
tests/test_followup.py
Normal file
73
tests/test_followup.py
Normal file
|
|
@ -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)
|
||||
Loading…
Reference in a new issue