From 4e57de63bd1aee3078d116678785966e595946dd Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Sat, 17 Oct 2026 13:03:14 +0000 Subject: [PATCH] =?UTF-8?q?t-261017110430:=20=D0=BF=D0=BE=D0=B4=D0=B0?= =?UTF-8?q?=D1=87=D0=B0=20=D1=81=D0=BE=D0=B1=D1=8B=D1=82=D0=B8=D0=B9=20Eve?= =?UTF-8?q?ntStore=20=D0=B2=20=D0=BA=D0=BE=D0=BD=D1=82=D0=B5=D0=BA=D1=81?= =?UTF-8?q?=D1=82=20=D1=82=D0=B8=D0=BA=D0=B0=20(policy=20events=5Fcontext)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - scheduler: _events_digest() — содержательные события стора за ~90 мин (Message/Reply/Tasks/Escalations/AUM/MCP), шум (CycleTicked/Tool*/) отсечён; секция «СОБЫТИЯ ХАРНЕСА» в build_context по policy.events_context=on - commands/cli: policy --events-context on|off (переключатель, одобрен Alex) - tests/test_events_context.py: 9 hermetic-проверок (on/off/дефолт/пустой стор) --- cli.py | 9 +++- harness/commands.py | 12 ++++- harness/scheduler.py | 90 ++++++++++++++++++++++++++++++++++++ tests/test_events_context.py | 87 ++++++++++++++++++++++++++++++++++ 4 files changed, 195 insertions(+), 3 deletions(-) create mode 100644 tests/test_events_context.py diff --git a/cli.py b/cli.py index e43a36d..8ac0d00 100644 --- a/cli.py +++ b/cli.py @@ -31,6 +31,12 @@ def main(): p_pol = sub.add_parser("policy", help="задать политику") p_pol.add_argument("--identity", default=None) p_pol.add_argument("--cpu-threshold", type=int, default=None) + # t-261017110430: вкл/выкл секцию свежих событий EventStore в контексте + # тика (values: off | on; on = короткая выжимка последних содержательных + # событий стора — что происходило в харнесе между тиками) + p_pol.add_argument("--events-context", default=None, + choices=["off", "on"], + help="events_context: on = секция «СОБЫТИЯ ХАРНЕСА (EventStore)» в контексте тика") p_rq = sub.add_parser("requeue", help="вернуть failed-задачу в очередь") p_rq.add_argument("task_id") @@ -282,7 +288,8 @@ def main(): commands.mark_awaiting_delivery(args.task_id, reason=args.reason) print(f"OK: {args.task_id} помечена — результат будет доставлен в TG при готовности") elif args.cmd == "policy": - pol = commands.set_policy(identity=args.identity, cpu_alert_threshold=args.cpu_threshold) + pol = commands.set_policy(identity=args.identity, cpu_alert_threshold=args.cpu_threshold, + events_context=args.events_context) print("OK:", json.dumps(pol, ensure_ascii=False)) elif args.cmd == "requeue": es.init_db() diff --git a/harness/commands.py b/harness/commands.py index 79d6143..bbc48d0 100644 --- a/harness/commands.py +++ b/harness/commands.py @@ -445,8 +445,14 @@ def deliver_artifact(artifact_id: str, kind: str, send, owner: str = "manual", return True -def set_policy(identity: str = None, cpu_alert_threshold: int = None): - """PolicyChanged: идентичность + пороги. Мержим поверх текущей политики.""" +def set_policy(identity: str = None, cpu_alert_threshold: int = None, + events_context: str = None): + """PolicyChanged: идентичность + пороги. Мержим поверх текущей политики. + + events_context (t-261017110430): 'on' — в контекст тика добавляется секция + свежих содержательных событий EventStore; 'off' — секции нет. Это и есть + «переключатель подачи событий EventStore в контекст», одобренный Alex. + """ model = es.run_projections() current = dict(model.get("policy") or {}) patch = {} @@ -454,6 +460,8 @@ def set_policy(identity: str = None, cpu_alert_threshold: int = None): patch["identity"] = identity if cpu_alert_threshold is not None: patch["cpu_alert_threshold"] = cpu_alert_threshold + if events_context is not None: + patch["events_context"] = events_context current.update(patch) es.append_event("harness", "policy", "PolicyChanged", current) return current diff --git a/harness/scheduler.py b/harness/scheduler.py index 1980409..091b301 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -367,6 +367,82 @@ def gate_allows_llm(level_cfg, signals) -> bool: return any(order.get(s[0], 3) <= rank for s in signals) +# t-261017110430: какие события EventStore показывать в контексте тика. +# Шум (Tool*/BudgetSpent/CycleTicked/DigestWritten/GateSkipped/ReflexObserved) +# не берём — это 90% стора и оно не про «что произошло», а про механику. +_EVENTS_CTX_TYPES = ( + "MessageReceived", "ReplySent", + "TaskSubmitted", "TaskCompleted", "TaskClaimed", "TaskPromoted", + "EscalationRaised", "EscalationAcknowledged", + "AumMeasured", "PortalMcpCalled", "DeliverySent", + "ArtifactDelivered", "DecisionMade", "CoreLabel", +) +# человекочитаемая выжимка события: одной строкой, без сырого JSON +_EVENTS_CTX_LABEL = { + "MessageReceived": "входящее", + "ReplySent": "ответ в чат", + "TaskSubmitted": "новая задача", + "TaskCompleted": "задача закрыта", + "TaskClaimed": "задача взята", + "TaskPromoted": "перевод задачи", + "EscalationRaised": "эскалация", + "EscalationAcknowledged": "эскалация закрыта", + "AumMeasured": "замер AUM", + "PortalMcpCalled": "портал-вызов", + "DeliverySent": "доставка в чат", + "ArtifactDelivered": "артефакт доставлен", + "DecisionMade": "решение", + "CoreLabel": "ядро: метка", +} + + +def _events_digest(limit: int = 12, lookback_min: int = 90) -> list: + """Свежие содержательные события стора — компактные строки для контекста. + + Возвращает СТАРЫЕ→НОВЫЕ (хронология читается естественно). Пустой список — + событий подходящего типа не было. Ошибки глотаются на месте вызова + (build_context): контекст не должен падать из-за выжимки. + """ + db = es._conn() + try: + since = (_now() - timedelta(minutes=lookback_min)).isoformat() + rows = db.execute( + "SELECT timestamp, aggregate, aggregate_id, event_type, data " + "FROM events WHERE event_type IN (%s) AND timestamp > ? " + "ORDER BY id DESC LIMIT ?" % ",".join("?" * len(_EVENTS_CTX_TYPES)), + (*_EVENTS_CTX_TYPES, since, limit), + ).fetchall() + finally: + db.close() + out = [] + for r in reversed(rows): + try: + d = json.loads(r["data"] or "{}") + except Exception: + d = {} + et = r["event_type"] + lbl = _EVENTS_CTX_LABEL.get(et, et) + aid = r["aggregate_id"] or "" + # приоритет полей: у каждого типа — своё «о чём это». + # DecisionMade с purpose=addressing — механика адресации групп, не + # контент: показываем «адресация → выбор», а не пустоту. + what = (d.get("answer") or d.get("text") or d.get("title") + or d.get("reason") or d.get("note") or d.get("tool") + or d.get("result") or "") + what = str(what).replace("\n", " ").strip() + if et == "DecisionMade" and d.get("purpose") == "addressing": + continue # механика адресации групп — не контент, пропускаем + if et == "DecisionMade": + what = what or str(d.get("choice") or "?") + if et == "TaskClaimed" and not what: + what = f"уровнем {d.get('level', '?')}" + if et == "AumMeasured": + what = f"AUM={d.get('aum', '?')} у.е., burn 7д={d.get('burn_7d', '?')}" + ts = (r["timestamp"] or "")[11:16] # HH:MM + out.append(f"{ts} {lbl} [{aid[:28]}]: {what[:140]}") + return out + + def build_context(level_cfg, model: dict, batch=None): """Сборка контекста тика: мандат + память уровня + выжимки нижних + батч задач. @@ -442,6 +518,20 @@ def build_context(level_cfg, model: dict, batch=None): parts.append("ЭСКАЛАЦИИ (важное с нижних уровней; отработай и закрой: " "./venv/bin/python cli.py escalation-ack ):\n" + "\n".join(lines)) + # t-261017110430: свежие события EventStore — «что происходило в харнесе + # между тиками». Вкл/выкл — policy.events_context (cli.py policy + # --events-context on|off); по умолчанию off (сырой шум в тик не подаём). + if policy and str(policy.get("events_context", "off")).lower() == "on": + try: + ev_lines = _events_digest() + if ev_lines: + parts.append( + "СОБЫТИЯ ХАРНЕСА (EventStore, последние ~90 мин, старые→новые; " + "это уже произошло — не отвечай на них, используй как фон):\n" + + "\n".join(ev_lines)) + except Exception as e: + _log(f"events-context: выжимка событий не собрана ({str(e)[:120]})") + # инбокс: входящие TG-сообщения, ждущие ответа (в приоритете). # Голосовое с пустой расшифровкой (тишина/отказ) тоже показываем — # минутный уровень должен знать, что голосовое пришло, но текста нет, diff --git a/tests/test_events_context.py b/tests/test_events_context.py new file mode 100644 index 0000000..bc22256 --- /dev/null +++ b/tests/test_events_context.py @@ -0,0 +1,87 @@ +# -*- coding: utf-8 -*- +"""t-261017110430: подача событий EventStore в контекст тика (hermetic, без LLM). + +Запуск: ./venv/bin/python tests/test_events_context.py + +Проверяет: policy.events_context=on — в контексте мин/час/день появляется +секция «СОБЫТИЯ ХАРНЕСА» с содержательными событиями (MessageReceived, +TaskClaimed, …) и БЕЗ механического шума (CycleTicked/ToolCalled/…); +policy off (и дефолт) — секции нет. Выжимка не валит контекст при пустом +сторе. +""" +import os +import sys + +sys.path.insert(0, "/root/simbiot-harness") + +# изоляция стора (как test_gate.py): тест не пишет в боевой стор +import tempfile as _iso_tf # noqa: E402 +os.environ["SIMBIOT_DB"] = os.path.join( + _iso_tf.mkdtemp(prefix="simbiot_test_evt_"), "harness.db") + +from harness import es, scheduler # noqa: E402 +from harness.levels import LEVELS # 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) + + +es.init_db() +es.append_event("tg", "m-1", "MessageReceived", {"text": "привет, как дела?"}) +es.append_event("tg", "m-1", "ReplySent", {"text": "привет! всё ровно"}) +es.append_event("harness", "t-1", "TaskClaimed", {"level": "min"}) +# механический шум — его в контексте быть не должно +es.append_event("cycle", "x", "CycleTicked", {"level": "min"}) +es.append_event("trace", "y", "ToolCalled", {"tool": "shell"}) + + +def ctx(policy): + model = {"tasks": {}, "inbox": {}, "escalations": {}, "digests": {}, + "policy": policy if policy is not None else {}} + return scheduler.build_context(LEVELS["min"], model, batch=[]) + + +on = ctx({"events_context": "on"}) +check("on: секция СОБЫТИЯ ХАРНЕСА в контексте", "СОБЫТИЯ ХАРНЕСА" in on) +check("on: входящее с текстом", + "входящее [m-1]: привет, как дела?" in on) +check("on: ответ в чат", "ответ в чат [m-1]: привет! всё ровно" in on) +check("on: задача взята с уровнем", "задача взята [t-1]: уровнем min" in on) +check("on: шум CycleTicked/ToolCalled отсечён", + "CycleTicked" not in on and "ToolCalled" not in on) + +off = ctx({"events_context": "off"}) +check("off: секции нет", "СОБЫТИЯ ХАРНЕСА" not in off) + +default = ctx(None) +check("дефолт (нет policy): секции нет", "СОБЫТИЯ ХАРНЕСА" not in default) + +# пустой стор (lookback 90 мин): не падает и не добавляет секцию. +# es.DB_PATH кешируется на импорте — смена SIMBIOT_DB после импорта не +# действует, поэтому патчим модульный глобал напрямую (с возвратом). +_db_backup, es.DB_PATH = es.DB_PATH, os.path.join( + _iso_tf.mkdtemp(prefix="simbiot_test_evt2_"), "harness.db") +try: + es.init_db() + empty = ctx({"events_context": "on"}) + check("пустой стор: контекст собран, секции нет", + isinstance(empty, str) and "СОБЫТИЯ ХАРНЕСА" not in empty) +finally: + es.DB_PATH = _db_backup + +# часовой уровень: секция там тоже (тот же build_context) +on_hour = scheduler.build_context( + LEVELS["hour"], {"tasks": {}, "inbox": {}, "escalations": {}, "digests": {}, + "policy": {"events_context": "on"}}, batch=[]) +check("hour: секция в контексте", "СОБЫТИЯ ХАРНЕСА" in on_hour) + +print() +if fails: + print("FAILED:", len(fails), fails) + sys.exit(1) +print("OK: все проверки прошли")