328 lines
No EOL
14 KiB
Python
328 lines
No EOL
14 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
|
||
);
|
||
"""
|
||
)
|
||
# идемпотентность тиков: один уровень — один 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()
|
||
|
||
|
||
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)
|
||
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 == "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]
|
||
# 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() |