"""Simbiot Harness — ES-спина: append-only event store + проекции.

Образец — кооп-портал (aggregate, aggregate_id, event_type, data, version).
Идемпотентные команды; тики защищены UNIQUE(level, seq).
"""
import json
import os
import sqlite3
import threading
import time
from datetime import datetime, timezone

# t-261004113351-2858c1: путь БД из paths (SIMBIOT_HOME → <home>/db/harness.db;
# без хома — legacy harness/harness.db). Точечная SIMBIOT_DB сильнее схемы.
from . import paths as _paths
DB_PATH = _paths.db_path()

# RLock обязателен: rebuild_projections держит лок и зовёт
# run_projections, который берёт его же — обычный Lock давил сам себя.
_LOCK = threading.RLock()
_read_model = None  # инкрементальный read model (глобал для инкрементальных проекций)


def _conn():
    db = sqlite3.connect(DB_PATH, timeout=30)
    db.execute("PRAGMA journal_mode=WAL")
    db.execute("PRAGMA busy_timeout=30000")
    db.row_factory = sqlite3.Row
    return db


def init_db():
    with _LOCK:
        db = _conn()
        db.executescript(
            """
            CREATE TABLE IF NOT EXISTS events (
                id           INTEGER PRIMARY KEY AUTOINCREMENT,
                aggregate    TEXT NOT NULL,
                aggregate_id TEXT NOT NULL,
                event_type   TEXT NOT NULL,
                data         TEXT NOT NULL,
                timestamp    TEXT NOT NULL,
                version      INTEGER NOT NULL
            );
            CREATE UNIQUE INDEX IF NOT EXISTS idx_events_agg
                ON events(aggregate, aggregate_id, version);
            CREATE TABLE IF NOT EXISTS projection_meta (
                name  TEXT PRIMARY KEY,
                last_event INTEGER NOT NULL,
                updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS read_model (
                name  TEXT PRIMARY KEY,
                model TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );
            """
        )
        # идемпотентная доставка (t-260930083937): ledger «доставлено раз и
        # навсегда» + локи захвата отправки; дубли в чат больше невозможны
        db.executescript(_DELIVERY_DDL)
        # идемпотентность тиков: один уровень — один seq
        db.executescript(
            """
            CREATE UNIQUE INDEX IF NOT EXISTS idx_cycle_tick
                ON events(aggregate, aggregate_id) WHERE aggregate='cycle';
            """
        )
        # aggregate_id для cycle = "level:seq", поэтому общий уникальный индекс
        # (aggregate, aggregate_id) уже даёт UNIQUE(level, seq).
        db.commit()
        db.close()
    _ensure_signature_schema()


_signature_schema_ready = False


