267 lines
No EOL
14 KiB
Python
267 lines
No EOL
14 KiB
Python
# -*- 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"
|
||
|
||
|
||
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"
|
||
# 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) |