t-261017110430: подача событий EventStore в контекст тика (policy events_context)

- 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/дефолт/пустой стор)
This commit is contained in:
Simbiot Harness 2026-10-17 13:03:14 +00:00
parent dd2d30c05c
commit 4e57de63bd
4 changed files with 195 additions and 3 deletions

9
cli.py
View file

@ -31,6 +31,12 @@ def main():
p_pol = sub.add_parser("policy", help="задать политику") p_pol = sub.add_parser("policy", help="задать политику")
p_pol.add_argument("--identity", default=None) p_pol.add_argument("--identity", default=None)
p_pol.add_argument("--cpu-threshold", type=int, 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 = sub.add_parser("requeue", help="вернуть failed-задачу в очередь")
p_rq.add_argument("task_id") p_rq.add_argument("task_id")
@ -282,7 +288,8 @@ def main():
commands.mark_awaiting_delivery(args.task_id, reason=args.reason) commands.mark_awaiting_delivery(args.task_id, reason=args.reason)
print(f"OK: {args.task_id} помечена — результат будет доставлен в TG при готовности") print(f"OK: {args.task_id} помечена — результат будет доставлен в TG при готовности")
elif args.cmd == "policy": 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)) print("OK:", json.dumps(pol, ensure_ascii=False))
elif args.cmd == "requeue": elif args.cmd == "requeue":
es.init_db() es.init_db()

View file

@ -445,8 +445,14 @@ def deliver_artifact(artifact_id: str, kind: str, send, owner: str = "manual",
return True return True
def set_policy(identity: str = None, cpu_alert_threshold: int = None): def set_policy(identity: str = None, cpu_alert_threshold: int = None,
"""PolicyChanged: идентичность + пороги. Мержим поверх текущей политики.""" events_context: str = None):
"""PolicyChanged: идентичность + пороги. Мержим поверх текущей политики.
events_context (t-261017110430): 'on' — в контекст тика добавляется секция
свежих содержательных событий EventStore; 'off' — секции нет. Это и есть
«переключатель подачи событий EventStore в контекст», одобренный Alex.
"""
model = es.run_projections() model = es.run_projections()
current = dict(model.get("policy") or {}) current = dict(model.get("policy") or {})
patch = {} patch = {}
@ -454,6 +460,8 @@ def set_policy(identity: str = None, cpu_alert_threshold: int = None):
patch["identity"] = identity patch["identity"] = identity
if cpu_alert_threshold is not None: if cpu_alert_threshold is not None:
patch["cpu_alert_threshold"] = cpu_alert_threshold patch["cpu_alert_threshold"] = cpu_alert_threshold
if events_context is not None:
patch["events_context"] = events_context
current.update(patch) current.update(patch)
es.append_event("harness", "policy", "PolicyChanged", current) es.append_event("harness", "policy", "PolicyChanged", current)
return current return current

View file

@ -367,6 +367,82 @@ def gate_allows_llm(level_cfg, signals) -> bool:
return any(order.get(s[0], 3) <= rank for s in signals) 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): def build_context(level_cfg, model: dict, batch=None):
"""Сборка контекста тика: мандат + память уровня + выжимки нижних + батч задач. """Сборка контекста тика: мандат + память уровня + выжимки нижних + батч задач.
@ -442,6 +518,20 @@ def build_context(level_cfg, model: dict, batch=None):
parts.append("ЭСКАЛАЦИИ (важное с нижних уровней; отработай и закрой: " parts.append("ЭСКАЛАЦИИ (важное с нижних уровней; отработай и закрой: "
"./venv/bin/python cli.py escalation-ack <id>):\n" + "\n".join(lines)) "./venv/bin/python cli.py escalation-ack <id>):\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-сообщения, ждущие ответа (в приоритете). # инбокс: входящие TG-сообщения, ждущие ответа (в приоритете).
# Голосовое с пустой расшифровкой (тишина/отказ) тоже показываем — # Голосовое с пустой расшифровкой (тишина/отказ) тоже показываем —
# минутный уровень должен знать, что голосовое пришло, но текста нет, # минутный уровень должен знать, что голосовое пришло, но текста нет,

View file

@ -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: все проверки прошли")