def _ensure_signature_schema():
    """Крипто-аудит (t-260930083716): колонки подписи + ретро-генезис.

    Миграция идемпотентна, лёгкая (ALTER TABLE / UPDATE по индексу), гонки
    между процессами безопасны (IF NOT EXISTS / WHERE sig IS NULL).
    Существующие события ДО внедрения подписи — ретро-маркер генезиса
    sig='genesis:<id>': честная граница доверия (см. harness/audit.py).
    """
    global _signature_schema_ready
    if _signature_schema_ready:
        return
    from . import audit
    with _LOCK:
        db = _conn()
        cols = {r["name"] for r in db.execute("PRAGMA table_info(events)").fetchall()}
        for col in ("sig", "key_id", "prev_sig"):
            if col not in cols:
                try:
                    db.execute(f"ALTER TABLE events ADD COLUMN {col} TEXT")
                except sqlite3.OperationalError as e:
                    # гонка параллельных демонов на старте (t-261002072440):
                    # другой процесс уже добавил колонку между PRAGMA и ALTER
                    if "duplicate column" in str(e).lower():
                        pass
                    else:
                        raise
        db.commit()
        # водяной знак генезиса: раз и навсегда = max(id) на момент первой
        # миграции; Unsigned-строки ≤ знака — «до внедрения подписи» (genesis),
        # позже — legacy (старый код демонов после включения). Окно гонки в
        # миллисекунды между водяным знаком и UPDATE — наихудший случай: одна
        # лишняя строка под genesis, на целостность не влияет. INSERT OR IGNORE:
        # два демона стартуют одновременно — второй не валит bootstrap
        # (t-261002072440, UNIQUE constraint на projection_meta.name).
        row = db.execute("SELECT last_event FROM projection_meta "
                         "WHERE name='sig_genesis_upto'").fetchone()
        upto = row["last_event"] if row else None
        if upto is None:
            upto = db.execute("SELECT COALESCE(MAX(id), 0) FROM events").fetchone()[0]
            db.execute("INSERT OR IGNORE INTO projection_meta (name, last_event, updated_at) "
                       "VALUES ('sig_genesis_upto', ?, ?)",
                       (upto, datetime.now(timezone.utc).isoformat()))
        n = db.execute(
            "UPDATE events SET sig='genesis:'||id WHERE sig IS NULL AND id <= ?",
            (upto,)).rowcount
        db.commit()
        db.close()
    if n:
        print(f"[audit] retro-genesis: {n} событий до внедрения подписи", flush=True)
    _signature_schema_ready = True


# ────────────────── ИДЕМПОТЕНТНАЯ ДОСТАВКА (t-260930083937) ──────────────────
# Дубли в чат приходили из гонки push (тик) vs pull (delivery-демон): оба
# проверяли delivery_status из stale-модели, пока TG-отправка (до 35с) была
# в полёте → по два DeliverySent на задачу (разница 23–34с в сторе).
# Фикс — два атомарных примитива поверх SQLite:
#   ledger  — факт «артефакт доставлен» (INSERT OR IGNORE = только первая);
#   claim   — захват права отправки на TTL (одно атомарное выражение — без
#             BEGIN IMMEDIATE гонок между процессами/потоками не бывает).

_DELIVERY_DDL = """
CREATE TABLE IF NOT EXISTS delivery_ledger (
    artifact_id TEXT PRIMARY KEY,
    kind        TEXT NOT NULL,
    ts          TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS delivery_locks (
    artifact_id TEXT PRIMARY KEY,
    owner       TEXT,
    claimed_at  REAL NOT NULL,
    expires_at  REAL NOT NULL
);
"""

_delivery_tables_ready = False


def _ensure_delivery_tables(db):
    """Ленивое создание таблиц доставки: живой демон обновляется без
    ручной миграции (init_db вызывается не всеми путями)."""
    global _delivery_tables_ready
    if _delivery_tables_ready:
        return
    db.executescript(_DELIVERY_DDL)
    db.commit()
    _delivery_tables_ready = True


def delivery_recorded(artifact_id: str) -> bool:
    """Артефакт уже доставлен? (метка delivered по артефакт-идентификатору)."""
    db = _conn()
    try:
        _ensure_delivery_tables(db)
        row = db.execute("SELECT 1 FROM delivery_ledger WHERE artifact_id=?",
                          (artifact_id,)).fetchone()
        return row is not None
    finally:
        db.close()


def record_delivery(artifact_id: str, kind: str = "task") -> bool:
    """Зафиксировать доставку. True — только для ПЕРВОЙ (INSERT OR IGNORE):
    повторная пометка не проходит, дубль доставки невозможен даже при
    параллельном срабатывании нескольких уровней (мин/час/демон)."""
    with _LOCK:
        db = _conn()
        try:
            _ensure_delivery_tables(db)
            cur = db.execute(
                "INSERT OR IGNORE INTO delivery_ledger (artifact_id, kind, ts) VALUES (?,?,?)",
                (artifact_id, kind, datetime.now(timezone.utc).isoformat()))
            db.commit()
            return cur.rowcount > 0
        finally:
            db.close()


