253 lines
No EOL
13 KiB
Python
253 lines
No EOL
13 KiB
Python
"""Simbiot Harness — без-LLM команды бота (/лестница, /статус, /задачи, /помощь).
|
||
|
||
Секундный уровень (рефлекс, без LLM) отвечает на команды мгновенно —
|
||
минутный LLM-контекст на них не тратится (t-260929210755):
|
||
answer_pending() в _reflex_tick находит в инбоксе входящие-команды,
|
||
хендлер собирает текст из read model, send_reply отправляет.
|
||
ReplySent закрывает входящее — мин-уровень команду больше не видит
|
||
(build_context их дополнительно фильтрует на случай гонки).
|
||
Не отправилось (TG недоступен) — следующий сек-тик попробует снова:
|
||
ReplySent пишется только после попытки отправки, команда остаётся «без ответа».
|
||
|
||
Регистрация в чате: setMyCommands — кириллические имена; если Bot API
|
||
кириллицу не принимает, фолбэк на транслит. Вызывается стартом tg-поллера
|
||
и CLI `cli.py tg-commands`. Алиасы парсер понимает независимо от того,
|
||
какие имена зарегистрированы.
|
||
"""
|
||
import html
|
||
import json
|
||
import time
|
||
from datetime import datetime, timezone
|
||
|
||
from . import es
|
||
|
||
# cooldown после неудачной отправки: сек-тик пробует команды раз в минуту,
|
||
# а не каждую секунду — лежащий TG не вешает рефлекс на 15с-таймаутах
|
||
_FAIL_COOLDOWN = 60
|
||
_last_fail_ts = 0.0
|
||
|
||
# каноническое имя → алиасы (регистронезависимые, без слэша)
|
||
_ALIASES = {
|
||
"лестница": ("лестница", "ladder", "lestnica"),
|
||
"статус": ("статус", "status"),
|
||
"задачи": ("задачи", "tasks", "zadachi", "list"),
|
||
"помощь": ("помощь", "help", "pomoshch", "start", "начать", "команды", "commands"),
|
||
}
|
||
_ALIAS_TO_CMD = {}
|
||
for _c, _aliases in _ALIASES.items():
|
||
for _a in _aliases:
|
||
_ALIAS_TO_CMD[_a] = _c
|
||
|
||
_DESCRIPTIONS = {
|
||
"лестница": "Срез лестницы: уровни, занятость, задачи, эскалации",
|
||
"статус": "Сводка харнеса: тики, задачи, спенд",
|
||
"задачи": "Доска задач: готово и в работе",
|
||
"помощь": "Что я умею и как со мной говорить",
|
||
}
|
||
# транслит-фолбэк, если Bot API не примет кириллицу в именах команд
|
||
_LATIN = {"лестница": "lestnica", "статус": "status", "задачи": "zadachi", "помощь": "pomoshch"}
|
||
|
||
|
||
def parse(text) -> "str | None":
|
||
"""Сообщение — команда? Возвращает каноническое имя или None.
|
||
|
||
Команда: "/имя" (первый токен; @бот отбрасывается) или ОДИНОЧНОЕ
|
||
слово-алиас без слэша («лестница»). Фраза с пробелами без слэша —
|
||
не команда (обычная речь, её слышит минутный LLM-уровень).
|
||
Алиас "list" (e-261001124837-ab3fa7): «доска» из чата — частая
|
||
реплика; без алиаса мин-уровень рожал из неё S-клоны на доске.
|
||
"""
|
||
t = (text or "").strip()
|
||
if not t:
|
||
return None
|
||
if t.startswith("/"):
|
||
tok = t.split()[0].lstrip("/")
|
||
if "@" in tok: # /cmd@botname — адресация боту
|
||
tok = tok.split("@", 1)[0]
|
||
else:
|
||
if len(t.split()) > 1:
|
||
return None
|
||
tok = t
|
||
tok = tok.strip(".,!?;:«»\"'()").lower()
|
||
return _ALIAS_TO_CMD.get(tok)
|
||
|
||
|
||
def _esc(s) -> str:
|
||
"""TG-отправка идёт с parse_mode=HTML — динамику экранируем."""
|
||
return html.escape(str(s or ""), quote=False)
|
||
|
||
|
||
def cmd_ladder() -> str:
|
||
from . import tg_inbox
|
||
return "⚡️ Срез лестницы:\n\n" + _esc(tg_inbox.ladder_snapshot())
|
||
|
||
|
||
def cmd_status() -> str:
|
||
model = es.run_projections()
|
||
cs = model.get("cycle_state") or {}
|
||
tasks = list((model.get("tasks") or {}).values())
|
||
counts = {}
|
||
for t in tasks:
|
||
counts[t.get("status") or "?"] = counts.get(t.get("status") or "?", 0) + 1
|
||
spend = model.get("spend") or []
|
||
tokens = sum(int(e.get("prompt_tokens") or 0) + int(e.get("completion_tokens") or 0)
|
||
for e in spend)
|
||
open_esc = sum(1 for e in (model.get("escalations") or {}).values()
|
||
if isinstance(e, dict) and e.get("status") == "open")
|
||
lines = [f"📊 Статус харнеса · {datetime.now(timezone.utc).strftime('%d.%m %H:%M')} UTC"]
|
||
for name in ("sec", "min", "hour", "day"):
|
||
st = cs.get(name) or {}
|
||
err = f", ошибок {st['errors']}" if st.get("errors") else ""
|
||
lines.append(f" {name}: тиков {st.get('ticks', 0)}{err}")
|
||
parts = [f"{k} {counts[k]}" for k in ("ready", "running", "blocked", "failed", "done") if counts.get(k)]
|
||
lines.append("📋 задачи: " + (" · ".join(parts) if parts else "нет"))
|
||
lines.append(f"🚨 эскалаций открытых: {open_esc}")
|
||
lines.append(f"💰 спенд (последние {len(spend)} тиков): ~{tokens:,} токенов".replace(",", " "))
|
||
return "\n".join(lines)
|
||
|
||
|
||
def cmd_tasks() -> str:
|
||
model = es.run_projections()
|
||
tasks = list((model.get("tasks") or {}).values())
|
||
tasks.sort(key=lambda t: t.get("created") or "", reverse=True)
|
||
board = [t for t in tasks if t.get("status") in ("ready", "running")]
|
||
done = sum(1 for t in tasks if t.get("status") == "done")
|
||
lines = [f"📋 Доска: готово {sum(1 for t in tasks if t.get('status') == 'ready')}"
|
||
f" · в работе {sum(1 for t in tasks if t.get('status') == 'running')}"
|
||
f" · сделано {done}"]
|
||
for t in board[:12]:
|
||
lines.append(f" [{t.get('size', '?')}] {_esc((t.get('title') or '')[:70])}")
|
||
if len(board) > 12:
|
||
lines.append(f" … ещё {len(board) - 12}")
|
||
lines.append("Выполняют часовой/дневной уровни; детали — на дашборде.")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def cmd_help() -> str:
|
||
lines = ["🤖 Simbiot — команды (мгновенные, без LLM):"]
|
||
for c, d in _DESCRIPTIONS.items():
|
||
lines.append(f" /{_LATIN[c]} (или /{c}) — {d}")
|
||
lines.append("")
|
||
lines.append("Обычный текст и голосовые уходят минутному пульсу (LLM) — отвечу по смыслу.")
|
||
lines.append("Большую просьбу можно заказать задачей: результат придёт в чат сам, когда будет готов.")
|
||
return "\n".join(lines)
|
||
|
||
|
||
_HANDLERS = {
|
||
"лестница": cmd_ladder,
|
||
"статус": cmd_status,
|
||
"задачи": cmd_tasks,
|
||
"помощь": cmd_help,
|
||
}
|
||
|
||
|
||
def answer(msg_id: str, cmd: str) -> str:
|
||
"""Собирает ответ команды и отправляет. ReplySent закрывает входящее.
|
||
|
||
Команда — не задача и не LLM-тик: это рефлекс. CommandHandled (после
|
||
успешной отправки) — только наблюдаемость: видно, что входящее закрыто
|
||
именно командой, а не минутным триажем.
|
||
"""
|
||
from . import tg_inbox
|
||
text = _HANDLERS[cmd]()
|
||
# короткий таймаут: сек-тик не должен висеть на TG дольше пары секунд
|
||
tg_inbox.send_reply(msg_id, text, timeout=15)
|
||
try:
|
||
es.append_event("inbox", msg_id, "CommandHandled", {
|
||
"command": cmd, "level": "sec", "handler": "tg_commands",
|
||
})
|
||
except Exception:
|
||
pass # событие — наблюдаемость; ReplySent уже закрыл входящее
|
||
return text
|
||
|
||
|
||
def answer_pending(model: "dict | None" = None):
|
||
"""Отвечает на все входящие-команды без ответа. [(msg_id, cmd), …].
|
||
|
||
Идемпотентно: replied-сообщения пропускаются (проекция свежая —
|
||
send_reply сам перепроверяет по стору и кинет ValueError при гонке
|
||
с минутным уровнем — это не ошибка, просто пропускаем).
|
||
"""
|
||
if model is None:
|
||
model = es.run_projections()
|
||
# есть ли вообще команды-кандидаты: если нет — cooldown не проверяем
|
||
pending = [m for m in (model.get("inbox") or {}).values()
|
||
if isinstance(m, dict) and not m.get("replied") and not m.get("ignored")
|
||
and parse(m.get("text") or "")]
|
||
if not pending:
|
||
return []
|
||
global _last_fail_ts
|
||
if time.time() - _last_fail_ts < _FAIL_COOLDOWN:
|
||
return [] # недавно не смогли отправить — ждём, рефлекс не висит на TG
|
||
handled = []
|
||
for m in pending:
|
||
mid = m.get("id") or (f"m-{m.get('update_id')}" if m.get("update_id") else None)
|
||
if not mid:
|
||
continue
|
||
cmd = parse(m.get("text") or "")
|
||
try:
|
||
answer(mid, cmd)
|
||
handled.append((mid, cmd))
|
||
except ValueError:
|
||
pass # уже отвечено (гонка) — не ошибка
|
||
except Exception as e:
|
||
# TG недоступен и пр. — НЕ отвечено; ставим cooldown и повторим
|
||
# не раньше чем через минуту (рефлекс тикает каждую секунду)
|
||
_last_fail_ts = time.time()
|
||
print(f"[tg-cmd] {mid} /{cmd} не отвечено: {e} "
|
||
f"(cooldown {_FAIL_COOLDOWN}с)", flush=True)
|
||
return handled
|
||
|
||
|
||
# ────────────── регистрация в чате (setMyCommands) ──────────────
|
||
|
||
|
||
def commands_spec(latin: bool = False) -> list:
|
||
return [{"command": (_LATIN[c] if latin else c), "description": d}
|
||
for c, d in _DESCRIPTIONS.items()]
|
||
|
||
|
||
def register_bot_commands():
|
||
"""setMyCommands: latin-имена + кириллические подписи (без 400-пробы).
|
||
|
||
Telegram Bot API принимает в ИМЕНАХ команд только [a-z0-9_]:
|
||
кириллическое имя → 400 BOT_COMMAND_INVALID (подтверждено телом
|
||
ошибки API, t-261004160532-690fcb) — «сначала кириллица, потом
|
||
фолбэк» гарантированно падал первым вызовом при каждом старте.
|
||
В описаниях кириллица валидна, поэтому регистрируем сразу
|
||
единственно возможную комбинацию: latin-имя + кириллическая
|
||
подпись — именно её юзер и видит в меню «/».
|
||
Кириллицей команды остаются доступными и без меню: парсер
|
||
_ALIASES понимает слова «лестница», «статус», «задачи»,
|
||
«помощь» (без слэша) и /лестница-подобные токены.
|
||
|
||
Возвращает 'cyrillic' (подписи кириллицей — то, что видит юзер)
|
||
или None (нет токена / не удалось).
|
||
Повторный вызов идемпотентен — перезапишет список тем же содержимым.
|
||
"""
|
||
from . import tg_inbox
|
||
tg_inbox._load_env()
|
||
if not tg_inbox.TG_TOKEN:
|
||
return None
|
||
try:
|
||
r = tg_inbox._api("setMyCommands",
|
||
{"commands": json.dumps(commands_spec(latin=True), ensure_ascii=False)})
|
||
if r.get("ok"):
|
||
return "cyrillic"
|
||
print(f"[tg-cmd] setMyCommands отклонён: {str(r)[:200]}", flush=True)
|
||
except Exception as e:
|
||
print(f"[tg-cmd] setMyCommands error: {e}", flush=True)
|
||
return None
|
||
|
||
|
||
def get_my_commands() -> list:
|
||
"""Текущий список команд у бота ([(name, description), …])."""
|
||
from . import tg_inbox
|
||
tg_inbox._load_env()
|
||
if not tg_inbox.TG_TOKEN:
|
||
return []
|
||
try:
|
||
r = tg_inbox._api("getMyCommands", {})
|
||
return [(c.get("command"), c.get("description")) for c in r.get("result", [])]
|
||
except Exception:
|
||
return [] |