Andrey Epifancev 07357732da Хвост пакета со спринтом — refresh по JQL проекта (#108)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-24 19:23:46 +04:00

438 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Очередь команд: пакеты, выдача агенту по одной, отчёты, отмена, повтор, сверка с зеркалом. Механика — здесь; смысл
типов команд (проверки постановки, паузы, журнал синхронизаций, отчёты) даёт модуль через CommandHooks (hub-modules,
решение 4). Новых команд хаб сам не создаёт: повтор после failed — новый пакет по запросу человека. Команду, выданную
агенту и оставшуюся без отчёта, хаб довыдаёт при следующем claim — это доставка той же выдачи (bridge-reliability,
решение 1)."""
import json
import logging
from datetime import timedelta
from . import backup as _backup # noqa: F401 (имя модуля занято параметром backup в claim)
from . import db, registry, settings
from .errors import CommandError
log = logging.getLogger(__name__)
TERMINAL = {"done", "failed", "skipped", "cancelled", "confirmed_by_state"}
MAX_DELIVERIES = 3 # выдач без отчёта, после которых команда — failed (bridge-reliability, решение 2)
ERROR_LIMIT = 1500 # символов ошибки в command.error (spec commands)
ONLINE_MINUTES = 5
class ClaimedError(Exception):
def __init__(self, claimed_at):
super().__init__(f"команда взята агентом {claimed_at}; отмена — с force")
self.claimed_at = claimed_at
class NotClaimedError(Exception):
"""Отчёт по команде, которую агенту не выдавали (status = queued) — API отвечает 409."""
def _cmd(row) -> dict:
d = dict(row)
d["params"] = json.loads(d["params"])
d["result"] = json.loads(d["result"]) if d.get("result") else None
return d
def get_command(conn, command_id) -> dict | None:
row = conn.execute("SELECT * FROM command WHERE id = ?", (command_id,)).fetchone()
return _cmd(row) if row else None
def _dump(value) -> str:
return json.dumps(value, ensure_ascii=False)
def _hooks(ctype):
"""Обработчики модуля, владеющего типом команды; None — тип никому не известен."""
m = registry.module_of(ctype)
return m.commands if m else None
def _read_types() -> set[str]:
return {t for m in registry.modules() if m.commands for t in m.commands.read_types}
# --- пакеты -----------------------------------------------------------------
def _batch_hooks(cmds):
"""Обработчики модуля пакета. Типы из разных модулей в одном пакете нельзя; ни одного известного типа — первый
модуль реестра: его validate выдаёт ошибку о неизвестном типе."""
owners = []
for c in cmds if isinstance(cmds, list) else []:
m = registry.module_of(c["type"]) if isinstance(c, dict) and isinstance(c.get("type"), str) else None
if m is not None and m not in owners:
owners.append(m)
if len(owners) > 1:
raise CommandError("команды разных модулей в одном пакете")
if owners:
return owners[0].commands
return next(m.commands for m in registry.modules() if m.commands)
def create_batch(conn, title, author, cmds, files_dir=None, project=None) -> dict:
h = _batch_hooks(cmds)
cmds = h.validate(cmds)
cmds = h.check_batch(conn, cmds, project, files_dir)
tail = h.tail(conn, cmds, project)
if tail is not None:
cmds.append(tail)
return insert_batch(conn, title, author, cmds, project)
def estimate(conn, types) -> int:
"""Оценка длительности, с: сумма пауз типов (паузу даёт модуль-владелец типа)."""
return sum(_pause(conn, t) for t in types)
def _pause(conn, ctype) -> int:
h = _hooks(ctype)
return int(h.pause(conn, ctype)) if h else 0
def insert_batch(conn, title, author, cmds, project=None) -> dict:
"""Вставка пакета без проверок и хвоста; транзакция своя (вложенная идёт через SAVEPOINT)."""
est = estimate(conn, [c["type"] for c in cmds])
with db.transaction(conn):
cur = conn.execute("INSERT INTO batch(title, author, created_at, estimate_seconds, project) VALUES (?, ?, ?, ?, ?)",
(title, author or "api", db.now(), est, project))
batch_id = cur.lastrowid
defined = {}
for seq, c in enumerate(cmds, 1):
h = _hooks(c["type"])
deps = [defined[k] for k in (h.temp_keys_used(c) if h else []) if k in defined]
cur = conn.execute("INSERT INTO command(batch_id, seq, type, params, depends_on) VALUES (?, ?, ?, ?, ?)",
(batch_id, seq, c["type"], _dump(c["params"]), deps[0] if deps else None))
if h and h.temp_definer and c["type"] == h.temp_definer[0]:
defined[c["params"][h.temp_definer[1]]] = cur.lastrowid
return {"id": batch_id, "estimate_seconds": est, "count": len(cmds)}
def _own_types(agent: str) -> list[str]:
"""Типы команд модуля агента (имя агента = имя модуля); у модуля без команд — пусто."""
m = registry.module(agent)
return sorted(m.commands.types) if m and m.commands else []
def _in(types: list[str]) -> str:
return ",".join("?" * len(types))
def has_active(conn, module: str | None = None) -> bool:
"""В очереди есть команда queued или claimed; module — только типы этого модуля, None — любого."""
if module is None:
return conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone() is not None
types = _own_types(module)
return conn.execute(f"SELECT 1 FROM command WHERE status IN ('queued', 'claimed') AND type IN ({_in(types)}) LIMIT 1",
types).fetchone() is not None
def agent_version(conn, agent: str):
"""Версия агента из его последней выдачи; None — агент не отчитывался."""
row = conn.execute("SELECT agent_version FROM agent_status WHERE agent = ?", (agent,)).fetchone()
return row["agent_version"] if row else None
def _status_of(cmds, cancelled_at=None) -> str:
sts = [c["status"] for c in cmds]
if not sts:
return "done"
if all(s in TERMINAL for s in sts):
return "cancelled" if cancelled_at else "done" # отменённый пакет — отменён, даже с частью done (T5.6)
if any(s == "claimed" for s in sts) or any(s in TERMINAL for s in sts):
return "running"
return "queued"
def batch_view(conn, batch_id, with_commands=True) -> dict | None:
b = conn.execute("SELECT * FROM batch WHERE id = ?", (batch_id,)).fetchone()
if b is None:
return None
cmds = [_cmd(r) for r in conn.execute("SELECT * FROM command WHERE batch_id = ? ORDER BY seq", (batch_id,))]
d = dict(b)
d["status"] = _status_of(cmds, b["cancelled_at"])
d["counts"] = {s: sum(1 for c in cmds if c["status"] == s) for s in sorted({c["status"] for c in cmds})}
first = registry.module_of(cmds[0]["type"]) if cmds else None
d["module"] = first.name if first else None
d["remaining_seconds"] = estimate(conn, [c["type"] for c in cmds if c["status"] in ("queued", "claimed")])
if with_commands:
d["commands"] = cmds
return d
def list_batches(conn, limit=20, project=None, module=None) -> list[dict]:
"""Последние пакеты (новые первыми); module — только пакеты этого модуля (по типу первой команды), фильтр
применяется до limit."""
where, args = ("WHERE project = ?", [project]) if project is not None else ("", [])
out = []
for r in conn.execute(f"SELECT id FROM batch {where} ORDER BY id DESC", args).fetchall():
if len(out) >= int(limit):
break
view = batch_view(conn, r["id"], with_commands=False)
if module is None or view["module"] == module:
out.append(view)
return out
# --- выдача -----------------------------------------------------------------
def _mapping(conn, batch_id, definer) -> dict:
"""temp_key → настоящий ключ по завершённым командам-определителям пакета."""
ctype, in_field, out_field = definer
rows = conn.execute(f"SELECT json_extract(params, '$.{in_field}') AS t, json_extract(result, '$.{out_field}') AS k "
"FROM command WHERE batch_id = ? AND type = ? AND status IN ('done', 'confirmed_by_state')",
(batch_id, ctype)).fetchall()
return {r["t"]: r["k"] for r in rows if r["k"]}
def _definer_status(conn, batch_id, definer, temp_key):
ctype, in_field, _ = definer
r = conn.execute(f"SELECT status FROM command WHERE batch_id = ? AND type = ? "
f"AND json_extract(params, '$.{in_field}') = ?", (batch_id, ctype, temp_key)).fetchone()
return r["status"] if r else "failed"
def _skip_reason(conn, batch_id, definer, unresolved) -> str:
"""Текст skipped для зависимой команды: отменённое создание называется отменой (O-10 #16, решение 10)."""
cancelled = [k for k in unresolved if _definer_status(conn, batch_id, definer, k) == "cancelled"]
return f"создание задачи {cancelled[0]} отменено" if cancelled else f"задача {unresolved[0]} не создана"
def _next(conn, agent: str) -> dict | None:
types = _own_types(agent)
rows = conn.execute("SELECT c.* FROM command c JOIN batch b ON b.id = c.batch_id "
f"WHERE c.status = 'queued' AND b.cancelled_at IS NULL AND c.type IN ({_in(types)}) "
"ORDER BY b.created_at, b.id, c.seq", types).fetchall()
for row in rows:
cmd = _cmd(row)
h = _hooks(cmd["type"])
needed = h.temp_keys_used(cmd) if h and h.temp_definer else []
if needed:
mapping = _mapping(conn, cmd["batch_id"], h.temp_definer)
unresolved = [k for k in needed if k not in mapping]
if any(_definer_status(conn, cmd["batch_id"], h.temp_definer, k) not in TERMINAL for k in unresolved):
return None # создание ещё не завершено — ждём (одна в полёте)
cmd["params"] = h.substitute(cmd["type"], cmd["params"], mapping)
unresolved = h.drop_unresolved(cmd, unresolved)
if unresolved:
conn.execute("UPDATE command SET status = 'skipped', finished_at = ?, error = ? WHERE id = ?",
(db.now(), _skip_reason(conn, cmd["batch_id"], h.temp_definer, unresolved), cmd["id"]))
continue
return cmd
return None
def _redeliver(conn, now, agent: str) -> dict | None:
"""Самая старая взятая команда без отчёта — выдать снова. Агент приходит за новой командой, только когда своей
незавершённой нет, значит выданную он не получил или потерял. После MAX_DELIVERIES выдач — failed. Берутся только
команды типов модуля агента: взятая другим агентом не довыдаётся."""
types = _own_types(agent)
while True:
row = conn.execute(f"SELECT * FROM command WHERE status = 'claimed' AND type IN ({_in(types)}) "
"ORDER BY claimed_at, id LIMIT 1", types).fetchone()
if row is None:
return None
cmd = _cmd(row)
if cmd["deliveries"] < MAX_DELIVERIES:
return cmd
conn.execute("UPDATE command SET status = 'failed', finished_at = ?, error = ? WHERE id = ?",
(now, f"агент не отчитался после {MAX_DELIVERIES} выдач", cmd["id"]))
h = _hooks(cmd["type"])
if h:
h.on_failed(conn, cmd, now)
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE current_command_id = ?", (cmd["id"],))
log.warning("команда %s: агент не отчитался после %s выдач — failed", cmd["id"], MAX_DELIVERIES)
def deliver_params(conn, cmd, files_dir) -> dict:
"""Параметры команды для агента: дописывает модуль-владелец типа (в базе params остаются без дописанного: повтор
команды проходит проверку как новая)."""
h = _hooks(cmd["type"])
return h.deliver_params(conn, cmd, files_dir) if h else cmd["params"]
def claim(conn, agent: str, agent_version=None, settings_version=None, last_call_at=None, backup=None,
files_dir=None) -> dict | None:
now = db.now()
to_backup = None
with db.transaction(conn):
conn.execute("UPDATE agent_status SET last_seen_at = ?, agent_version = COALESCE(?, agent_version), "
"settings_version = COALESCE(?, settings_version), last_call_at = COALESCE(?, last_call_at) WHERE agent = ?",
(now, None if agent_version is None else str(agent_version), settings_version, last_call_at, agent))
cmd = _redeliver(conn, now, agent)
if cmd is not None:
pause = _pause(conn, cmd["type"])
conn.execute("UPDATE command SET claimed_at = ?, pause_seconds = ?, deliveries = deliveries + 1 WHERE id = ?",
(now, pause, cmd["id"]))
h = _hooks(cmd["type"])
if h:
h.on_claim(conn, cmd, now, True)
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE agent = ?", (cmd["id"], agent))
log.info("команда %s выдана снова (выдача %s)", cmd["id"], cmd["deliveries"] + 1)
else:
cmd = _next(conn, agent)
if cmd is None:
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE agent = ?", (agent,))
return None
pause = _pause(conn, cmd["type"])
first = conn.execute("SELECT COUNT(*) AS n FROM command WHERE batch_id = ? AND status != 'queued'",
(cmd["batch_id"],)).fetchone()["n"] == 0
read = sorted(_read_types())
has_writes = conn.execute(f"SELECT 1 FROM command WHERE batch_id = ? AND type NOT IN ({','.join('?' * len(read))}) LIMIT 1",
(cmd["batch_id"], *read)).fetchone() is not None
conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, pause_seconds = ?, params = ?, deliveries = 1 "
"WHERE id = ?", (now, pause, _dump(cmd["params"]), cmd["id"]))
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE agent = ?", (cmd["id"], agent))
h = _hooks(cmd["type"])
if h:
h.on_claim(conn, cmd, now, False)
if first and has_writes:
to_backup = f"batch-{cmd['batch_id']}"
if to_backup and backup:
try:
backup(to_backup)
except Exception: # сбой копии не должен держать очередь: команда уже взята
log.warning("резервная копия %s не создана, команда %s выдаётся без неё", to_backup, cmd["id"], exc_info=True)
return {"id": cmd["id"], "type": cmd["type"], "params": deliver_params(conn, cmd, files_dir), "pause_seconds": pause,
"batch_id": cmd["batch_id"]}
# --- отчёты -----------------------------------------------------------------
def report_result(conn, command_id, status, result=None, error=None, jira_calls=0, rate_limited=0, files_dir=None) -> dict:
if status not in ("done", "failed"):
raise ValueError("status — done или failed")
error = None if error is None else str(error)[:ERROR_LIMIT]
early = get_command(conn, command_id)
eh = _hooks(early["type"]) if early is not None and early["status"] == "claimed" else None
prepared = bool(eh and eh.before_result)
completed = False
try:
if prepared: # вне транзакции отчёта: модуль готовит тяжёлое, не держа блокировку базы
try:
eh.before_result(conn, early, status, result, files_dir)
except ValueError as e: # не путать с «status — done или failed» (422 bad_status в API)
raise RuntimeError(f"before_result: {e}") from e
with db.transaction(conn):
cmd = get_command(conn, command_id)
if cmd is None:
raise KeyError(command_id)
if cmd["status"] == "queued":
raise NotClaimedError(f"команда {command_id} не выдана")
h = _hooks(cmd["type"])
if cmd["status"] in TERMINAL:
# поздний отчёт по команде, которую хаб уже закрыл (force-отмена, предел выдач): модуль может принять
# данные, статус не трогаем
if h and h.on_late_result(conn, cmd, status, result, jira_calls, rate_limited):
return get_command(conn, command_id)
return cmd # идемпотентно; терминальный статус не затирается
now = db.now()
if h:
status, why = h.check_result(conn, cmd, status, result, files_dir)
if why is not None:
error = why[:ERROR_LIMIT]
conn.execute("UPDATE command SET status = ?, finished_at = ?, result = ?, error = ?, jira_calls = ?, rate_limited = ? WHERE id = ?",
(status, now, _dump(result) if result is not None else None, error, int(jira_calls or 0), int(rate_limited or 0), command_id))
conn.execute("UPDATE agent_status SET current_command_id = NULL, last_seen_at = ? WHERE current_command_id = ?",
(now, command_id))
if h:
h.on_result(conn, cmd, status, result, jira_calls, rate_limited, now, files_dir)
if h:
h.after_result(conn, cmd, status)
completed = True
return get_command(conn, command_id)
finally:
if prepared and not completed: # ранний выход или исключение: модуль снимает подготовку (abort_result)
ah = eh.abort_result
if ah is not None:
try:
ah(conn, early)
except Exception: # noqa: BLE001
log.exception("команда %s: abort_result", command_id)
def reconcile(conn) -> list[int]:
"""Взятые команды без ответа, чей результат уже виден в зеркале → confirmed_by_state."""
changed = []
for row in conn.execute("SELECT * FROM command WHERE status = 'claimed'").fetchall():
cmd = _cmd(row)
h = _hooks(cmd["type"])
if h is None:
continue
ok, result = h.check_confirmed(conn, cmd)
if not ok:
continue
with db.transaction(conn):
cur = conn.execute("UPDATE command SET status = 'confirmed_by_state', finished_at = ?, result = ? "
"WHERE id = ? AND status = 'claimed'",
(db.now(), _dump(result) if result else None, cmd["id"]))
if cur.rowcount == 0: # отчёт агента или отмена опередили сверку
continue
changed.append(cmd["id"])
return changed
# --- отмена и повтор ----------------------------------------------------------
def cancel_batch(conn, batch_id) -> dict:
with db.transaction(conn):
if conn.execute("SELECT 1 FROM batch WHERE id = ?", (batch_id,)).fetchone() is None:
raise KeyError(batch_id)
now = db.now()
# завершённый пакет (нет queued/claimed) отменённым не становится: отменять нечего
conn.execute("UPDATE batch SET cancelled_at = ? WHERE id = ? AND cancelled_at IS NULL AND EXISTS "
"(SELECT 1 FROM command WHERE batch_id = ? AND status IN ('queued', 'claimed'))", (now, batch_id, batch_id))
conn.execute("UPDATE command SET status = 'cancelled', finished_at = ? WHERE batch_id = ? AND status = 'queued'", (now, batch_id))
view = batch_view(conn, batch_id)
# взятые агентом команды отмена пакета не останавливает: они будут выполнены (M-11 #12)
view["still_claimed"] = [{"id": c["id"], "type": c["type"]} for c in view["commands"] if c["status"] == "claimed"]
return view
def cancel_command(conn, command_id, force=False) -> dict:
with db.transaction(conn):
cmd = get_command(conn, command_id)
if cmd is None:
raise KeyError(command_id)
if cmd["status"] == "claimed" and not force:
raise ClaimedError(cmd["claimed_at"])
if cmd["status"] in ("queued", "claimed"):
conn.execute("UPDATE command SET status = 'cancelled', finished_at = ? WHERE id = ?", (db.now(), command_id))
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE current_command_id = ?", (command_id,))
return get_command(conn, command_id)
def retry_command(conn, command_id, author, files_dir=None, project=None) -> dict:
cmd = get_command(conn, command_id)
if cmd is None:
raise KeyError(command_id)
origin = conn.execute("SELECT project FROM batch WHERE id = ?", (cmd["batch_id"],)).fetchone()["project"]
if project is not None and project != origin:
raise CommandError(f"команда #{command_id} не из проекта {project}")
h = _hooks(cmd["type"])
params = h.retry_params(cmd) if h else cmd["params"]
return create_batch(conn, f"Повтор команды {command_id}", author, [{"type": cmd["type"], "params": params}],
files_dir=files_dir, project=origin)
# --- агент и автообновление ----------------------------------------------------
def agent_view(conn, agent: str) -> dict:
d = dict(conn.execute("SELECT * FROM agent_status WHERE agent = ?", (agent,)).fetchone())
d["module"] = agent
types = _own_types(agent)
now = db.utcnow()
d["online"] = bool(d["last_seen_at"]) and now - db.parse(d["last_seen_at"]) < timedelta(minutes=ONLINE_MINUTES)
m = registry.module(agent)
min_version = int(settings.get(conn, m.agent_min_version_key) or 0) if m and m.agent_min_version_key else 0
d["warning"] = "обновите агента" if d["agent_version"] and int(d["agent_version"]) < min_version else None
d["rate_limited_24h"] = conn.execute(
f"SELECT COALESCE(SUM(rate_limited), 0) AS n FROM command WHERE finished_at >= ? AND type IN ({_in(types)})",
((now - timedelta(days=1)).isoformat(), *types)).fetchone()["n"]
d["redelivered"] = [dict(r) for r in conn.execute(
"SELECT id, type, claimed_at, deliveries FROM command WHERE status = 'claimed' AND deliveries > 1 "
f"AND type IN ({_in(types)}) ORDER BY id", types)]
return d