def forget_delivery(artifact_id: str) -> bool:
    """Сброс факта доставки: requeue доработки — новый результат должен
    уйти в чат заново (старый ≠ новый)."""
    with _LOCK:
        db = _conn()
        try:
            _ensure_delivery_tables(db)
            cur = db.execute("DELETE FROM delivery_ledger WHERE artifact_id=?",
                             (artifact_id,))
            db.execute("DELETE FROM delivery_locks WHERE artifact_id=?",
                       (artifact_id,))
            db.commit()
            return cur.rowcount > 0
        finally:
            db.close()


def delivery_claim(artifact_id: str, owner: str = "", ttl_sec: int = 120) -> bool:
    """Атомарный захват права отправки (пометка «доставляю» ДО отправки).

    Одно выражение INSERT…SELECT…WHERE NOT EXISTS + ON CONFLICT — атомарно
    на уровне SQLite, гонки между потоками/процессами исключены:
      - лока нет / истёк  → захват, True;
      - лок жив (кто-то уже отправляет) → False, отправитель пропускает.
    TTL > таймаута TG-отправки (35с текст / 90с файл) с запасом: живой
    sender не истечёт раньше конца отправки. Успех НЕ освобождает лок —
    он истекает сам: ранний release после успешной отправки открыл бы
    окно для параллельного дубля (второй уже начал читать модель).
    """
    with _LOCK:
        db = _conn()
        try:
            _ensure_delivery_tables(db)
            now = time.time()
            # WHERE NOT EXISTS отфильтровывает insert при живом локе →
            # rowcount=0 (чужой захват); при истёкшем/отсутствующем локе
            # INSERT срабатывает, ON CONFLICT обновляет существующую строку.
            cur = db.execute(
                """
                INSERT INTO delivery_locks (artifact_id, owner, claimed_at, expires_at)
                SELECT ?, ?, ?, ?
                WHERE NOT EXISTS (
                    SELECT 1 FROM delivery_locks
                    WHERE artifact_id = ? AND expires_at > ?
                )
                ON CONFLICT(artifact_id) DO UPDATE SET
                    owner = excluded.owner,
                    claimed_at = excluded.claimed_at,
                    expires_at = excluded.expires_at
                """,
                (artifact_id, owner, now, now + ttl_sec, artifact_id, now))
            db.commit()
            return cur.rowcount > 0
        finally:
            db.close()


def delivery_release(artifact_id: str):
    """Освободить захват при сбое отправки: демон досыплет на след. проходе.
    Вызывается ТОЛЬКО при неудаче — после успешной отправки лок истекает сам
    (ранний release = окно для дубля)."""
    with _LOCK:
        db = _conn()
        try:
            _ensure_delivery_tables(db)
            db.execute("DELETE FROM delivery_locks WHERE artifact_id=?", (artifact_id,))
            db.commit()
        finally:
            db.close()


