feat(scheduler): fetch origin + детектор дрейфа репо в дневном тике (t-261007082307-2e2294) — _repo_sync_check: git fetch + behind/ahead замер RepoSynced (1/день, дедуп), эскалация при behind>0 два дня ПОДРЯД (счётчик из истории стора — рестарт демона не сбрасывает); fetch-fail фиксируется честно (behind=unknown), тик не валит; порог SIMBIOT_REPO_SYNC_MAX_AGE_DAYS (default 2); тест 17 проверок hermetic (git-песочница), весь сюит 20/20 зелёный; живой прогон: behind=0 ahead=13, baseline записан
This commit is contained in:
root
parent
79e12548a9
commit
2e31249992
2 changed files with 283 additions and 1 deletions
|
|
@ -9,6 +9,8 @@
|
|||
"""
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
|
|
@ -16,6 +18,7 @@ from datetime import datetime, timezone, timedelta
|
|||
|
||||
from . import es
|
||||
from .levels import LEVELS
|
||||
from . import paths as _paths
|
||||
from . import runtime
|
||||
from . import commands
|
||||
from .commands import _ORDER
|
||||
|
|
@ -602,6 +605,119 @@ def run_level_tick(level_cfg, seq: int, model: dict):
|
|||
keepalive.join(timeout=3)
|
||||
|
||||
|
||||
# ── репо-гигиена: fetch origin в дневном ритме (t-261007082307-2e2294) ──────
|
||||
# Инцидент: локальная линия 2 дня не видела 32 коммита origin/master (найдено
|
||||
# случайно в день 10-07). Причина: git fetch нигде не выполнялся регулярно —
|
||||
# расхождение не имело детектора. Фикс: детерминированный шаг в дневном тике
|
||||
# (не в память LLM) — fetch + behind-статус + эскалация при behind>0 после
|
||||
# двух дней подряд (стейт из истории RepoSynced в сторе, не из файла —
|
||||
# рестарт демона счётчик дней не сбрасывает).
|
||||
_REPO_SYNC_MAX_AGE_DAYS = int(os.environ.get(
|
||||
"SIMBIOT_REPO_SYNC_MAX_AGE_DAYS", "2"))
|
||||
|
||||
|
||||
def _repo_sync_check(level_cfg):
|
||||
"""git fetch origin + замер behind: пишем RepoSynced, эскалируем дрейф.
|
||||
|
||||
Вызывается из дневного тика (раз в сутки). Один замер на день: повторный
|
||||
вызов в тот же день НЕ плодит записи (append_event версионирует, а не
|
||||
дедуплицирует — проверяем стор перед записью). Ошибка git (нет сети/
|
||||
кредов) фиксируется в событии без падения тика: behind=unknown честнее
|
||||
выдуманного 0, но не блокирует работу.
|
||||
"""
|
||||
today = _now().strftime("%Y-%m-%d")
|
||||
day_key = f"repo-sync:{today}"
|
||||
root = _paths.REPO_ROOT
|
||||
try:
|
||||
subprocess.run(["git", "fetch", "origin"], cwd=root,
|
||||
capture_output=True, timeout=180, check=True)
|
||||
fetch_ok, err = True, None
|
||||
except Exception as e:
|
||||
fetch_ok, err = False, str(e)[:300]
|
||||
behind = ahead = None
|
||||
if fetch_ok:
|
||||
try:
|
||||
def _cnt(args):
|
||||
r = subprocess.run(["git", "rev-list", "--count"] + args,
|
||||
cwd=root, capture_output=True, timeout=30,
|
||||
check=True)
|
||||
return int(r.stdout.decode().strip() or 0)
|
||||
behind = _cnt(["master..origin/master"])
|
||||
ahead = _cnt(["origin/master..master"])
|
||||
except Exception as e:
|
||||
err = str(e)[:300]
|
||||
if not _synced_today(day_key):
|
||||
es.append_event("harness", day_key, "RepoSynced", {
|
||||
"ts": _now().isoformat(), "fetch_ok": fetch_ok,
|
||||
"behind": behind, "ahead": ahead, "error": err,
|
||||
})
|
||||
if behind is not None and behind > 0:
|
||||
_maybe_escalate_repo_drift(level_cfg, today, behind, ahead)
|
||||
return {"fetch_ok": fetch_ok, "behind": behind, "ahead": ahead}
|
||||
|
||||
|
||||
def _synced_today(day_key: str) -> bool:
|
||||
"""RepoSynced на эту дату уже записан? (анти-дубль в рамках дня)"""
|
||||
try:
|
||||
db = sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", timeout=10, uri=True)
|
||||
row = db.execute(
|
||||
"SELECT 1 FROM events WHERE event_type='RepoSynced' "
|
||||
"AND aggregate_id=? LIMIT 1", (day_key,)).fetchone()
|
||||
db.close()
|
||||
return row is not None
|
||||
except Exception:
|
||||
return False # стор недоступен → пишем (лучше дубль, чем потеря)
|
||||
|
||||
|
||||
def _consecutive_days_behind(today: str, today_behind) -> int:
|
||||
"""Сколько дней ПОДРЯД (заканчивая сегодня) behind был > 0.
|
||||
|
||||
Сегодня берётся из ЖИВОГО замера (событие дня могло быть записано
|
||||
раньше с другим значением — замер один на день). Прошлые дни — из
|
||||
истории RepoSynced в сторе: рестарт демона счётчик не сбрасывает.
|
||||
"""
|
||||
if not today_behind or int(today_behind) <= 0:
|
||||
return 0
|
||||
try:
|
||||
db = sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", timeout=10, uri=True)
|
||||
rows = db.execute(
|
||||
"SELECT DISTINCT substr(timestamp,1,10) d FROM events "
|
||||
"WHERE event_type='RepoSynced' "
|
||||
"AND COALESCE(json_extract(data,'$.behind'),0) > 0 "
|
||||
"AND substr(timestamp,1,10) < ? ORDER BY d DESC",
|
||||
(today,)).fetchall()
|
||||
db.close()
|
||||
except Exception:
|
||||
return 1 # подсчёт прошлых дней не удался — считаем только сегодня
|
||||
run, expect = 1, (
|
||||
datetime.strptime(today, "%Y-%m-%d") - timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
for (d,) in rows:
|
||||
if d == expect:
|
||||
run += 1
|
||||
expect = (datetime.strptime(expect, "%Y-%m-%d")
|
||||
- timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
else:
|
||||
break
|
||||
return run
|
||||
|
||||
|
||||
def _maybe_escalate_repo_drift(level_cfg, today: str, behind: int, ahead):
|
||||
"""Эскалация при behind>0 N дней ПОДРЯД (порог SIMBIOT_REPO_SYNC_MAX_AGE_DAYS)."""
|
||||
days = _consecutive_days_behind(today, behind)
|
||||
if days < _REPO_SYNC_MAX_AGE_DAYS:
|
||||
return
|
||||
import secrets as _secrets
|
||||
esc_id = f"e-{_now().strftime('%y%m%d%H%M%S')}-{_secrets.token_hex(3)}"
|
||||
es.append_event("harness", esc_id, "EscalationRaised", {
|
||||
"id": esc_id, "from_level": level_cfg.name,
|
||||
"to_level": commands.level_above(level_cfg.name) or "day",
|
||||
"reason": f"репо отстаёт от origin/master: behind={behind} уже "
|
||||
f"{days} дн. подряд — само не разрешится, нужен merge/pull",
|
||||
"context": f"RepoSynced repo-sync:{today}: behind={behind}, ahead={ahead}",
|
||||
"severity": "warn", "task_id": None, "source": "repo-sync",
|
||||
})
|
||||
|
||||
|
||||
def process_level(level_cfg, model: dict):
|
||||
"""Тик одного уровня: занят? → absorb; иначе работа."""
|
||||
st = _runners.setdefault(level_cfg.name, {"busy": False, "task": None})
|
||||
|
|
@ -617,8 +733,14 @@ def process_level(level_cfg, model: dict):
|
|||
return "already" # тик этого такта уже был (идемпотентность)
|
||||
|
||||
# северная звезда (t-260929161138): дневной уровень раз в сутки пишёт
|
||||
# AumMeasured снапшот (AUM + burn 9router) + обновляет aum_target в policy
|
||||
# AumMeasured снапшот (AUM + burn 9router) + обновляет aum_target в policy;
|
||||
# репо-гигиена (t-261007082307-2e2294): fetch origin + детектор дрейфа —
|
||||
# расхождение с origin больше не может висеть незамеченным 2+ дня
|
||||
if level_cfg.name == "day":
|
||||
try:
|
||||
_repo_sync_check(level_cfg)
|
||||
except Exception:
|
||||
pass # репо-синк — best effort, не валить дневной тик
|
||||
try:
|
||||
from . import aum as _aum
|
||||
snap = _aum.measure()
|
||||
|
|
|
|||
160
tests/test_repo_sync.py
Normal file
160
tests/test_repo_sync.py
Normal file
|
|
@ -0,0 +1,160 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
"""t-261007082307-2e2294: репо-гигиена — fetch origin в дневном тике (hermetic).
|
||||
|
||||
Запуск: ./venv/bin/python tests/test_repo_sync.py
|
||||
|
||||
Проверяет: _repo_sync_check пишет RepoSynced с behind/ahead (fetch в git-
|
||||
песочнице), один замер на день (повтор не плодит записи), при behind>0
|
||||
первый день эскалации НЕТ, второй день подряд — EscalationRaised (счётчик
|
||||
дней из истории стора, рестарт демона его не сбрасывает), ошибка fetch
|
||||
не валит шаг (behind=unknown честно фиксируется), env-порог
|
||||
SIMBIOT_REPO_SYNC_MAX_AGE_DAYS управляет чувствительностью.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from datetime import timedelta
|
||||
|
||||
# ── изоляция: своя БД в /tmp, git-песочница вместо реального репо ──
|
||||
_DST = "/tmp/simbiot-test-reposync.db"
|
||||
for p in (_DST, _DST + "-wal", _DST + "-shm"):
|
||||
if os.path.exists(p):
|
||||
os.unlink(p)
|
||||
os.environ["SIMBIOT_DB"] = _DST
|
||||
os.environ["SIMBIOT_REPO_SYNC_MAX_AGE_DAYS"] = "2"
|
||||
sys.path.insert(0, "/root/simbiot-harness")
|
||||
|
||||
from harness import es, paths # noqa: E402
|
||||
from harness.levels import LEVELS # noqa: E402
|
||||
from harness import scheduler # noqa: E402
|
||||
|
||||
fails = []
|
||||
|
||||
|
||||
def check(name, cond, extra=""):
|
||||
print(("PASS" if cond else "FAIL"), name, ("| " + extra) if not cond else "")
|
||||
if not cond:
|
||||
fails.append(name)
|
||||
|
||||
|
||||
def _git(cwd, *args):
|
||||
r = subprocess.run(["git"] + list(args), cwd=cwd,
|
||||
capture_output=True, timeout=30, check=True)
|
||||
return r.stdout.decode().strip()
|
||||
|
||||
|
||||
def synced_events(limit=200):
|
||||
out = []
|
||||
for e in es.get_events(limit=limit):
|
||||
if e["event_type"] == "RepoSynced":
|
||||
e["data"] = json.loads(e["data"])
|
||||
out.append(e)
|
||||
return out
|
||||
|
||||
|
||||
def escalations(limit=500):
|
||||
return [e for e in es.get_events(limit=limit)
|
||||
if e["event_type"] == "EscalationRaised"]
|
||||
|
||||
|
||||
def make_sandbox():
|
||||
"""bare origin + клон: init запушен, локально ещё 1 незапушенный коммит."""
|
||||
base = tempfile.mkdtemp(prefix="simbiot-reposync-")
|
||||
origin = os.path.join(base, "origin.git")
|
||||
_git(base, "init", "--bare", "-b", "master", origin)
|
||||
clone = os.path.join(base, "clone")
|
||||
_git(base, "clone", origin, clone)
|
||||
for d in (clone,):
|
||||
_git(d, "config", "user.email", "test@simbiot.local")
|
||||
_git(d, "config", "user.name", "simbiot-test")
|
||||
_git(clone, "commit", "--allow-empty", "-m", "init")
|
||||
_git(clone, "push", "origin", "master")
|
||||
_git(clone, "commit", "--allow-empty", "-m", "local-only") # ahead=1
|
||||
return clone, origin, base
|
||||
|
||||
|
||||
es.init_db()
|
||||
clone, origin, base = make_sandbox()
|
||||
_real_root = paths.REPO_ROOT
|
||||
paths.REPO_ROOT = clone # песочница вместо реального репо
|
||||
try:
|
||||
# ── 1. happy path: fetch ok, behind=0/ahead=1, RepoSynced записан ──
|
||||
res = scheduler._repo_sync_check(LEVELS["day"])
|
||||
check("fetch ok", res.get("fetch_ok") is True)
|
||||
check("behind=0", res.get("behind") == 0)
|
||||
check("ahead=1 (незапушенный коммит виден)", res.get("ahead") == 1)
|
||||
evs = synced_events(50)
|
||||
check("RepoSynced записан", len(evs) == 1, f"{len(evs)}")
|
||||
today = evs[0]["aggregate_id"].split(":", 1)[1]
|
||||
check("aggregate_id = repo-sync:<сегодня>",
|
||||
evs[0]["aggregate_id"] == "repo-sync:" +
|
||||
scheduler._now().strftime("%Y-%m-%d"))
|
||||
|
||||
# ── 2. повтор в тот же день — замер тот же, записей не больше ──
|
||||
again = scheduler._repo_sync_check(LEVELS["day"])
|
||||
check("повтор: те же числа", again.get("behind") == 0 and again.get("ahead") == 1)
|
||||
check("повтор в тот же день не плодит записи", len(synced_events(50)) == 1)
|
||||
|
||||
# ── 3. upstream ушёл вперёд: behind=1, эскалации ещё НЕТ (1-й день) ──
|
||||
work = os.path.join(base, "worktree")
|
||||
_git(base, "clone", origin, work)
|
||||
_git(work, "config", "user.email", "test@simbiot.local")
|
||||
_git(work, "config", "user.name", "simbiot-test")
|
||||
_git(work, "commit", "--allow-empty", "-m", "upstream change")
|
||||
_git(work, "push", "origin", "master")
|
||||
res3 = scheduler._repo_sync_check(LEVELS["day"])
|
||||
check("behind=1 увиден", res3.get("behind") == 1)
|
||||
check("эскалации НЕТ в первый день", len(escalations()) == 0)
|
||||
|
||||
# ── 4. вчера тоже был behind>0 → сегодня 2-й подряд → эскалация ──
|
||||
yday = (scheduler._now() - timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
eid, _v = es.append_event("harness", f"repo-sync:{yday}", "RepoSynced", {
|
||||
"ts": yday + "T06:00:00+00:00", "fetch_ok": True,
|
||||
"behind": 1, "ahead": 1})
|
||||
db = es._conn()
|
||||
db.execute("UPDATE events SET timestamp=? WHERE id=?",
|
||||
(yday + "T06:00:00.000000+00:00", eid))
|
||||
db.commit()
|
||||
db.close()
|
||||
res4 = scheduler._repo_sync_check(LEVELS["day"])
|
||||
check("behind=1 стабилен", res4.get("behind") == 1)
|
||||
escs = escalations()
|
||||
check("эскалация ПОДНЯТА на 2-й день подряд", len(escs) == 1, f"{len(escs)}")
|
||||
if escs:
|
||||
reason = json.loads(escs[-1]["data"]).get("reason", "")
|
||||
check("причина упоминает behind и 2 дня",
|
||||
"behind=1" in reason and "2 дн" in reason, reason[:120])
|
||||
finally:
|
||||
paths.REPO_ROOT = _real_root
|
||||
shutil.rmtree(base, ignore_errors=True)
|
||||
|
||||
# ── 5. ошибка fetch не валит шаг (свежая БД: дедуп дня не мешает) ──
|
||||
_DST2 = "/tmp/simbiot-test-reposync2.db"
|
||||
for p in (_DST2, _DST2 + "-wal", _DST2 + "-shm"):
|
||||
if os.path.exists(p):
|
||||
os.unlink(p)
|
||||
es.DB_PATH = _DST2
|
||||
es._signature_schema_ready = False # латч на процесс: новая БД требует миграции
|
||||
es.init_db()
|
||||
paths.REPO_ROOT = "/nonexistent-repo-dir"
|
||||
try:
|
||||
res5 = scheduler._repo_sync_check(LEVELS["day"])
|
||||
check("fetch-fail: fetch_ok=False, не упал", res5.get("fetch_ok") is False)
|
||||
evs5 = synced_events(50)
|
||||
check("fetch-fail зафиксирован событием", len(evs5) == 1)
|
||||
if evs5:
|
||||
check("behind=unknown (None), не выдуманный 0",
|
||||
evs5[0]["data"].get("behind") is None)
|
||||
check("ошибка сохранена", bool(evs5[0]["data"].get("error")))
|
||||
finally:
|
||||
paths.REPO_ROOT = _real_root
|
||||
|
||||
# ── 6. env-порог читается при импорте ──
|
||||
check("порог = 2 из env", scheduler._REPO_SYNC_MAX_AGE_DAYS == 2)
|
||||
|
||||
print()
|
||||
print("ИТОГ:", "ВСЕ ОК" if not fails else f"ПРОВАЛЫ: {fails}")
|
||||
sys.exit(1 if fails else 0)
|
||||
Loading…
Reference in a new issue