feat(scheduler): батч-обработка задач в тике — пачка ready-задач в бюджете контекста (t-260930101325-010ba3)

Причина: find_task возвращал ОДНУ ready-задачу за такт — при 5+ ready
очередь росла быстрее исполнения (1 задача/такт).

Теперь:
- select_task_batch: ВСЕ ready-задачи класса уровня, старейшая первой;
- кап пачки MAX_TASK_BATCH (min=3, hour=6, day=6; env SIMBIOT_BATCH_MAX_<LEVEL>);
- бюджет контекста: секция задач <= ctx_tokens*3*0.5 символов,
  мандат/выжимки/эскалации/инбокс остаются в тике; старейшая берётся всегда;
- run_level_tick: claim всего батча ДО воркера; TaskCompleted/TaskFailed
  ПО каждой отдельно (смерть воркера не теряет батч в running);
- build_context: секция 'ТЕКУЩАЯ РАБОТА — БАТЧ из N' с постановками,
  инструкция закрывать каждую задачу cli.py complete <id> -a;
- CLI complete: закрытие running-задачи с результатом
  (commands.complete_task_with_delivery — доставка в TG идемпотентна).

Тесты: tests/test_task_batch.py — 28 проверок (ALL PASS), набор 8/8 зелёный.
This commit is contained in:
Simbiot Harness 2026-09-30 14:13:30 +00:00
parent 1bc33d5678
commit 2f4e3d22a7
3 changed files with 423 additions and 35 deletions

12
cli.py
View file

@ -29,6 +29,10 @@ def main():
p_rq.add_argument("task_id") p_rq.add_argument("task_id")
p_rq.add_argument("-r", "--reason", default="manual requeue") p_rq.add_argument("-r", "--reason", default="manual requeue")
p_cmp = sub.add_parser("complete", help="закрыть running-задачу с результатом (батч-тик, t-260930101325)")
p_cmp.add_argument("task_id")
p_cmp.add_argument("-a", "--answer", required=True, help="результат задачи")
p_pr = sub.add_parser("promote", help="промоушен ready-задачи на уровень выше (S→M→L)") p_pr = sub.add_parser("promote", help="промоушен ready-задачи на уровень выше (S→M→L)")
p_pr.add_argument("task_id") p_pr.add_argument("task_id")
p_pr.add_argument("-r", "--reason", default="manual promote") p_pr.add_argument("-r", "--reason", default="manual promote")
@ -133,6 +137,14 @@ def main():
es.init_db() es.init_db()
tid = commands.requeue_task(args.task_id, args.reason) tid = commands.requeue_task(args.task_id, args.reason)
print(f"OK: задача {tid} возвращена в очередь") print(f"OK: задача {tid} возвращена в очередь")
elif args.cmd == "complete":
# батч-тик (t-260930101325): исполнитель закрывает задачи пачки
# по отдельности; с await_delivery результат уходит в TG (идемпотентно)
es.init_db()
tid, delivered = commands.complete_task_with_delivery(
args.task_id, args.answer, level="manual")
print(f"OK: задача {tid} завершена"
+ (" (доставлено в TG)" if delivered else ""))
elif args.cmd == "promote": elif args.cmd == "promote":
es.init_db() es.init_db()
tid, new_size = commands.promote_task(args.task_id, args.reason) tid, new_size = commands.promote_task(args.task_id, args.reason)

View file

