1bc33d5678
Инцидент: Alex получал презентацию трижды (m-516173239/241). Две причины: 1) гонка push (тик) vs pull (delivery-демон): оба проходили проверку delivery_status != 'sent', пока TG-отправка (до 35с) была в полёте; 2) ad-hoc curl sendDocument из мин-тиков вообще без метки delivered. Фикс — delivery_ledger (INSERT OR IGNORE: факт доставки раз и навсегда) + delivery_locks (атомарный claim ДО отправки, TTL 120с; одно SQLite- выражение = нет гонок между потоками и процессами). deliver_result / deliver_artifact: ledger → claim → отправка → запись; сбой → release, демон досыплет. Requeue сбрасывает ledger (доработка доставляется заново). Новое: cli.py tg-deliver-file (идемпотентная отправка файлов, --force), tg-notify -k <ключ>, tg_inbox.send_document (multipart на чистом urllib), deliver_pending pre-filter по ledger, мандаты уровней: прямые curl/api.telegram отправки запрещены. DeliverySkipped логирует пропуски. Тесты: tests/test_delivery_idempotency.py — 49 проверок (гонка потоков, межпроцессный claim, сбои+ретраи, requeue, CLI). Регрессия: 6/6 тестов ОК.
559 lines
No EOL
26 KiB
Python
559 lines
No EOL
26 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()
|
||
|
||
|
||
# ────────────────── ИДЕМПОТЕНТНАЯ ДОСТАВКА (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 при дубликате."""
|
||
with _LOCK:
|
||
db = _conn()
|
||
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()
|
||
try:
|
||
cur = db.execute(
|
||
"INSERT INTO events (aggregate,aggregate_id,event_type,data,timestamp,version) VALUES (?,?,?,?,?,?)",
|
||
(aggregate, aggregate_id, event_type,
|
||
json.dumps(data, ensure_ascii=False), ts, version),
|
||
)
|
||
db.commit()
|
||
eid = cur.lastrowid
|
||
except sqlite3.IntegrityError:
|
||
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"))
|
||
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 == "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,
|
||
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 == "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,
|
||
"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 == "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":
|
||
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
|
||
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() |