simbiot-harness/tests/test_task_batch.py

# -*- coding: utf-8 -*-
"""t-261001090908-18a4f5: тест синхронизирован с контекст-схемой 11a8ccc (min 256k / hour 512k / day 1M), hermetic.

Запуск: ./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 (t-261002170836: часовой подъём L)", sorted(batch) == ["t-l", "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 тика ──
# Бюджет выводится из контекст-схемы уровня (11a8ccc: min 256k → hour 512k →
# day 1M ctx_tokens): budget = ctx_tokens · CHARS_PER_TOKEN · TASK_BUDGET_SHARE,
# для hour = 512_000 · 3 · 0.5 = 768_000 симв на батч. Размеры деталей тоже
# выведены из бюджета — схема не должна рассыпаться при следующей смене уровней.
budget = int(LEVELS["hour"].ctx_tokens * scheduler.CHARS_PER_TOKEN * scheduler.TASK_BUDGET_SHARE)
check("5: контекст-схема 11a8ccc (min 256k / hour 512k / day 1M)",
      LEVELS["min"].ctx_tokens == 256_000 and LEVELS["hour"].ctx_tokens == 512_000
      and LEVELS["day"].ctx_tokens == 1_000_000,
      f"min={LEVELS['min'].ctx_tokens} hour={LEVELS['hour'].ctx_tokens} day={LEVELS['day'].ctx_tokens}")
# Задача с деталями budget//5 - 300 симв стоит ≈ budget/5 (263+details);
# 5 шт влезают (сумма < 1.0·budget), 6-я пробивает (кап 6 не достигнут).
big_details = "х" * (budget // 5 - 300)
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])
check("5: бюджетная отсечка (5 из 8, кап 6 не достигнут)", len(batch) == 5, f"n={len(batch)}")
check("5: суммарная стоимость ≤ бюджету hour (768К симв)", 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. Первая (старейшая) берётся ВСЕГДА, даже если бюджет пробит ──
# Размер t-huge — от бюджета (details = 2·budget симв): влезть следом для
# t-s (~263 симв) места нет, но первую задача берётся безусловно.
fattail = {"t-huge": _task("t-huge", details="y" * (budget * 2), 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-e": _task("t-e", size="XL", created="2026-09-30T10:00:00",
                            details="XL-постановка")})  # t-e — XL, не часовой
ctx = scheduler.build_context(LEVELS["hour"], fake)
check("8: секция батча есть (3, без XL)", "ТЕКУЩАЯ РАБОТА — БАТЧ из 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: XL-постановка НЕ в батче (только day, в доске — можно)",
      "XL-постановка" not in ctx.split("ДОСКА ЗАДАЧ")[0].split("ТЕКУЩАЯ РАБОТА")[1]
      if "ТЕКУЩАЯ РАБОТА" in ctx else True, "t-e в батче")
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"

    # t-261015000631-13aef0: клейм теперь условный и атомарный (es.claim_task_if_ready);
    # стаб без __getattr__-заглушки: всегда успешен, как готовая задача
    def claim_task_if_ready(self, task_id, claim):
        written.append(("TaskClaimed", task_id))
        return ("x", 1)


class _FakeRunner:
    @staticmethod
    def run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None,
                     circle=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"
# hermeticity: протухшая /tmp-БД (подписи под старым ключом) валила аудит — чистим
for _p in ("/tmp/simbiot_batch_e2e.db", "/tmp/simbiot_batch_e2e.db-wal", "/tmp/simbiot_batch_e2e.db-shm"):
    if os.path.exists(_p):
        os.unlink(_p)
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)