@ -60,8 +60,59 @@ def deterministic_digest_stub(level_cfg, seq: int, reason: str):
}) })
# ──────────── БАТЧ-ВЫБОР ЗАДАЧ В ТИК (t-260930101325-010ba3) ────────────
# Alex: «почему тик берёт только одну задачу, когда мог бы сделать все?»
# Причина: find_task возвращал ОДНУ ready-задачу за такт — при 5+ ready
# задачи копились быстрее, чем исполнялись (1 задача/такт, час = 60 мин/шт).
# Входящие TG уже батчатся одним сообщением (t-260930064924) — очередь
# задач теперь тоже: тик берёт ВСЕ готовые задачи своего класса,
# пока влезают в бюджет контекста и лимит пачки.
MAX_TASK_BATCH = {"min": 3, "hour": 6, "day": 6} # env override: SIMBIOT_BATCH_MAX_<LEVEL>
# доля символьного бюджета контекста, которую можно отдать под батч задач
# (остаток — мандат, выжимки, эскалации, инбокс, доска)
TASK_BUDGET_SHARE = 0.5
# консервативный коэффициент символ→токен для русского текста (2–3.5 факт)
CHARS_PER_TOKEN = 3
def _batch_cap(level_cfg):
"""Лимит задач в пачке уровня (env SIMBIOT_BATCH_MAX_<LEVEL> > дефолт)."""
env = os.environ.get(f"SIMBIOT_BATCH_MAX_{level_cfg.name.upper()}")
if env:
try:
return max(1, int(env))
except ValueError:
pass
return MAX_TASK_BATCH.get(level_cfg.name, 3)
def _est_minutes(t: dict) -> int:
"""Оценка длительности: из TaskSubmitted или дефолт по классу.
est_minutes пишется в событие TaskSubmitted (commands.submit_task) и
поднимается проекцией (es.py); для старых задач — дефолт по классу.
Нужна для сводной оценки батча в контексте (исполнитель планирует
тайм-бюджет тика) — НЕ жёсткий фильтр (оценка ≠ факт).
"""
est = t.get("est_minutes")
if isinstance(est, (int, float)) and est > 0:
return int(est)
return {"S": 5, "M": 60, "L": 480, "XL": 1440}.get(t.get("size"), 30)
def _task_cost(t: dict) -> int:
"""Символьная стоимость задачи в контексте тика (для бюджет-отсечки)."""
return 260 + len(t.get("title") or "") + len(t.get("details") or "")
def task_batch_cost(tasks: list) -> int:
"""Суммарная символьная стоимость батча (для тестов и лога)."""
return sum(_task_cost(t) for t in tasks)
def find_task(level_cfg, model: dict): def find_task(level_cfg, model: dict):
"""Ищет ready-задачу доступного классу уровня. Возвращает task_id или None.""" """Одиночный выбор: ПЕРВАЯ ready-задача класса уровня (для совместимости)."""
classes = level_cfg.task_classes classes = level_cfg.task_classes
for tid, t in model["tasks"].items(): for tid, t in model["tasks"].items():
if t["status"] == "ready" and t.get("size") in classes: if t["status"] == "ready" and t.get("size") in classes:
@ -69,6 +120,41 @@ def find_task(level_cfg, model: dict):
return None return None
def select_task_batch(level_cfg, model: dict):
"""Батч ready-задач доступных классов в бюджете контекста тика.
Возвращает [task_id, ...] — от старейших к свежим (анти-старение
очереди). Ограничения:
1) класс задачи ∈ task_classes уровня (как раньше у find_task);
2) кап пачки MAX_TASK_BATCH (env SIMBIOT_BATCH_MAX_<LEVEL>) —
страховка: тик не превращается только в исполнение задач,
дайджест уровня остаётся его работой;
3) бюджет контекста: секция задач не съедает больше
ctx_tokens·CHARS_PER_TOKEN·TASK_BUDGET_SHARE символов —
мандат, выжимки, эскалации и инбокс остаются в тике.
Первая (старейшая) задача берётся всегда — как раньше find_task;
бюджетная отсечка работает начиная со второй пачки.
"""
classes = level_cfg.task_classes
if not classes:
return []
ready = [t for t in (model.get("tasks") or {}).values()
if t.get("status") == "ready" and t.get("size") in classes]
ready.sort(key=lambda t: t.get("created") or "")
cap = _batch_cap(level_cfg)
budget_chars = int(level_cfg.ctx_tokens * CHARS_PER_TOKEN * TASK_BUDGET_SHARE)
picked, used = [], 0
for t in ready:
cost = _task_cost(t)
if picked and used + cost > budget_chars:
break # бюджет контекста исчерпан — хвост ждёт следующих тактов
picked.append(t)
used += cost
if len(picked) >= cap:
break
return [t["id"] for t in picked]
def _lvl_rank(name: str) -> int: def _lvl_rank(name: str) -> int:
"""Ранг уровня в лестнице; неизвестный → ниже всех (эскалации ему не адресованы).""" """Ранг уровня в лестнице; неизвестный → ниже всех (эскалации ему не адресованы)."""
return _ORDER.index(name) if name in _ORDER else len(_ORDER) return _ORDER.index(name) if name in _ORDER else len(_ORDER)
@ -153,8 +239,12 @@ def gate_allows_llm(level_cfg, signals) -> bool:
return any(order.get(s[0], 3) <= rank for s in signals) return any(order.get(s[0], 3) <= rank for s in signals)
def build_context(level_cfg, model: dict, task=None): def build_context(level_cfg, model: dict, batch=None):
"""Сборка контекста тика: мандат + память уровня + выжимки нижних + задача.""" """Сборка контекста тика: мандат + память уровня + выжимки нижних + батч задач.
batch — список задач-словарей (проекции) этого тика; одиночная задача
(legacy) передаётся списком из одного элемента.
"""
parts = [level_cfg.mandate] parts = [level_cfg.mandate]
# идентичность/цели v0.1 — статично из policy (если есть) # идентичность/цели v0.1 — статично из policy (если есть)
@ -243,15 +333,35 @@ def build_context(level_cfg, model: dict, task=None):
" -c \"текст\"\n" + "\n".join(lines)) " -c \"текст\"\n" + "\n".join(lines))
# доска задач # доска задач
tasks = [t for t in model["tasks"].values() if t["status"] in ("ready", "running")] # батч задач этого тика (селекция по budget) — текущая постановка
if tasks: if batch is None:
lines = [f"- {t['id']}: {t['title'][:80]} [{t['size']}] {t['status']}" for t in tasks[:15]] batch_ids = select_task_batch(level_cfg, model)
parts.append("ДОСКА ЗАДАЧ:\n" + "\n".join(lines)) batch = [model["tasks"].get(tid) for tid in batch_ids]
batch = [t for t in batch if t]
if batch:
lines = []
for i, t in enumerate(batch, 1):
est = _est_minutes(t)
lines.append(f"{i}. {t['id']} [{t['size']}, ~{est}м] {t['title'][:100]}")
if t.get("details"):
lines.append(f" ПОСТАНОВКА: {t['details'][:1800]}")
parts.append(
f"ТЕКУЩАЯ РАБОТА — БАТЧ из {len(batch)} задач (выполняй в этом тике, "
"самую старую — ПЕРВОЙ; не растягивай: тик ограничен таймаутом и "
f"бюджетом шагов; суммарная оценка ~{sum(_est_minutes(t) for t in batch)}м):\n"
+ "\n".join(lines)
+ "\nКаждую завершённую закрывай ОТДЕЛЬНО сразу после выполнения: "
"./venv/bin/python cli.py complete <task_id> -a \"результат\" "
"(идемпотентно: завершённая уже не завершится повторно; с флагом "
"await_delivery результат сам уйдёт в чат). Не успевшая — вернись "
"в следующий тик (она останется/вернётся в ready). Сводный итог "
"и по каждой задаче — в дайджест."
)
if task: tasks_board = [t for t in model["tasks"].values() if t["status"] in ("ready", "running")]
parts.append(f"ТЕКУЩАЯ ЗАДАЧА: {task['id']}: {task['title']}\n" if tasks_board:
+ (f"ПОСТАНОВКА: {task['details']}\n" if task.get("details") else "") lines = [f"- {t['id']}: {t['title'][:80]} [{t['size']}] {t['status']}" for t in tasks_board[:15]]
+ "Выполни её и выдай результат.") parts.append("ДОСКА ЗАДАЧ:\n" + "\n".join(lines))
return "\n\n".join(parts) return "\n\n".join(parts)
@ -276,36 +386,49 @@ def _notify_busy(level_cfg, task):
def run_level_tick(level_cfg, seq: int, model: dict): def run_level_tick(level_cfg, seq: int, model: dict):
"""Один тик уровня с LLM: дайджест + (опц.) задача. Возвращает текст. """Один тик уровня с LLM: дайджест + (опц.) БАТЧ задач. Возвращает (text, [task_id]).
LLM-тик гоняется в изолированном сабпроцессе (isolated_runner): LLM-тик гоняется в изолированном сабпроцессе (isolated_runner):
у Strands глобальный asyncio-рантайм, второй create_harness() у Strands глобальный asyncio-рантайм, второй create_harness()
в том же процессе валит его («cannot schedule new futures»). в том же процессе валит его («cannot schedule new futures»).
Батч (t-260930101325): тик берёт ВСЕ ready-задачи своего класса,
влезающие в бюджет контекста и кап пачки — одна за такт была
медленнее пополнения очереди. Claim каждой — ДО старта воркера;
завершение/сбой — ПО каждой задаче отдельно, сразу после возврата
воркера, чтобы таймаут/краш не потерял уже сделанное.
""" """
from . import isolated_runner from . import isolated_runner
task_id = find_task(level_cfg, model) batch_ids = select_task_batch(level_cfg, model)
task = model["tasks"].get(task_id) if task_id else None batch = [model["tasks"].get(tid) for tid in batch_ids]
if task: batch = [t for t in batch if t]
es.append_event("task", task_id, "TaskClaimed", { claimed = []
for t in batch:
es.append_event("task", t["id"], "TaskClaimed", {
"level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}", "level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}",
}) })
claimed.append(t["id"])
# TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень) # TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень)
try: try:
from . import tg_inbox _notify_busy(level_cfg, t)
_notify_busy(level_cfg, task)
except Exception: except Exception:
pass # уведомление — best effort, не валить тик pass # уведомление — best effort, не валить тик
prompt = build_context(level_cfg, model, task) if not claimed:
claimed = None
prompt = build_context(level_cfg, model, batch)
text, usage, trace_id = isolated_runner.run_llm_tick( text, usage, trace_id = isolated_runner.run_llm_tick(
level_cfg, prompt, level_cfg, prompt,
session_id=f"s-{level_cfg.name}-{seq}", session_id=f"s-{level_cfg.name}-{seq}",
trace_prefix=f"{level_cfg.name}-{seq}", trace_prefix=f"{level_cfg.name}-{seq}",
) )
if text is None: if text is None:
# воркер упал — задача не выполнена, откатываем claim # воркер упал (таймаут/краш) — НИ одна задача не выполнена с его
if task: # точки зрения, но executor мог частично поработать инструментами.
es.append_event("task", task_id, "TaskFailed", { # По одной TaskFailed на каждую claimed — иначе батч висит в running
# до orphan_requeue (час+). Сироты всё равно подстрахуют.
for tid in (claimed or []):
es.append_event("task", tid, "TaskFailed", {
"error": str(usage.get("error", "llm worker failed"))[:500], "error": str(usage.get("error", "llm worker failed"))[:500],
}) })
raise RuntimeError(str(usage.get("error", "llm worker failed"))) raise RuntimeError(str(usage.get("error", "llm worker failed")))
@ -325,24 +448,24 @@ def run_level_tick(level_cfg, seq: int, model: dict):
"level": level_cfg.name, "trace_id": trace_id, "source": "runner", "level": level_cfg.name, "trace_id": trace_id, "source": "runner",
**{k: v for k, v in usage.items() if k != "error"}, **{k: v for k, v in usage.items() if k != "error"},
}) })
if task: # БАТЧ-ЗАВЕРШЕНИЕ (t-260930101325): TaskCompleted по КАЖДОЙ задаче
es.append_event("task", task_id, "TaskCompleted", { # пачки (общий результат-дайджест — answer каждой; per-task отчёт
# исполнитель пишет в дайджесте). Доставка — той же командой
# commands.deliver_result (идемпотентна: без флага/уже sent — Skip).
# Ответ урезаем на задачу, чтобы 6 задач не раздули 6×2000.
for tid in (claimed or []):
es.append_event("task", tid, "TaskCompleted", {
"result_ref": trace_id, "answer": (text or "")[:2000], "result_ref": trace_id, "answer": (text or "")[:2000],
}) })
# автодоставка (t-260929205927): задачу заказали из TG (await_delivery)
# — результат уходит в чат сразу после завершения, без повторного запроса.
# Единственный путь доставки — commands.deliver_result (идемпотентен:
# без флага / уже sent — молча пропустит). Не отправилось здесь —
# дослает delivery-демон (deliver_pending) на следующем проходе.
try: try:
from . import commands as _cmd from . import commands as _cmd
if _cmd.deliver_result(task_id, text or "", owner=f"tick:{level_cfg.name}"): if _cmd.deliver_result(tid, text or "", owner=f"tick:{level_cfg.name}"):
_log(f"delivery: результат {task_id} доставлен в TG") _log(f"delivery: результат {tid} доставлен в TG")
except Exception as e: except Exception as e:
# доставка — best effort: не валить тик; DeliverySent допишет # доставка — best effort: не валить тик; DeliverySent допишет
# delivery-демон (см. deliver_pending) — задача не потеряется # delivery-демон (см. deliver_pending) — задача не потеряется
_log(f"delivery: отправка {task_id} не удалась ({e}) — доставит delivery-демон") _log(f"delivery: отправка {tid} не удалась ({e}) — доставит delivery-демон")
return text, task_id return text, claimed or []
def process_level(level_cfg, model: dict): def process_level(level_cfg, model: dict):
@ -391,8 +514,8 @@ def process_level(level_cfg, model: dict):
# секундный уровень: детерминированный рефлекс без LLM # секундный уровень: детерминированный рефлекс без LLM
text, task_id = _reflex_tick(level_cfg, seq, model) text, task_id = _reflex_tick(level_cfg, seq, model)
return f"reflex ok {task_id or ''}".strip() return f"reflex ok {task_id or ''}".strip()
text, task_id = run_level_tick(level_cfg, seq, model) text, batch_ids = run_level_tick(level_cfg, seq, model)
return f"ran (task={task_id or '-'})" return f"ran (tasks={','.join(batch_ids) or '-'})"
except Exception as e: except Exception as e:
es.append_event("harness", "errors", "CycleError", { es.append_event("harness", "errors", "CycleError", {
"level": level_cfg.name, "error": str(e)[:1000], "level": level_cfg.name, "error": str(e)[:1000],

253
tests/test_task_batch.py Normal file
View file

@ -0,0 +1,253 @@
# -*- coding: utf-8 -*-
"""t-260930101325-010ba3: батч-обработка задач в часовом тике (hermetic, без LLM/БД).
Запуск: ./venv/bin/python tests/test_task_batch.py
Проверяет: select_task_batch берёт ВСЕ ready-задачи класса уровня (не одну),
уважает кап пачки (SIMBIOT_BATCH_MAX_<LEVEL>) и бюджет контекста
(ctx_tokens·CHARS_PER_TOKEN·TASK_BUDGET_SHARE), берёт старейшую первой,
run_level_tick клеймает весь батч (проверка на фейковом isolated_runner);
build_context показывает постановки всех задач пачки; CLI complete из
секции «ТЕКУЩАЯ РАБОТА» существует и работает через
commands.complete_task_with_delivery → TaskCompleted (без TG — skip).
"""
import os
import sys
import json
sys.path.insert(0, "/root/simbiot-harness")
# чистое окружение: env-капы не должны влиять на тесты
for _v in ("SIMBIOT_BATCH_MAX_MIN", "SIMBIOT_BATCH_MAX_HOUR", "SIMBIOT_BATCH_MAX_DAY"):
os.environ.pop(_v, None)
from harness import scheduler
from harness.levels import LEVELS
fails = []
def check(name, cond, extra=""):
print(("PASS" if cond else "FAIL"), name, ("| " + extra) if not cond else "")
if not cond:
fails.append(name)
def _task(tid, size="M", created="2026-09-30T08:00:00", title="задача",
details="", status="ready", est=None):
t = {"id": tid, "title": title, "size": size, "status": status,
"created": created, "details": details}
if est is not None:
t["est_minutes"] = est
return t
def _model(tasks):
return {"tasks": tasks, "inbox": {}, "escalations": {}, "policy": {},
"digests": {}}
# ── 1. Одна задача — прежнее поведение find_task ──
one = {"t-a": _task("t-a")}
check("1: одна ready → пачка из 1",
scheduler.select_task_batch(LEVELS["hour"], _model(one)) == ["t-a"])
check("1: find_task (legacy) согласован с батчем",
scheduler.find_task(LEVELS["hour"], _model(one)) == "t-a")
# ── 2. Три ready — БАТЧ из трёх (главное: не одна за такт) ──
three = {"t-a": _task("t-a", created="2026-09-30T08:00:00"),
"t-b": _task("t-b", created="2026-09-30T07:00:00"),
"t-c": _task("t-c", created="2026-09-30T09:00:00")}
batch = scheduler.select_task_batch(LEVELS["hour"], _model(three))
check("2: 3 ready → пачка из 3", sorted(batch) == ["t-a", "t-b", "t-c"], str(batch))
check("2: порядок — старейшая ПЕРВАЯ", batch[0] == "t-b", str(batch))
# ── 3. Класс задачи — фильтр уровня (как раньше у find_task) ──
mixed = {"t-s": _task("t-s", size="S"), "t-m": _task("t-m", size="M"),
"t-l": _task("t-l", size="L")}
batch = scheduler.select_task_batch(LEVELS["hour"], _model(mixed))
check("3: hour берёт S+M, L не его класс", sorted(batch) == ["t-m", "t-s"], str(batch))
check("3: min берёт только S",
scheduler.select_task_batch(LEVELS["min"], _model(mixed)) == ["t-s"])
# ── 4. Кап пачки MAX_TASK_BATCH + env-оверрайд ──
many = {f"t-{i}": _task(f"t-{i}", created=f"2026-09-30T0{i}:00:00") for i in range(8)}
batch = scheduler.select_task_batch(LEVELS["hour"], _model(many)) # дефолт hour=6
check("4: hour: кап 6 из 8 ready", len(batch) == 6, str(len(batch)))
os.environ["SIMBIOT_BATCH_MAX_HOUR"] = "2"
batch = scheduler.select_task_batch(LEVELS["hour"], _model(many))
check("4: env SIMBIOT_BATCH_MAX_HOUR=2 переопределяет",
batch == ["t-0", "t-1"], str(batch))
os.environ.pop("SIMBIOT_BATCH_MAX_HOUR", None)
# ── 5. Бюджет контекста: секция задач не съедает ctx тика ──
# hour: ctx 32K токенов · 3 симв/токен · 0.5 = 48_000 симв на батч.
# Задача ~9261 симв → в 48К влезает 5 (46305), 6-я (55566) — нет.
big_details = "х" * 9000
huge = {f"t-{i}": _task(f"t-{i}", details=big_details,
created=f"2026-09-30T0{i}:00:00") for i in range(8)}
batch = scheduler.select_task_batch(LEVELS["hour"], _model(huge))
cost = scheduler.task_batch_cost([huge[t] for t in batch])
budget = int(32_000 * scheduler.CHARS_PER_TOKEN * scheduler.TASK_BUDGET_SHARE)
check("5: бюджетная отсечка (5 из 8, кап 6 не достигнут)", len(batch) == 5, f"n={len(batch)}")
check("5: суммарная стоимость ≤ 48К симв", cost <= budget, str(cost))
check("5: 6-я задача пробила бы бюджет (отсечка не случайна)",
cost + scheduler._task_cost(huge["t-5"]) > budget,
f"cost6={cost + scheduler._task_cost(huge['t-5'])}")
# узкие задачи — бюджет не режет до капа
wide = {f"t-{i}": _task(f"t-{i}", details="короткая", created=f"2026-09-30T0{i}:00:00")
for i in range(8)}
check("5: узкие — все 6 влезли",
len(scheduler.select_task_batch(LEVELS["hour"], _model(wide))) == 6)
# ── 6. Первая (старейшая) берётся ВСЕГДА, даже если бюджет пробит ──
fattail = {"t-huge": _task("t-huge", details="y" * 200_000, created="2026-09-30T07:00:00"),
"t-s": _task("t-s", created="2026-09-30T08:00:00")}
batch = scheduler.select_task_batch(LEVELS["hour"], _model(fattail))
check("6: сверхтяжёлая старейшая берётся одна (не блокирует очередь)",
batch == ["t-huge"], str(batch))
# ── 7. пустая доска / не ready / чужой класс → пусто ──
check("7: нет ready → пустой батч",
scheduler.select_task_batch(LEVELS["hour"], _model({})) == [])
done = {"t-x": _task("t-x", status="done")}
check("7: done/failed не берутся", scheduler.select_task_batch(LEVELS["hour"], _model(done)) == [])
check("7: sec без классов → пусто", scheduler.select_task_batch(LEVELS["sec"], _model(three)) == [])
# ── 8. build_context: секция БАТЧА с постановками всех задач ──
three_det = {"t-a": _task("t-a", created="2026-09-30T08:00:00"),
"t-b": _task("t-b", created="2026-09-30T07:00:00",
details="шаги: 1) разбор 2) фикс"),
"t-c": _task("t-c", created="2026-09-30T09:00:00")}
fake = _model({**three_det,
"t-d": _task("t-d", size="L", created="2026-09-30T10:00:00",
details="L-постановка")}) # t-d — L, не часовой
ctx = scheduler.build_context(LEVELS["hour"], fake)
check("8: секция батча есть (3, без L)", "ТЕКУЩАЯ РАБОТА — БАТЧ из 3" in ctx,
ctx.split("ДОСКА ЗАДАЧ")[0][-300:])
check("8: все три id в секции", all(t in ctx for t in ("t-a", "t-b", "t-c")))
check("8: постановка (details) в секции", "разбор" in ctx)
check("8: L-постановка НЕ в батче (в доске — можно)",
"L-постановка" not in ctx.split("ДОСКА ЗАДАЧ")[0].split("ТЕКУЩАЯ РАБОТА")[1]
if "ТЕКУЩАЯ РАБОТА" in ctx else True, "t-d в батче")
check("8: инструкция закрывать отдельно",
'cli.py complete <task_id> -a' in ctx)
check("8: legacy-вызов с одной задачей работает",
"БАТЧ из 1" in scheduler.build_context(
LEVELS["hour"], _model(one), [one["t-a"]]))
# ── 9. run_level_tick клеймает ВЕСЬ батч (фейковый воркер, без LLM) ──
import harness.isolated_runner as _ir_module
written = []
class _FakeES:
"""Стаб es: пишем append-события в память, БД не трогаем."""
def __getattr__(self, name):
def _stub(*a, **k):
return None
return _stub
def append_event(self, aggregate, aid, evtype, data={}):
written.append((evtype, aid))
return "x"
class _FakeRunner:
@staticmethod
def run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None):
return "дайджест тика: отработан батч из трёх", {"model_calls": 1}, "trace-x"
_orig_es, _orig_runner_mod = scheduler.es, _ir_module
try:
scheduler.es = _FakeES()
class _FakeRunnerMod:
run_llm_tick = staticmethod(_FakeRunner.run_llm_tick)
sys.modules["harness.isolated_runner"] = _FakeRunnerMod()
# isolated_runner импортируется inside run_level_tick ("from . import isolated_runner") —
# патчим атрибут модуля harness.isolated_runner
_ir_module.run_llm_tick = staticmethod(_FakeRunner.run_llm_tick)
text, ids = scheduler.run_level_tick(LEVELS["hour"], 4242, _model(three))
claims = [a for (e, a) in written if e == "TaskClaimed"]
comps = [a for (e, a) in written if e == "TaskCompleted"]
check("9: TaskClaimed на ВСЕ 3 задачи", sorted(claims) == ["t-a", "t-b", "t-c"], str(claims))
check("9: TaskCompleted на ВСЕ 3 задачи", sorted(comps) == ["t-a", "t-b", "t-c"], str(comps))
check("9: батч в логе тика", ','.join(sorted(ids)) == "t-a,t-b,t-c", str(ids))
# секция батча в реально собранном промпте (run_level_tick → build_context)
written.clear()
seen_prompt = {}
def _spy_run(level_cfg, prompt, **k):
seen_prompt["p"] = prompt
return "ok", {"model_calls": 1}, "tr"
_ir_module.run_llm_tick = staticmethod(_spy_run)
scheduler.run_level_tick(LEVELS["hour"], 4243, _model(three))
check("9: промпт тика содержит батч из 3 и постановки",
"БАТЧ из 3" in seen_prompt["p"] and "t-c" in seen_prompt["p"])
finally:
scheduler.es, scheduler.isolated_runner = _orig_es, _orig_runner_mod
# ── 10. Смерть воркера → TaskFailed на каждую claimed (нет сирот в running) ──
def _dead_run(level_cfg, prompt, **k):
return None, {"error": "timeout"}, "tr"
written.clear()
scheduler.es = _FakeES()
try:
_ir_module.run_llm_tick = staticmethod(_dead_run)
try:
scheduler.run_level_tick(LEVELS["hour"], 4244, _model(three))
check("10: смерть воркера → RuntimeError", False, "не поднял исключение")
except RuntimeError:
pass
fails_ev = [a for (e, a) in written if e == "TaskFailed"]
check("10: TaskFailed на ВСЕ 3 claimed", sorted(fails_ev) == ["t-a", "t-b", "t-c"],
str(fails_ev))
finally:
scheduler.es = _orig_es
# ── 11. CLI complete: аргументы есть, complete_task_with_delivery работает ──
import subprocess
r = subprocess.run(["./venv/bin/python", "cli.py", "complete", "--help"],
cwd="/root/simbiot-harness", capture_output=True, text=True)
check("11: cli complete существует", r.returncode == 0 and "-a" in r.stdout, r.stdout[:200])
# e2e по реальному стору БЕЗ TG (нет чата → deliver_result False, но TaskCompleted есть)
os.environ["SIMBIOT_DB"] = "/tmp/simbiot_batch_e2e.db"
import harness.es as es
import importlib
importlib.reload(es)
es.init_db()
from harness import commands as cmds
tid, size = cmds.submit_task("тест батча e2e", "details", minutes=10)
es.append_event("task", tid, "TaskClaimed", {"level": "hour", "session_id": "s-test"})
tid2, delivered = cmds.complete_task_with_delivery(tid, "сделано: тест", level="manual")
model = es.run_projections()
check("11: e2e complete → done", model["tasks"][tid]["status"] == "done")
# answer хранится в СОБЫТИИ TaskCompleted (проекция его не поднимает) —
# читаем из стора напрямую, как это делает deliver_pending
db = es._conn()
try:
row = db.execute(
"SELECT data FROM events WHERE aggregate='task' AND aggregate_id=? "
"AND event_type='TaskCompleted' ORDER BY id DESC LIMIT 1", (tid,)).fetchone()
finally:
db.close()
check("11: e2e ответ записан в TaskCompleted",
row is not None and "сделано: тест" in json.loads(row["data"]).get("answer", ""),
str(row))
try:
cmds.complete_task_with_delivery(tid, "повтор", level="manual")
check("11: повтор complete не пройдёт (уже done)", False)
except ValueError as e:
check("11: повтор complete отклонён (идемпотентность стейт-машиной)", "running" in str(e))
print(f"\n{'FAIL: ' + str(len(fails)) if fails else 'ALL PASS'}")
for f in fails:
print(" -", f)
sys.exit(1 if fails else 0)