def append_event(aggregate: str, aggregate_id: str, event_type: str, data: dict):
    """Append-only запись события. Возвращает (event_id, version) или None при дубликате.

    Крипто-аудит (t-260930083716): каждое событие подписывается приватным
    ключом харнеса (ed25519) в момент записи; prev_sig связывает с предыдущим
    (тампер-цепочка). Отказ подписи НЕ валит запись — событие уходит без sig
    (верификатор пометит legacy), харнес продолжает работать.
    """
    with _LOCK:
        _ensure_signature_schema()  # лениво: колонки подписи/genesis должны
        # существовать ДО INSERT (append_event зовут и без init_db —
        # t-260930083716: без этого трейсы молча терялись на копиях БД)
        db = _conn()
        # Атомарное окно «прочитать prev → вставить» (t-261002072440):
        # без транзакции два демона читали prev ДО вставки друг друга и
        # линковались «сквозь» чужую строку — link-check рвался на ровном
        # месте, проекции падали. BEGIN IMMEDIATE сериализует писателей
        # между процессами (WAL, busy_timeout 30с в _conn).
        try:
            db.execute("BEGIN IMMEDIATE")
        except sqlite3.OperationalError:
            pass  # экстремальная нагрузка — пишем как раньше (audit допускает prev-скип)
        row = db.execute(
            "SELECT MAX(version) AS v FROM events WHERE aggregate=? AND aggregate_id=?",
            (aggregate, aggregate_id),
        ).fetchone()
        version = (row["v"] or 0) + 1
        ts = datetime.now(timezone.utc).isoformat()
        data_str = json.dumps(data, ensure_ascii=False)
        sig_f = {}
        try:
            from . import audit as _audit
            prev = db.execute(
                "SELECT sig FROM events ORDER BY id DESC LIMIT 1").fetchone()
            sig_f = _audit.sign_event_row(aggregate, aggregate_id, event_type,
                                          data_str, ts, version,
                                          prev["sig"] if prev else None)
            # t-261008084600 (rogue-key guard): стор с чужими подписями не подпишем
            # своим ключом. Инциденты 06.10/08.10: cli из /tmp-копии репо наследовал
            # SIMBIOT_HOME прод-стора, не находил рядом свой .audit-ключ, генерил
            # НОВЫЙ и вписывал его подпись в прод-стор → run_projections падал
            # «стор скомпрометирован» каждую секунду. Правило: если стор уже подписан,
            # key_id нового ключа обязан совпадать с key_id существующих событий;
            # иначе пишем без подписи (legacy-граница, проекции живы).
            if sig_f.get("key_id"):
                dom = db.execute(
                    "SELECT key_id FROM events WHERE key_id IS NOT NULL "
                    "ORDER BY id DESC LIMIT 1").fetchone()
                if dom and dom["key_id"] and dom["key_id"] != sig_f["key_id"]:
                    sig_f = {}
  # чужой ключ → без подписи; верификатор пометит legacy
        except Exception:
            sig_f = {}  # без подписи (legacy) — аудит отразит, харнес не падает
        try:
            cur = db.execute(
                "INSERT INTO events (aggregate,aggregate_id,event_type,data,timestamp,version,"
                "sig,key_id,prev_sig) VALUES (?,?,?,?,?,?,?,?,?)",
                (aggregate, aggregate_id, event_type, data_str, ts, version,
                 sig_f.get("sig"), sig_f.get("key_id"), sig_f.get("prev_sig")),
            )
            db.commit()
            eid = cur.lastrowid
        except sqlite3.IntegrityError:
            db.rollback()
            db.close()
            return None  # дубль (тик с тем же level:seq) — идемпотентно пропускаем
        db.close()
        return (eid, version)


def append_tick(level: str, seq: int):
    """CycleTicked идемпотентно: aggregate_id = 'level:seq' → UNIQUE даёт защиту."""
    return append_event("cycle", f"{level}:{seq}", "CycleTicked", {"level": level, "seq": seq})


def get_events(after_id: int = 0, limit: int = 1000):
    db = _conn()
    rows = db.execute(
        "SELECT * FROM events WHERE id > ? ORDER BY id LIMIT ?", (after_id, limit)
    ).fetchall()
    db.close()
    return [dict(r) for r in rows]


def count_events() -> int:
    db = _conn()
    n = db.execute("SELECT COUNT(*) AS c FROM events").fetchone()["c"]
    db.close()
    return n


# ────────────────────────── ПРОЕКЦИИ ──────────────────────────

def _default_model():
    return {
        "tasks": {},        # task_id → {title, size, status, level, session_id, result_ref, ...}
        "digests": {},      # (level, period) → {summary_ref, text}
        "escalations": {},  # esc_id → {from_level, to_level, reason, status: open|acked}
        "cycle_state": {},  # level → {busy, current_task, last_tick}
        "metrics": [],      # [{level, ticks, absorbed, llm_ticks, tokens, cost}]
        "agents": {},       # level → {model, calls}
    }


