"""Simbiot Harness — CLI: таски, статус, ручные тики.""" import argparse import json import sys import os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from harness import commands, es def main(): ap = argparse.ArgumentParser(prog="simbiot", description="Simbiot Harness CLI") sub = ap.add_subparsers(dest="cmd", required=True) p_task = sub.add_parser("task", help="подать задачу") p_task.add_argument("title") p_task.add_argument("-d", "--details", default="") p_task.add_argument("-m", "--minutes", type=int, default=5, help="оценка длительности (класс S≤5м M≤60м L≤8ч XL)") p_pol = sub.add_parser("policy", help="задать политику") p_pol.add_argument("--identity", default=None) p_pol.add_argument("--cpu-threshold", type=int, default=None) p_rq = sub.add_parser("requeue", help="вернуть failed-задачу в очередь") p_rq.add_argument("task_id") p_rq.add_argument("-r", "--reason", default="manual requeue") p_pr = sub.add_parser("promote", help="промоушен ready-задачи на уровень выше (S→M→L)") p_pr.add_argument("task_id") p_pr.add_argument("-r", "--reason", default="manual promote") p_tgr = sub.add_parser("tg-reply", help="ответить на входящее TG-сообщение") p_tgr.add_argument("msg_id", help="id входящего (m-NNN)") p_tgr.add_argument("text") p_tgt = sub.add_parser("tg-transcribe", help="расшифровать голосовое (m-NNN) вручную") p_tgt.add_argument("msg_id", help="id входящего (m-NNN)") p_tgt.add_argument("--force", action="store_true", help="перерасшифровать заново") p_lad = sub.add_parser("ladder", help="срез лестницы: уровни, занятость, задачи, эскалации") p_not = sub.add_parser("tg-notify", help="отправить сообщение в чат харнеса (уведомление)") p_not.add_argument("text") p_esc = sub.add_parser("escalate", help="эскалация важного на уровень выше (min→hour→day)") p_esc.add_argument("reason") p_esc.add_argument("-f", "--from-level", default="min", choices=["min", "hour"], help="кто эскалирует") p_esc.add_argument("-c", "--context", default="", help="контекст для верхнего уровня") p_esc.add_argument("-s", "--severity", default="warn", choices=["info", "warn", "critical"]) p_esc.add_argument("-t", "--task-id", default=None, help="связанная задача") p_escl = sub.add_parser("escalations", help="список эскалаций") p_escl.add_argument("--open", action="store_true", help="только открытые") p_ack = sub.add_parser("escalation-ack", help="закрыть эскалацию (принята уровнем выше)") p_ack.add_argument("esc_id") p_ack.add_argument("-l", "--level", default="hour", help="кто принял") p_ack.add_argument("-n", "--note", default="", help="что сделали с эскалацией") sub.add_parser("status", help="состояние харнеса") sub.add_parser("tasks", help="доска задач") p_ev = sub.add_parser("events", help="хвост стора (последние N событий)") p_ev.add_argument("-n", type=int, default=15) p_ev.add_argument("--after", type=int, default=None, help="вывести события ПОСЛЕ этого id (с головы, по возрастанию)") p_init = sub.add_parser("init", help="инициализация стора") p_init.add_argument("--identity", default="Simbiot Harness — ритмичный агент кооператива") args = ap.parse_args() if args.cmd == "init": es.init_db() commands.set_policy(identity=args.identity) print("OK: стор инициализирован, политика записана") elif args.cmd == "task": tid, size = commands.submit_task(args.title, args.details, args.minutes) print(f"OK: задача {tid} [{size}] подана") elif args.cmd == "policy": pol = commands.set_policy(identity=args.identity, cpu_alert_threshold=args.cpu_threshold) print("OK:", json.dumps(pol, ensure_ascii=False)) elif args.cmd == "requeue": es.init_db() tid = commands.requeue_task(args.task_id, args.reason) print(f"OK: задача {tid} возвращена в очередь") elif args.cmd == "promote": es.init_db() tid, new_size = commands.promote_task(args.task_id, args.reason) print(f"OK: задача {tid} повышена до [{new_size}]") elif args.cmd == "tg-reply": es.init_db() from harness import tg_inbox low = args.text.strip().lower() if low in ("/лестница", "/ladder", "/статус", "/status", "лестница", "status"): # срез лестницы по запросу из TG — видимостью управляет сам харнес args.text = "📊 " + tg_inbox.ladder_snapshot() mid = tg_inbox.send_reply(args.msg_id, args.text) print(f"OK: ответ на {mid} отправлен") elif args.cmd == "tg-transcribe": es.init_db() from harness import voice from harness import tg_inbox model = es.run_projections() m = (model.get("inbox") or {}).get(args.msg_id) if m is None: print(f"ERROR: входящее {args.msg_id} не найдено") sys.exit(1) if args.force: # сбрасываем прошлую расшифровку в read model (событие поверх) # — transcribe_message идемпотентен по флагу transcribed es.append_event("inbox", args.msg_id, "MessageTranscribed", { "transcript": "", "engine": None, "reason": None, "reset": True, }) print("(force: прошлый транскрипт сброшен)") if m.get("transcribed") and not args.force: print(f"Уже расшифровано ({m.get('engine')}): {m.get('transcript')}") else: text, engine = voice.transcribe_message( args.msg_id, file_id=m.get("file_id") or "", file_unique_id=m.get("file_unique_id") or "", duration=m.get("duration"), kind=m.get("kind") or "voice") if text: print(f"OK ({engine}): {text}") else: reason = m.get("transcribe_reason") print(f"ПУСТО (engine={engine})" + (f", причина: {reason}" if reason else "") + " — транскрипт записан в стор как пустой") elif args.cmd == "ladder": es.init_db() from harness import tg_inbox print(tg_inbox.ladder_snapshot()) elif args.cmd == "tg-notify": es.init_db() from harness import tg_inbox r = tg_inbox.notify_chat(args.text) print("OK: уведомление отправлено" if r else "SKIP: некому отправить (чат не известен)") elif args.cmd == "escalate": es.init_db() eid, to_level = commands.raise_escalation( args.from_level, args.reason, args.context, args.severity, args.task_id) print(f"OK: эскалация {eid} → {to_level} ({args.severity})") elif args.cmd == "escalations": es.init_db() rows = commands.list_escalations(only_open=args.open) if not rows: print("(эскалаций нет)") for e in rows: ctx = f" | {e.get('context','')[:60]}" if e.get("context") else "" print(f"{e['id']} [{e.get('severity','info'):8}] {e.get('status','?'):5} " f"{e.get('from_level')} → {e.get('to_level')} " f"{e.get('reason','')[:70]}{ctx}") elif args.cmd == "escalation-ack": es.init_db() eid = commands.acknowledge_escalation(args.esc_id, args.level, args.note) print(f"OK: эскалация {eid} закрыта ({args.level})") elif args.cmd == "status": es.init_db() print(json.dumps(commands.status(), ensure_ascii=False, indent=1)) elif args.cmd == "tasks": es.init_db() for t in commands.list_tasks(): print(f"{t['id']} [{t['size']:2}] {t['status']:7} {t['title'][:70]}") elif args.cmd == "events": es.init_db() if args.after is None: # хвост стора: последние N по возрастанию id total = es.count_events() evs = es.get_events(max(0, total - args.n), args.n) else: evs = es.get_events(args.after, args.n) for ev in evs: d = ev["data"][:110] print(f"{ev['id']:4} {ev['timestamp'][11:19]} {ev['aggregate']:8} {ev['event_type']:16} {d}") if __name__ == "__main__": main()