From 2e31249992ab73550616c4cd09a69642a4a24234 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 8 Oct 2026 10:32:05 +0300 Subject: [PATCH] =?UTF-8?q?feat(scheduler):=20fetch=20origin=20+=20=D0=B4?= =?UTF-8?q?=D0=B5=D1=82=D0=B5=D0=BA=D1=82=D0=BE=D1=80=20=D0=B4=D1=80=D0=B5?= =?UTF-8?q?=D0=B9=D1=84=D0=B0=20=D1=80=D0=B5=D0=BF=D0=BE=20=D0=B2=20=D0=B4?= =?UTF-8?q?=D0=BD=D0=B5=D0=B2=D0=BD=D0=BE=D0=BC=20=D1=82=D0=B8=D0=BA=D0=B5?= =?UTF-8?q?=20(t-261007082307-2e2294)=20=E2=80=94=20=5Frepo=5Fsync=5Fcheck?= =?UTF-8?q?:=20git=20fetch=20+=20behind/ahead=20=D0=B7=D0=B0=D0=BC=D0=B5?= =?UTF-8?q?=D1=80=20RepoSynced=20(1/=D0=B4=D0=B5=D0=BD=D1=8C,=20=D0=B4?= =?UTF-8?q?=D0=B5=D0=B4=D1=83=D0=BF),=20=D1=8D=D1=81=D0=BA=D0=B0=D0=BB?= =?UTF-8?q?=D0=B0=D1=86=D0=B8=D1=8F=20=D0=BF=D1=80=D0=B8=20behind>0=20?= =?UTF-8?q?=D0=B4=D0=B2=D0=B0=20=D0=B4=D0=BD=D1=8F=20=D0=9F=D0=9E=D0=94?= =?UTF-8?q?=D0=A0=D0=AF=D0=94=20(=D1=81=D1=87=D1=91=D1=82=D1=87=D0=B8?= =?UTF-8?q?=D0=BA=20=D0=B8=D0=B7=20=D0=B8=D1=81=D1=82=D0=BE=D1=80=D0=B8?= =?UTF-8?q?=D0=B8=20=D1=81=D1=82=D0=BE=D1=80=D0=B0=20=E2=80=94=20=D1=80?= =?UTF-8?q?=D0=B5=D1=81=D1=82=D0=B0=D1=80=D1=82=20=D0=B4=D0=B5=D0=BC=D0=BE?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=BD=D0=B5=20=D1=81=D0=B1=D1=80=D0=B0=D1=81?= =?UTF-8?q?=D1=8B=D0=B2=D0=B0=D0=B5=D1=82);=20fetch-fail=20=D1=84=D0=B8?= =?UTF-8?q?=D0=BA=D1=81=D0=B8=D1=80=D1=83=D0=B5=D1=82=D1=81=D1=8F=20=D1=87?= =?UTF-8?q?=D0=B5=D1=81=D1=82=D0=BD=D0=BE=20(behind=3Dunknown),=20=D1=82?= =?UTF-8?q?=D0=B8=D0=BA=20=D0=BD=D0=B5=20=D0=B2=D0=B0=D0=BB=D0=B8=D1=82;?= =?UTF-8?q?=20=D0=BF=D0=BE=D1=80=D0=BE=D0=B3=20SIMBIOT=5FREPO=5FSYNC=5FMAX?= =?UTF-8?q?=5FAGE=5FDAYS=20(default=202);=20=D1=82=D0=B5=D1=81=D1=82=2017?= =?UTF-8?q?=20=D0=BF=D1=80=D0=BE=D0=B2=D0=B5=D1=80=D0=BE=D0=BA=20hermetic?= =?UTF-8?q?=20(git-=D0=BF=D0=B5=D1=81=D0=BE=D1=87=D0=BD=D0=B8=D1=86=D0=B0)?= =?UTF-8?q?,=20=D0=B2=D0=B5=D1=81=D1=8C=20=D1=81=D1=8E=D0=B8=D1=82=2020/20?= =?UTF-8?q?=20=D0=B7=D0=B5=D0=BB=D1=91=D0=BD=D1=8B=D0=B9;=20=D0=B6=D0=B8?= =?UTF-8?q?=D0=B2=D0=BE=D0=B9=20=D0=BF=D1=80=D0=BE=D0=B3=D0=BE=D0=BD:=20be?= =?UTF-8?q?hind=3D0=20ahead=3D13,=20baseline=20=D0=B7=D0=B0=D0=BF=D0=B8?= =?UTF-8?q?=D1=81=D0=B0=D0=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- harness/scheduler.py | 124 ++++++++++++++++++++++++++++++- tests/test_repo_sync.py | 160 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 283 insertions(+), 1 deletion(-) create mode 100644 tests/test_repo_sync.py diff --git a/harness/scheduler.py b/harness/scheduler.py index 3acdc33..353e542 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -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() diff --git a/tests/test_repo_sync.py b/tests/test_repo_sync.py new file mode 100644 index 0000000..53de27b --- /dev/null +++ b/tests/test_repo_sync.py @@ -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)