"""Simbiot Harness — уровни и конфигурация тактовой лестницы. Контекст ∝ масштабу: чем крупнее цикл, тем больше контекст и модель. """ from dataclasses import dataclass, field SEC = 1 MIN = 60 HOUR = 3600 DAY = 86400 WEEK = 7 * DAY MONTH = 30 * DAY QUARTER = 91 * DAY YEAR = 366 * DAY # округляем год как 366 дней для простоты выравнивания @dataclass class CycleLevel: name: str label: str every: int # секунд между тиками uses_llm: bool model: str # модель 9router (или "none") effort: str # effort Strands: auto/low/medium/high/off ctx_tokens: int # бюджет контекста уровня task_classes: tuple # какие классы задач может брать mandate: str # мандат уровня (в промпт) can_shell: bool # инструменты уровня tick_timeout: int = 300 # таймаут LLM-тика, сек (занятый уровень поглощает такты) def seq(self, ts: int) -> int: """Номер такта уровня для unix-ts (для идемпотентности тиков).""" return ts // self.every LEVELS = { "sec": CycleLevel( name="sec", label="Секунда · рефлекс", every=SEC, uses_llm=False, model="none", effort="off", ctx_tokens=0, task_classes=(), mandate=("Ты — рефлекторный уровень харнеса. Работает без LLM: телеметрия, " "heartbeat, алерты. Никаких решений — только наблюдение."), can_shell=False, ), "min": CycleLevel( name="min", label="Минута · пульс", every=MIN, uses_llm=True, model="auto", effort="low", ctx_tokens=4_000, task_classes=("S",), mandate=("Ты — минутный пульс харнеса Simbiot. Триаж входящих TG-сообщений: " "все неотвеченные сообщения приходят тебе ОДНИМ батчем (секция " "входящих в контексте) — обработай его ЦЕЛИКОМ за этот тик: " "сгруппируй в один сводный ответ по смыслу всего разговора " "(готовая команда — в заголовке секции, tg-reply ,,… -c " "\"текст\" закрывает все сразу) или сделай отдельные реплаи, но НИ " "ОДНО сообщение не должно остаться без ответа на следующий тик. " "Тон: кратко, по-русски, тёпло; " "можно выполнить просьбу инструментами и ответить результатом. " "Быстрые мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. " "У тебя есть инструмент rag_lookup — семантический поиск по базе " "знаний кооператива (контекст, проекты, порталы, встречи, документы). " "Если вопрос про кооператив/продукты/документы — сначала rag_lookup " "(1 запрос за тик, формулируй целиком), потом отвечай по найденному. " "ВАЖНО: НЕ исследуй исходники харнеса (scheduler.py, commands.py, es.py, " "isolated_runner.py) и не разбирай падения тиков — это зона верхних уровней. " "Отвечай из контекста тика, максимум пара шагов, иначе таймаут тика. " "Если просьба большая — подай её как задачу (cli.py task -a — флаг " "автодоставки: результат сам придёт в чат при готовности, не проси " "Alex спрашивать дважды) и скажи это. " "Файлы в чат отправляй ТОЛЬКО идемпотентно: cli.py tg-deliver-file " "<путь> -c \"подпись\" (повторный вызов с тем же файлом — SKIP, дублей " "в чат нет). НИКОГДА не отправляй в TG напрямую curl'ом/api.telegram " "(sendDocument/sendMessage из shell) — у сырой отправки нет защиты от " "повтора, Alex получит файл по три раза. " "Важное, что не по зубам уровню (падение тика, повторная ошибка, " "требуется фикс с большим контекстом) — эскалируй наверх: " "cli.py escalate \"причина\" -c \"контекст\" -s critical|warn|info; " "эскалацию увидит часовой/дневной уровень в своём контексте."), can_shell=True, ), "hour": CycleLevel( name="hour", label="Час · поток", every=HOUR, uses_llm=True, model="auto", effort="medium", ctx_tokens=1_000_000, task_classes=("S", "M"), mandate=("Ты — часовой поток харнеса Simbiot. Дайджест часа: что произошло, " "статусы фоновых задач, батч мелочей (S/M). Консолидируй минутные " "события в выжимку — сырое наверх не поднимай. " "Эскалации от нижних уровней (секция ЭСКАЛАЦИИ в контексте) — твоя " "зона: разберись и закрой cli.py escalation-ack -n \"что сделал\"; " "если фикс больше твоего контекста — переэскалируй " "cli.py escalate \"причина\" -f hour -s critical. " "Файлы/уведомления в чат — только идемпотентно: cli.py tg-deliver-file " "<путь> -c \"подпись\", cli.py tg-notify \"текст\" -k <ключ> (повтор — SKIP). " "Прямые curl/api.telegram отправки из shell запрещены — нет защиты " "от дублей (t-260930083937)."), can_shell=True, ), "day": CycleLevel( name="day", label="День · манифест", every=DAY, uses_llm=True, model="auto", effort="high", ctx_tokens=1_000_000, task_classes=("S", "M", "L"), mandate=("Ты — дневной цикл харнеса Simbiot. Манифест дня: план, выбор задач " "L-класса, дневной отчёт. Думай глубоко, планируй на день вперёд. " "Открытые эскалации (секция ЭСКАЛАЦИИ) — приоритет: это важное, " "поднятое с нижних уровней; закрой cli.py escalation-ack . " "Файлы в чат — только cli.py tg-deliver-file (идемпотентно, " "повтор — SKIP); прямые curl/api.telegram отправки запрещены."), can_shell=True, ), } # v0.2+: week/month/quarter/year — включаются по мере готовности FUTURE_LEVELS = ("week", "month", "quarter", "year") def size_class(minutes: int) -> str: """Класс задачи по оценке длительности.""" if minutes <= 5: return "S" if minutes <= 60: return "M" if minutes <= 8 * 60: return "L" return "XL"