780 lines
36 KiB
Python
780 lines
36 KiB
Python
"""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-<ms>-<update_id>" — покрывает
|
||
# 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_<portal|tg>.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()
|
||
|