d8a7c89ef9
- es: TaskClosed → статус closed (терминальный, обратимый через requeue; closed_by manual|archive) - commands: close_task (ручной, только failed, идемпотентен), archive_stale_failed(max_age_hours=24, now=None) — демон-архиватор - scheduler._delivery_and_orphans_loop: каждый проход (15с) архивирует failed>24ч - requeue принимает closed → ready (доработка из архива) - cli: simbiot close <id> [-r reason] - dashboard: порядок ready→running→failed→done→архив, счётчик «архив» в статусбаре, схема закрытых фейлов под одной строкой-счётчиком (тумблер показывает полностью), pill.closed - тест tests/test_close_stale_failed.py: 15 проверок (закрытие, порог через симуляцию +25ч, done не трогается, идемпотентность, полный цикл closed→ready→failed→closed)
688 lines
No EOL
34 KiB
Python
688 lines
No EOL
34 KiB
Python
"""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
|
||
|
||
DB_PATH = os.environ.get("SIMBIOT_DB", os.path.join(os.path.dirname(__file__), "harness.db"))
|
||
|
||
# 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)
|
||
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)
|
||
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 == "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() |