def _load_meta(db, name):
    row = db.execute("SELECT last_event FROM projection_meta WHERE name=?", (name,)).fetchone()
    return row["last_event"] if row else 0


def _save_meta(db, name, last_event):
    db.execute(
        "INSERT INTO projection_meta (name, last_event, updated_at) VALUES (?,?,?) "
        "ON CONFLICT(name) DO UPDATE SET last_event=excluded.last_event, updated_at=excluded.updated_at",
        (name, last_event, datetime.now(timezone.utc).isoformat()),
    )


def _apply_event(model: dict, ev: dict):
    etype, data = ev["event_type"], json.loads(ev["data"])
    aid = ev["aggregate_id"]

    if not isinstance(data, dict):
        data = {}
    if etype == "TaskSubmitted":
        model["tasks"][aid] = {
            "id": aid, "title": data.get("title", ""), "size": data.get("size", "M"),
            "status": "ready", "level": None, "session_id": None,
            "result_ref": None, "created": ev["timestamp"],
            # details — постановка задачи; исполнитель видит её в контексте
            # тика (build_context). Без этого часовой уровень выполнял задачу
            # по одному заголовку и терял принципы/нюансы постановки.
            "details": (data.get("details") or "")[:2000],
        }
    elif etype == "TaskClaimed":
        t = model["tasks"].get(aid)
        if t:
            t.update(status="running", level=data.get("level"), session_id=data.get("session_id"),
                     claimed_at=ev["timestamp"])
    elif etype == "TaskCompleted":
        t = model["tasks"].get(aid)
        if t:
            t.update(status="done", result_ref=data.get("result_ref"), finished=ev["timestamp"])
    elif etype == "TaskBlocked":
        # t-261004122019: задача есть, но не ready — ждёт уточнений/внешнего
        # события. scheduler берёт только ready, так что blocked выпадает из
        # батчей; обратно — requeue (blocked → ready).
        t = model["tasks"].get(aid)
        if t:
            t.update(status="blocked", block_reason=(data.get("reason") or "")[:300],
                     blocked_by=data.get("by") or "manual",
                     blocked_at=ev["timestamp"])
    elif etype == "TaskFailed":
        t = model["tasks"].get(aid)
        if t:
            t.update(status="failed", error=data.get("error"), finished=ev["timestamp"])
    elif etype == "TaskClosed":
        # Авто-схлопывание старых фейлов (просьба Alex 03.10, e-261003094432):
        # failed-задачи старше FAILED_STALE_HOURS закрываются демон-архиватором
        # в статус closed — отдельная «архивная» группа после done. Обратимо:
        # requeue принимает closed → ready (осознанная доработка).
        t = model["tasks"].get(aid)
        if t and t.get("status") in ("failed",):
            t.update(status="closed", finished=ev["timestamp"],
                     closed_by=data.get("by") or "archive")
    elif etype == "TaskPromoted":
        t = model["tasks"].get(aid)
        if t:
            if data.get("from") and data.get("to") and data["from"] != data["to"]:
                t.update(size=data["to"])
            t.update(status="ready", level=None, session_id=None,
                     claimed_at=None,
                     promoted_to=data.get("to"), promotions=(t.get("promotions") or 0) + 1)
            # requeue доработки (from==to): старая доставка не считается —
            # новый результат должен уйти в чат заново (t-260930083937).
            # Без сброса флаг висел бы 'sent' и резал досылку навсегда.
            t.pop("delivery_status", None)
            t.pop("delivery_sent_at", None)
            # requeue сбрасывает факт доставки: работа будет сделана заново,
            # новый результат должен снова уйти в чат (старый ≠ новый)
            if t.get("await_delivery"):
                t.update(delivery_status="await")
    elif etype == "TaskAwaitDelivery":
        # автодоставка (t-260929205927): задачу заказали из TG — результат
        # должен уйти в чат сам, без повторного запроса. Флаг на задаче;
        # scheduler-демон ловит TaskCompleted и рассылает DeliverySent.
        t = model["tasks"].get(aid)
        if t:
            t.update(await_delivery=True, delivery_status="await",
                     delivery_reply_to=data.get("reply_to"),
                     delivery_reason=(data.get("reason") or "")[:300])
    elif etype == "DeliverySent":
        # доставлено в чат (результат готовой задачи); best-effort флаг:
        # доставка идемпотентна — повторная не отправит (sent раз и навсегда)
        t = model["tasks"].get(aid)
        if t:
            t.update(delivery_status="sent", delivery_sent_at=ev["timestamp"],
                     delivered_text=(data.get("text") or "")[:2000],
                     delivery_tg_ok=bool(data.get("tg_ok")))
        else:
            model.setdefault("deliveries", {})[aid] = {
                "task_id": data.get("task_id"), "ts": ev["timestamp"],
                "tg_ok": bool(data.get("tg_ok")),
            }
    elif etype == "DeliverySkipped":
        # идемпотентная доставка (t-260930083937): артефакт уже доставлен /
        # захвачен другим отправителем — повтор НЕ отправлен, пропуск залогирован.
        # Сырьё (по одному событию на каждую проверку) наверх не идёт —
        # часовой уровень видит сводку skips на задаче.
        t = model["tasks"].get(aid)
        if t:
            t["delivery_skips"] = (t.get("delivery_skips") or 0) + 1
            t["last_skip_reason"] = (data.get("reason") or "")[:200]
    elif etype == "DigestWritten":
        key = f"{data['level']}:{data['period']}"
        model["digests"][key] = {
            "level": data["level"], "period": data["period"],
            "summary_ref": data.get("summary_ref"),
            "text": data.get("text", ""),
            "ts": ev["timestamp"],
        }
        # кап дайджестов: держим по 40 последних на уровень —
        # «наверх идут только дайджесты» не значит «все дайджесты навсегда»
        if len(model["digests"]) > 400:
            keep = sorted(model["digests"].items(),
                          key=lambda kv: kv[1].get("ts", ""), reverse=True)[:400]
            model["digests"] = dict(keep)
    elif etype == "EscalationRaised":
        model.setdefault("escalations", {})[aid] = {
            **data, "status": "open", "ts": ev["timestamp"],
        }
    elif etype == "EscalationAcknowledged":
        e = model.setdefault("escalations", {}).get(aid)
        if e:
            e["status"] = "acked"
            e["acked_at"] = ev["timestamp"]
            e["ack_level"] = data.get("level")
        else:
            model["escalations"][aid] = {
                "status": "acked", "ack_level": data.get("level"), "ts": ev["timestamp"],
            }
    elif etype == "CycleTicked":
        level = data.get("level", "?")
        cs = model["cycle_state"].setdefault(level, {"busy": False, "current_task": None, "ticks": 0})
        cs["ticks"] = cs.get("ticks", 0) + 1
        cs["last_tick"] = ev["timestamp"]
    elif etype == "TickAbsorbed":
        level = data.get("level", "?")
        cs = model["cycle_state"].setdefault(level, {"busy": False, "current_task": None, "ticks": 0})
        cs["absorbed"] = cs.get("absorbed", 0) + 1
    elif etype == "LevelPaused":
        model["cycle_state"].setdefault(data["level"], {})["paused"] = True
    elif etype == "LevelResumed":
        model["cycle_state"].setdefault(data["level"], {})["paused"] = False
    elif etype == "PolicyChanged":
        model.setdefault("policy", {}).update(data)
    elif etype == "AumMeasured":
        # северная звезда (t-260929161138): последняя метрика AUM/burn —
        # в policy (build_context инжектит всем уровням), история — кап 120 снапшотов
        model.setdefault("policy", {}).update(data.get("policy", {}))
        hist = model.setdefault("aum_history", [])
        hist.append({k: v for k, v in data.items() if k != "policy"})
        if len(hist) > 120:
            model["aum_history"] = hist[-120:]
    elif etype == "BudgetSpent":
        spend = model.setdefault("spend", [])
        spend.append(data)
        if len(spend) > 500:  # кап: не растить модель линейно
            model["spend"] = spend[-500:]
    elif etype == "MessageReceived":
        model.setdefault("inbox", {})[aid] = {
            **data, "replied": False, "ignored": False,
            "seen": False, "ts": ev["timestamp"],
        }
        inbox = model["inbox"]
        if len(inbox) > 100:  # кап инбокса: последние 100 входящих
            keep = sorted(inbox.items(), key=lambda kv: kv[1].get("ts", ""))[-100:]
            model["inbox"] = dict(keep)
    elif etype == "ReplySent":
        m = model.setdefault("inbox", {}).get(aid)
        if m:
            m["replied"] = True
            m["reply_text"] = data.get("text", "")
    elif etype == "MessageSeen":
        # индикация взятия в работу (e-261002202857): флаг seen → reaction 👀
        # ставится ровно один раз, повторные маркеры сообщений не найдут
        m = model.setdefault("inbox", {}).get(aid)
        if m:
            m["seen"] = True
            if isinstance(data, dict) and data.get("emoji"):
                m["seen_emoji"] = data["emoji"]
    elif etype == "MessageTranscribed":
        # голосовое → текст: минутный уровень видит его как обычное
        # входящее (kind=voice, text=транскрипт) и отвечает по смыслу.
        # reset=True — сброс прошлой расшифровки (cli tg-transcribe --force).
        if data.get("reset"):
            m = model.setdefault("inbox", {}).get(aid)
            if m:
                m["transcribed"] = False
                m.pop("transcript", None)
                m.pop("transcribe_reason", None)
                # текст возвращаем к исходнику (пустой у голосовых),
                # чтобы не отвечали на устаревший транскрипт
                m["text"] = (m.get("original_text") or "")[:4000]
            return
        m = model.setdefault("inbox", {}).get(aid)
        if m:
            if not m.get("original_text_recorded"):
                m["original_text"] = m.get("text") or ""
                m["original_text_recorded"] = True
            m["transcribed"] = True
            m["transcript"] = data.get("transcript") or ""
            m["engine"] = data.get("engine")
            if data.get("reason"):
                m["transcribe_reason"] = data.get("reason")
            else:
                m.pop("transcribe_reason", None)
            if data.get("transcript"):
                m["text"] = data["transcript"][:4000]  # расшифровка = текст сообщения
            elif not m.get("original_text"):
                # пустая расшифровка затирает старый транскрипт из text
                m["text"] = ""
        else:
            model["inbox"][aid] = {
                "transcribed": True, "transcript": data.get("transcript") or "",
                "engine": data.get("engine"), "replied": False, "ignored": False,
                "ts": ev["timestamp"],
            }
    elif etype == "MessageIgnored":
        # канал портала (t-261001123121): доставка реплая помечена ReplySent'ом
        # (portal ReplySent — признак push'а) — отдельный маркер не нужен
        m = model.setdefault("inbox", {}).get(aid)
        if m:
            m["ignored"] = True
        else:
            model["inbox"][aid] = {"ignored": True, "reason": data.get("reason"), "ts": ev["timestamp"]}
    elif etype == "CycleError":
        # счётчик ошибок по уровням — материал для эскалаций (демон/CLI)
        cs = model.setdefault("cycle_state", {}).setdefault(
            data.get("level", "?"), {"busy": False, "current_task": None, "ticks": 0})
        cs["errors"] = cs.get("errors", 0) + 1
        cs["last_error"] = (data.get("error") or "")[:300]
    elif etype == "InitiativeSent":
        # инициативные реплики (t-260930073419): харнес сам пишет в чат с идеями
        # по кооперативу/AUM. Проекция — только последний слот (для анти-спама:
        # не чаще раза в MIN_INTERVAL_H и только в окне активности).
        snt = model.setdefault("initiative", {})
        snt["last_slot"] = data.get("slot")
        snt["last_ts"] = ev["timestamp"]
        snt.setdefault("sent", []).append({
            "slot": data.get("slot"), "topic": data.get("topic"),
            "text": (data.get("text") or "")[:600], "ts": ev["timestamp"],
            "tg_ok": bool(data.get("tg_ok")),
        })
        if len(snt["sent"]) > 50:  # кап истории реплик
            snt["sent"] = snt["sent"][-50:]
    # trace-события (InvocationStarted/ToolCalled/ToolResult) не материализуем — они для дебага


