"""Simbiot Harness — дашборд задач: веб-страница + JSON-API. Задача t-260929164722-60dae2 [M]: страница со всеми задачами харнеса (статусы/размеры/уровни), автообновление. Плюс лестница уровней, эскалации и свежие дайджесты — состояние харнеса одним взглядом. Дизайн: stdlib-only (http.server + ThreadingHTTPServer), никаких зависимостей и лишних процессов. Читает событие-стор напрямую через run_projections() (та же проекция, что у scheduler/CLI — консистентно). Порты: 8765 (dashboard) / 8766 (healthcheck, без авторизации), SIMBIOT_DASH_PORT/SIMBIOT_DASH_ADDR/SIMBIOT_DASH_TOKEN — из env. Открытый мир — токен SIMBIOT_DASH_TOKEN (query ?token= или cookie): страница и API без токена отдают 401 (кроме порта 8766 — там только "ok", без данных; healthcheck для systemd не должен падать на auth). """ import json import os import sys import threading import time from http import cookies from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from harness import commands, es # noqa: E402 _HERE = os.path.dirname(os.path.abspath(__file__)) _INDEX = os.path.join(_HERE, "dashboard.html") _ADDR = os.environ.get("SIMBIOT_DASH_ADDR", "0.0.0.0") _PORT = int(os.environ.get("SIMBIOT_DASH_PORT", "8765")) _HEALTH_PORT = int(os.environ.get("SIMBIOT_DASH_HEALTH_PORT", "8766")) _TOKEN = os.environ.get("SIMBIOT_DASH_TOKEN", "") _REFRESH = int(os.environ.get("SIMBIOT_DASH_REFRESH", "5")) _STATUS_ORDER = {"ready": 0, "running": 1, "blocked": 2, "failed": 3, "done": 4, "closed": 5} _SIZE_COLOR = {"S": "#1d7a3e", "M": "#a3690f", "L": "#FF4F00", "XL": "#b3261e"} # ── чат Simbiot прямо в дашборде (t-261001100740-99d55e) ────────────────── # Канал один для всех UI: БД портала agent_portal_chat (/root/coop/coop.db), # вход — sender='user' (пишет портал/дашборд), ответ агента — sender='agent' # (HTTP-push из сек-рефлекса portal_channel.push). Реплаи из стора харнеса # (ReplySent) — дубль текста ответа, берём из проекции инбокса (канал «portal», # у master-хода там же ходим TG-реплаи — их chat_id не 'portal:*', не мешаем). import sqlite3 as _sqlite3 _COOP_DB = "/root/coop/coop.db" _COOP_AGENT_CACHE = {"id": None, "name": None, "ts": 0.0} _CHAT_MAX = 8000 # ровно кап coop-портала в chat/send def _coop_agent(ttl: float = 60.0): """Агент Simbiot в БД портала (id нужен для адресации agent_portal_chat).""" now = time.time() if _COOP_AGENT_CACHE["id"] and now - _COOP_AGENT_CACHE["ts"] < ttl: return _COOP_AGENT_CACHE["id"], _COOP_AGENT_CACHE["name"] try: db = _sqlite3.connect(_COOP_DB) db.row_factory = _sqlite3.Row row = db.execute( "SELECT id, agent_name FROM agents WHERE harness_type='simbiot' " "ORDER BY (agent_name='Simbiot') DESC, id ASC LIMIT 1").fetchone() db.close() except Exception: return None, None if not row: return None, None _COOP_AGENT_CACHE.update({"id": int(row["id"]), "name": row["agent_name"], "ts": now}) return _COOP_AGENT_CACHE["id"], _COOP_AGENT_CACHE["name"] def _ts_to_ms(ts) -> int: """ISO-строка стора или int-ms → int-ms (единая сортировка чата).""" if isinstance(ts, (int, float)): return int(ts) from datetime import datetime try: return int(datetime.fromisoformat(str(ts).replace("Z", "+00:00")).timestamp() * 1000) except Exception: return 0 def build_chat(limit: int = 80) -> dict: """Лента чата: БД портала (primary) + реплаи стора (полнота текста батчей). Дедуп: вход пользователя в обоих источниках под одним ключом ('m-'+msg_id) — версию стора предпочитаем (там replied/reply_text). """ agent_id, agent_name = _coop_agent() if not agent_id: return {"ok": False, "error": "no_agent", "messages": []} db = _sqlite3.connect(_COOP_DB) db.row_factory = _sqlite3.Row rows = db.execute( "SELECT msg_id, sender, text, ts, consumed FROM agent_portal_chat " "WHERE agent_id=? ORDER BY ts DESC LIMIT ?", (agent_id, limit)).fetchall() db.close() store = {} try: model = es.run_projections() for m in (model.get("inbox") or {}).values(): if isinstance(m, dict) and str(m.get("chat_id") or "").startswith("portal"): m = dict(m) m["store_id"] = "m-" + str(m.get("update_id") or "") m["ts_ms"] = _ts_to_ms(m.get("ts") or m.get("date")) store[m["store_id"]] = m except Exception: pass # стор недоступен: БД портала всё равно покажет ленту msgs = [] # анкеры: msg_id agent-реплаев вида "reply--" — покрывает # push-строку в БД портала; реплай стора дублирует её — не показываем anchors = set() for r in rows: if r["sender"] == "agent" and str(r["msg_id"]).startswith("reply-"): anchors.add("-".join(str(r["msg_id"]).split("-", 2)[2:])) for r in reversed(rows): # старые → новые # служебные команды канала (/set-tg-bot с токеном, cmd-settg-*) — не # чата: токен секрет, в ленту UI не показываем (t-261001100740) if (r["sender"] == "user" and (str(r["msg_id"]).startswith("cmd-settg-") or str(r["text"]).startswith("/set-tg-bot"))): continue if r["sender"] == "agent" and str(r["text"]).lstrip().startswith("/set-tg"): continue key = "m-" + str(r["msg_id"]) st = store.get(key) if r["sender"] == "user" else None if st is not None and st.get("replied") and st.get("reply_text") \ and str(st.get("update_id") or "") not in anchors: # реплай стора НЕ отражён в БД портала — показываем ПАРУ: # вопрос пользователя + ответ агента из стора msgs.append({ "id": str(r["msg_id"]), "from": "user", "text": st.get("text") or r["text"], "ts": int(r["ts"] or 0), "channel": "portal", "pending": False, }) msgs.append({ "id": str(r["msg_id"]) + "#reply", "from": "agent", "text": st.get("reply_text"), "ts": _ts_to_ms(st.get("ts") or r["ts"]) or int(r["ts"] or 0), "channel": "portal", "pending": False, }) continue if st is not None: text = st.get("text") or r["text"] pending = not (st.get("replied") or r["consumed"]) else: text = r["text"] pending = bool(r["sender"] == "user" and not r["consumed"] and time.time() * 1000 - int(r["ts"] or 0) < 600_000) msgs.append({ "id": str(r["msg_id"]), "from": r["sender"], "text": text, "ts": int(r["ts"] or 0), "channel": "portal", "pending": bool(pending), }) # последние реплаи стора, которых ещё нет в БД (push идёт напрямую); # их нет как строк БД — показываем только если ретрай не доехал до портала seen_db_ids = {str(r["msg_id"]) for r in rows} for st in store.values(): if (str(st.get("update_id") or "") not in seen_db_ids and st.get("replied") and st.get("reply_text")): msgs.append({"id": st["store_id"], "from": "agent", "text": st.get("reply_text"), "ts": st.get("ts_ms") or 0, "channel": "portal", "pending": False}) msgs.sort(key=lambda m: m["ts"]) # схлопнуть подряд идущие ОДИНАКОВЫЕ agent-пуши (ретраи ретраят с новым # msg_id — settg-диагностика светится как шум ×3): keep-last + счётчик collapsed = [] for m in msgs: if (m["from"] == "agent" and collapsed and collapsed[-1]["from"] == "agent" and collapsed[-1].get("text") == m.get("text") and int(m["ts"]) - int(collapsed[-1]["ts"]) < 600_000): collapsed[-1]["repeat"] = collapsed[-1].get("repeat", 1) + 1 collapsed[-1]["ts"] = m["ts"] continue collapsed.append(m) out = {"ok": True, "agent": agent_name, "messages": collapsed[-limit:]} # живой статус каналов (t-261004103718-d8112b): UI рендерит индикаторы # вместо статичного «тоже дублируется в TG, если бот подключён» try: st = channel_status() tg = _tg_stats() tgst = st.get("telegram") or {} tgst["token_loaded"] = tg.get("token_loaded") tgst["received_24h"] = tg.get("received_24h") tgst["last_msg_ts"] = tg.get("last_msg_ts") # fallback: штампа поллера нет (старый main-харнес до рестарта), но # токен загружен и свежие TG-сообщения капают в стор → канал жив. if (not tgst.get("ok") and tg.get("token_loaded") and (tg.get("last_msg_age_s") is not None) and tg["last_msg_age_s"] <= 600): tgst["ok"] = True tgst["age_s"] = tg["last_msg_age_s"] tgst["detail"] = "жив по событиям стора (штампа поллера ещё нет)" st["telegram"] = tgst st["hermes"] = _hermes_marker() out["channels"] = st except Exception as e: # статус — диагностика: чат не должен падать из-за него, но причина # должна быть видна (journalctl), иначе «пропавшие индикаторы» # отлаживаются вслепую (t-261004103718: молчаливый pass скрыл сбой) print(f"[chat] channels status error: {e}", flush=True) return out def chat_send(text: str) -> dict: """Сообщение владельца агенту: прямая запись в agent_portal_chat. HTTP-роут портала требует юзер-сессию, дашборд авторизован своим токеном — пишем напрямую в ту же таблицу (sender='user', consumed=0): сек-рефлекс харнеса поднимет её ≤20 с и мин-LLM ответит в тот же канал. """ text = (text or "").strip() if not text: return {"ok": False, "error": "empty"} agent_id, _ = _coop_agent() if not agent_id: return {"ok": False, "error": "no_agent"} mid = f"dash-{agent_id}-{int(time.time() * 1000)}-{os.urandom(3).hex()}" db = _sqlite3.connect(_COOP_DB, timeout=10) try: db.execute( "INSERT INTO agent_portal_chat (msg_id, agent_id, sender, text, ts) " "VALUES (?,?,?,?,?)", (mid, agent_id, "user", text[:_CHAT_MAX], int(time.time() * 1000))) db.commit() finally: db.close() return {"ok": True, "msg_id": mid} def _utcnow() -> str: return time.strftime("%Y-%m-%dT%H:%M:%S+00:00", time.gmtime()) def build_state() -> dict: """Срез харнеса для API: задачи, лестница, эскалации, дайджесты, спенд. Полный run_projections() на каждый запрос — безопасно: инкрементальный чекпоинт в БД, работа пропорциональна дельте событий, не истории. """ model = es.run_projections() # отчёт тика, закрывшего задачу: digest с summary_ref == result_ref digest_by_ref = {} for d in (model.get("digests") or {}).values(): if d.get("summary_ref"): digest_by_ref[d["summary_ref"]] = d tasks = [] for t in model.get("tasks", {}).values(): report = digest_by_ref.get(t.get("result_ref")) or {} tasks.append({ "id": t.get("id"), "title": t.get("title", ""), "size": t.get("size", "?"), "status": t.get("status", "?"), "level": t.get("level"), "created": t.get("created"), "finished": t.get("finished"), "error": t.get("error"), # детали для раскрытия по клику (т-260929220004) "details": (t.get("details") or "")[:2000], "promotions": t.get("promotions"), "promoted_to": t.get("promoted_to"), "session_id": t.get("session_id"), # t-261001082111: время взятия в работу (TaskClaimed) — для # пометки «в работе» и таймера elapsed на карточке running "claimed_at": t.get("claimed_at"), "report_level": report.get("level"), "report_ts": report.get("ts"), "report": (report.get("text") or "")[:1500], }) # два стабильных прохода: новые сверху внутри статуса, # статусы по порядку ready→running→failed→done→closed (авто-архив # старых фейлов, e-261003094432 — failed>24ч уезжает в closed) tasks.sort(key=lambda x: x.get("created") or "", reverse=True) tasks.sort(key=lambda x: _STATUS_ORDER.get(x["status"], 9)) ladder = [] for name, cs in sorted(model.get("cycle_state", {}).items()): ladder.append({ "level": name, "busy": bool(cs.get("busy")), "paused": bool(cs.get("paused")), "ticks": cs.get("ticks", 0), "absorbed": cs.get("absorbed", 0), "errors": cs.get("errors", 0), "last_tick": cs.get("last_tick"), "last_error": cs.get("last_error"), }) escalations = [e for e in (model.get("escalations") or {}).values() if e.get("status") == "open"] escalations.sort(key=lambda e: e.get("ts", ""), reverse=True) digests = [] for d in model.get("digests", {}).values(): digests.append({ "level": d.get("level"), "period": d.get("period"), "text": d.get("text", ""), "ts": d.get("ts"), "summary_ref": d.get("summary_ref"), }) digests.sort(key=lambda d: d.get("ts") or "", reverse=True) spend = model.get("spend", [])[-12:] counts = {} for t in tasks: key = f"{t['status']}:{t['size']}" counts[key] = counts.get(key, 0) + 1 return { "generated_at": _utcnow(), "counts": { "tasks_total": len(tasks), "tasks_by_status": { s: sum(1 for t in tasks if t["status"] == s) for s in ("ready", "running", "blocked", "failed", "done", "closed") }, "tasks_by_status_size": counts, "escalations_open": len(escalations), "events_total": es.count_events(), "digests_total": len(model.get("digests", {})), }, "ladder": ladder, "tasks": tasks, "escalations": escalations, "digests": digests[:12], "spend": spend, "meta": {"refresh_seconds": _REFRESH, "token_required": bool(_TOKEN)}, } def build_agent(level: str) -> dict: """Детали одного агента-уровня (t-261002175316-eb24cc). Read-only проекция «на лету»: состояние уровня — из run_projections() (консистентно со scheduler/CLI), история тиков/ошибок/спенда — прямыми read-only запросами к стору (проверено: 64–210мс на 643к событий). """ import re as _re import sqlite3 as _sq from datetime import datetime, timedelta, timezone lvl = (level or "").strip().lower() if not _re.fullmatch(r"[a-z_]{1,24}", lvl): return {"error": "bad level: %r" % (level or "")} model = es.run_projections() cs = (model.get("cycle_state") or {}).get(lvl) or {} def _q(sql, args=()): db = _sq.connect(es.DB_PATH, timeout=10) try: return db.execute(sql, args).fetchall() finally: db.close() now = datetime.now(timezone.utc) c24 = (now - timedelta(hours=24)).isoformat() c7 = (now - timedelta(days=7)).isoformat() # ── счётчики 24ч/7д из стора (тихи, ошибки, спенд) ── def _count(etype, cutoff): return _q( "SELECT COUNT(*) FROM events WHERE event_type=? " "AND timestamp>=? AND json_extract(data,'$.level')=?", (etype, cutoff, lvl))[0][0] ticks_24, ticks_7 = _count("CycleTicked", c24), _count("CycleTicked", c7) errors_24, errors_7 = _count("CycleError", c24), _count("CycleError", c7) absorbed_24 = _count("TickAbsorbed", c24) absorbed_7 = _count("TickAbsorbed", c7) def _tokens(cutoff): row = _q( "SELECT COUNT(*), COALESCE(SUM(json_extract(data,'$.prompt_tokens')),0), " "COALESCE(SUM(json_extract(data,'$.completion_tokens')),0) " "FROM events WHERE event_type='BudgetSpent' AND timestamp>=? " "AND json_extract(data,'$.level')=?", (cutoff, lvl))[0] return {"calls": row[0], "prompt": row[1], "completion": row[2]} tok24, tok7 = _tokens(c24), _tokens(c7) # ── активность по часам суток (24 ведра, тики уровня за 24ч) ── hourly = [{"h": h, "c": c} for h, c in _q( "SELECT strftime('%H', timestamp) h, COUNT(*) c FROM events " "WHERE event_type='CycleTicked' AND timestamp>=? " "AND json_extract(data,'$.level')=? GROUP BY h ORDER BY h", (c24, lvl))] # ── последние тики и ошибки уровня (сырьё из стора, капы жёсткие) ── recent_ticks = [{"ts": ts, "seq": aid} for aid, ts in _q( "SELECT aggregate_id, timestamp FROM events " "WHERE event_type='CycleTicked' AND json_extract(data,'$.level')=? " "ORDER BY id DESC LIMIT 40", (lvl,))] last_err_text = None recent_errors = [] for ts, err in _q( "SELECT timestamp, json_extract(data,'$.error') FROM events " "WHERE event_type='CycleError' AND timestamp>=? " "AND json_extract(data,'$.level')=? ORDER BY id DESC LIMIT 8", (c7, lvl)): err = (err or "")[:300] recent_errors.append({"ts": ts, "error": err}) if last_err_text is None: last_err_text = err # ── LLM-вызовы уровня: последние 30 по токенам (график спенда) ── spend_rows = [{"trace_id": t, "prompt": p, "completion": cc} for t, p, cc in _q( "SELECT json_extract(data,'$.trace_id'), " "json_extract(data,'$.prompt_tokens'), " "json_extract(data,'$.completion_tokens') FROM events " "WHERE event_type='BudgetSpent' AND json_extract(data,'$.level')=? " "ORDER BY id DESC LIMIT 30", (lvl,))] spend_rows.reverse() # ── задачи уровня (из проекции) + время в работе ── lvl_tasks = [] time_in_work_24 = 0.0 for t in model.get("tasks", {}).values(): if t.get("level") != lvl: continue st, cl = t.get("status"), t.get("claimed_at") dur = None if cl: try: beg = datetime.fromisoformat(cl) end = datetime.fromisoformat(t["finished"]) if t.get("finished") else now if end.tzinfo is None: end = end.replace(tzinfo=timezone.utc) if beg.tzinfo is None: beg = beg.replace(tzinfo=timezone.utc) dur = max(0.0, (end - beg).total_seconds()) if cl >= c24: time_in_work_24 += dur except (ValueError, TypeError, KeyError): dur = None lvl_tasks.append({ "id": t.get("id"), "title": (t.get("title") or "")[:120], "size": t.get("size"), "status": st, "created": t.get("created"), "claimed_at": cl, "finished": t.get("finished"), "duration_sec": dur, "error": t.get("error"), "promoted_to": t.get("promoted_to"), }) lvl_tasks.sort(key=lambda x: x.get("claimed_at") or x.get("created") or "", reverse=True) current = [{"id": t["id"], "title": t["title"], "claimed_at": t["claimed_at"], "duration_sec": t["duration_sec"]} for t in lvl_tasks if t["status"] == "running"] # ── эскалации, задевающие уровень ── esc = [{"id": e.get("id"), "severity": e.get("severity"), "reason": (e.get("reason") or "")[:200], "from_level": e.get("from_level"), "to_level": e.get("to_level"), "ts": e.get("ts")} for e in (model.get("escalations") or {}).values() if e.get("status") == "open" and (e.get("from_level") == lvl or e.get("to_level") == lvl)] esc.sort(key=lambda e: e.get("ts") or "", reverse=True) # ── дайджесты именно этого уровня ── digests = [{"period": d.get("period"), "text": (d.get("text") or "")[:600], "ts": d.get("ts"), "summary_ref": d.get("summary_ref")} for d in (model.get("digests") or {}).values() if d.get("level") == lvl] digests.sort(key=lambda d: d.get("ts") or "", reverse=True) return { "generated_at": _utcnow(), "agent": { "level": lvl, "busy": bool(cs.get("busy")), "paused": bool(cs.get("paused")), "ticks": cs.get("ticks", 0), "absorbed": cs.get("absorbed", 0), "errors": cs.get("errors", 0), "last_tick": cs.get("last_tick"), "last_error": last_err_text, "ticks_24h": ticks_24, "absorbed_24h": absorbed_24, "errors_24h": errors_24, "ticks_7d": ticks_7, "absorbed_7d": absorbed_7, "errors_7d": errors_7, "llm_24h": tok24, "llm_7d": tok7, "time_in_work_24h_sec": round(time_in_work_24), "current_tasks": current, }, "tasks": lvl_tasks[:30], "digests": digests[:10], "escalations_open": esc, "recent_ticks": recent_ticks, "recent_errors": recent_errors, "hourly": hourly, "spend_rows": spend_rows, "meta": {"token_required": bool(_TOKEN)}, } class _Handler(BaseHTTPRequestHandler): server_version = "SimbiotDash/1.2" def log_message(self, fmt, *args): # тише в journalctl: одна строка print(f"[dash] {self.address_string()} {fmt % args}", flush=True) # ── auth ── def _authorized(self) -> bool: if not _TOKEN: return True supplied = None qs = self.path.partition("?")[2] from urllib.parse import parse_qs for v in parse_qs(qs).get("token", []): supplied = v if supplied is None: c = cookies.SimpleCookie(self.headers.get("Cookie", "")) if "simbiot_token" in c: supplied = c["simbiot_token"].value return supplied == _TOKEN def _forbidden(self): body = b'{"error": "unauthorized: pass ?token=SIMBIOT_DASH_TOKEN"}\n' self.send_response(401) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("WWW-Authenticate", 'Bearer realm="simbiot-dash"') self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _json(self, obj, code=200): body = json.dumps(obj, ensure_ascii=False, default=str).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _html(self, text: str): body = text.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) # ── routes ── def do_GET(self): route = self.path.partition("?")[0] if route == "/api/state": if not self._authorized(): return self._forbidden() try: return self._json(build_state()) except Exception as e: # стор бит/занят — 503, фронт покажет retry return self._json({"error": str(e)[:500]}, code=503) if route == "/api/chat": # чат Simbiot в дашборде (t-261001100740): read — авторизуем так же if not self._authorized(): return self._forbidden() try: return self._json(build_chat()) except Exception as e: return self._json({"ok": False, "error": str(e)[:500], "messages": []}, code=503) if route == "/api/agent": # детали агента-уровня (t-261002175316-eb24cc): read-only, # ?level=sec|min|hour|day (+ токен/cookie как у /api/state) if not self._authorized(): return self._forbidden() qs = self.path.partition("?")[2] from urllib.parse import parse_qs as _pqs lvl = (_pqs(qs).get("level") or [""])[0] try: return self._json(build_agent(lvl)) except Exception as e: return self._json({"error": str(e)[:500]}, code=503) if route == "/api/health": return self._json({"ok": True, "ts": _utcnow()}) if route in ("/", "/index.html"): if not self._authorized(): return self._forbidden() try: with open(_INDEX, encoding="utf-8") as f: return self._html(f.read()) except OSError as e: return self._json({"error": f"dashboard.html not found: {e}"}, code=500) return self._json({"error": "not found"}, code=404) def do_POST(self): route = self.path.partition("?")[0] if route == "/api/chat/send": # send чата (t-261001100740): та же авторизация, что у read-API if not self._authorized(): return self._forbidden() try: n = int(self.headers.get("Content-Length") or 0) except ValueError: n = 0 if n <= 0 or n > 64_000: return self._json({"ok": False, "error": "bad length"}, code=400) try: data = json.loads(self.rfile.read(n).decode("utf-8")) except Exception: return self._json({"ok": False, "error": "bad json"}, code=400) try: return self._json(chat_send(str(data.get("text") or ""))) except Exception as e: return self._json({"ok": False, "error": str(e)[:500]}, code=503) return self._json({"error": "not found"}, code=404) def _health_server(): """Отдельный мини-порт для healthcheck: без auth, только {ok:true}.""" class H(BaseHTTPRequestHandler): def log_message(self, fmt, *args): pass def do_GET(self): body = b'{"ok":true}' self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) srv = ThreadingHTTPServer(("127.0.0.1", _HEALTH_PORT), H) srv.serve_forever() def main(): es.init_db() # идемпотентно if not os.path.exists(_INDEX): print(f"[dash] FATAL: {_INDEX} не найден", file=sys.stderr) sys.exit(1) threading.Thread(target=_health_server, daemon=True).start() print(f"[dash] http://{_ADDR}:{_PORT}/ (health 127.0.0.1:{_HEALTH_PORT}/api/health) " f"refresh={_REFRESH}s token={'on' if _TOKEN else 'off'}", flush=True) ThreadingHTTPServer((_ADDR, _PORT), _Handler).serve_forever() # ── живой статус каналов портала/TG (t-261004103718-d8112b) ──────────────── # Каждый реальный пользователь канала (сек-рефлекс portal_channel.sec_reflex, # TG-поллер tg_inbox.poll_loop) пишет штамп harness/channel_.json # после УСПЕШНОГО обмена. Дашборд только читает штампы — сам портал не дёргает # (GET-проба poll-эндпоинта дублировала бы сек-рефлекс на его айсберге), # и getUpdates не пробует (конфликт long polling с процессом поллера). _CHANNEL_MAX_AGE_MS = 10 * 60 * 1000 # штамп старше 10 мин = канал вниз _portal_probe_cache = {"ts": 0.0, "res": None} def _portal_probe() -> dict: """Fallback-проба портала (когда штампа рефлекса нет/протух). main-харнес новой версии пишет штамп при каждом сек-рефлексе; пока он не перезапущен (рестарт откладывается до тихого окна — LLM-воркеры), дашборд пробует портал сам: GET /channel/poll?since=now — read-only, consumed ставится только отдельным POST (не делаем). Кэш 60с: клиент чата тикает каждые 10с — дёргать портал чаще незачем. """ import urllib.parse import urllib.request now = time.time() if _portal_probe_cache["res"] and now - _portal_probe_cache["ts"] < 60: return _portal_probe_cache["res"] try: from harness import portal_channel as _pch cfg = _pch._cfg() if not (cfg.get("portal_url") and cfg.get("agent_token") and cfg.get("agent_name")): res = {"ok": False, "age_s": None, "ts": 0, "detail": "канал не сконфигурён (нет /etc/coop-reporter.env)"} else: url = (f"{cfg['portal_url']}/api/agent/" f"{urllib.parse.quote(cfg['agent_name'])}" f"/channel/poll?since={int(now * 1000)}") req = urllib.request.Request( url, headers={"X-Agent-Token": cfg["agent_token"]}) with urllib.request.urlopen(req, timeout=5) as r: json.loads(r.read().decode()) res = {"ok": True, "age_s": 0, "ts": int(now * 1000), "detail": "проба ok (штампа рефлекса ещё нет — старый main-харнес)"} except Exception as e: res = {"ok": False, "age_s": None, "ts": 0, "detail": f"проба не прошла: {str(e)[:80]}"} _portal_probe_cache.update(ts=now, res=res) return res # Старый харнес на Гермесе (hermes-rita, 2.28.125.152 — docs/migration-hermes.md) # не узнаётся по /etc/coop-reporter.env (документированные ключи: PORTAL_URL, # AGENT_TOKEN). Без честного признака не рисуем "старый харнес" в UI. _HERMES_MARKER = None def _hermes_marker() -> str | None: """Признак, что эти файлы запущены на старом Гермес-харнесе (для UI). Пока легитимного маркера нет — None (Simbiot-дашборд ни с чем не путаем). """ return _HERMES_MARKER def channel_status() -> dict: """Живой статус каналов чата: portal (poll портала) + telegram (inbox). Каждый канал: ok (свежий успешный обмен), age_s (возраст штампа), ts (ms последнего успеха), detail (что именно успешно). """ now_ms = int(time.time() * 1000) def _read(name: str) -> dict: path = os.path.join(_HERE, f"channel_{name}.json") try: with open(path) as f: d = json.load(f) ts = int(d.get("ts") or 0) age = now_ms - ts if age < 0 or age > _CHANNEL_MAX_AGE_MS: raise FileNotFoundError # штамп протух → считаем как отсутствующий return {"ok": bool(d.get("ok")), "age_s": round(age / 1000), "ts": ts, "detail": str(d.get("detail") or "")[:120]} except (FileNotFoundError, ValueError, TypeError): # штампа нет (файл не создан / битый / старше 10 мин). # portal: честная fallback-проба (штамп пишет main-харнес новой # версии — до его рестарта проба даёт живую правду). if name == "portal": return _portal_probe() return {"ok": False, "age_s": None, "ts": 0, "detail": "нет свежего штампа поллера"} except Exception as e: return {"ok": False, "age_s": None, "ts": 0, "detail": f"штамп не читается: {str(e)[:80]}"} return {"portal": _read("portal"), "telegram": _read("tg"), "checked_at": now_ms} def _tg_stats(days: float = 1.0) -> dict: """TG-факты из стора: токен загружен? + входящие за последние 24ч.""" from datetime import datetime, timedelta, timezone tg_token = None try: from harness import tg_inbox as _tg _tg._load_env() tg_token = bool(_tg.TG_TOKEN) except Exception: pass n24, last_ts = 0, None try: db = es._conn() cutoff = (datetime.now(timezone.utc) - timedelta(hours=int(days * 24))).isoformat() # portal-канальные MessageReceived пишет НЕ TG-поллер — исключаем: # иначе живой портал-чат рисовал бы «зелёный» telegram (найдено в # t-261004103718: portal ok, tg штампа нет — а события капают). row = db.execute( "SELECT count(*) AS n, max(timestamp) AS last FROM events " "WHERE event_type='MessageReceived' AND timestamp >= ? " "AND (data IS NULL OR data NOT LIKE '%\"chat_id\": \"portal%')", (cutoff,)).fetchone() db.close() n24 = int(row["n"] or 0) last_ts = row["last"] except Exception: pass age_s = None if last_ts: try: from datetime import datetime as _dt age_s = max(0, round(time.time() - _dt.fromisoformat(str(last_ts)).timestamp())) except Exception: age_s = None return {"token_loaded": tg_token, "received_24h": n24, "last_msg_ts": last_ts, "last_msg_age_s": age_s} if __name__ == "__main__": main()