def _load_model(db):
    """Персистентный read model: без него каждый новый процесс
    стартовал с пустой моделью при уже продвинутом last_event —
    и терял TaskSubmitted/PolicyChanged навсегда."""
    row = db.execute("SELECT model FROM read_model WHERE name='main'").fetchone()
    if row:
        try:
            return json.loads(row["model"])
        except (ValueError, TypeError):
            pass
    return None


def _save_model(db, model):
    db.execute(
        "INSERT INTO read_model (name, model, updated_at) VALUES ('main', ?, ?) "
        "ON CONFLICT(name) DO UPDATE SET model=excluded.model, updated_at=excluded.updated_at",
        (json.dumps(model, ensure_ascii=False, default=str), datetime.now(timezone.utc).isoformat()),
    )


def run_projections():
    """Инкрементальная проекция с персистентным read model.

    Гонка устранена: модель ВСЕГДА стартует с персистентного снапшота
    (read_model-таблица), затем применяется дельта событий. In-memory
    кэш между процессами не доверяем — два процесса с общим чекпоинтом
    затирали друг друга (демон не видел задачу, поданную из CLI).
    """
    global _read_model
    with _LOCK:
        db = _conn()
        last = _load_meta(db, "main")
        # ВСЕГДА стартуем со снапшота на диске — устраняет межпроцессную гонку
        _read_model = _load_model(db) or _default_model()
        # снапшот мог быть записан другим процессом на большем id, чем meta
        snap_last = int(_read_model.get("_last_event", 0))
        if snap_last > last:
            last = snap_last
        rows = db.execute("SELECT * FROM events WHERE id > ? ORDER BY id", (last,)).fetchall()
        if not rows:
            db.close()
            return _read_model
        # верификация хвоста (t-260930083716): дельта событий должна быть
        # подписана и не ломать цепочку; подделка = RuntimeError (проекции
        # останавливаются — подделанное событие не попадает в read model)
        try:
            from . import audit as _audit
            rep = _audit.verify_rows([dict(r) for r in rows])
            if not rep["ok"]:
                raise RuntimeError(
                    f"audit: подпись события id={rep['first_bad']} не прошла — "
                    f"стор скомпрометирован, проекция остановлена")
        except RuntimeError:
            raise
        except Exception:
            pass  # сам аудит не валит проекции (нет ключа/либ) — полный аудит: cli verify
        for r in rows:
            _apply_event(_read_model, dict(r))
        new_last = rows[-1]["id"]
        _read_model["_last_event"] = new_last
        _save_meta(db, "main", new_last)
        _save_model(db, _read_model)
        db.commit()
        db.close()
        return _read_model


def rebuild_projections():
    """Полный replay с нуля (проекции одноразовые, стор — истина)."""
    global _read_model
    with _LOCK:
        db = _conn()
        db.executescript(
            """
            CREATE TABLE IF NOT EXISTS projection_meta (
                name TEXT PRIMARY KEY, last_event INTEGER NOT NULL, updated_at TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS read_model (
                name TEXT PRIMARY KEY, model TEXT NOT NULL, updated_at TEXT NOT NULL
            );
            DELETE FROM projection_meta;
            DELETE FROM read_model;
            """
        )
        db.commit()
        db.close()
        _read_model = None
        return run_projections()