"""Очередь команд: пакеты, выдача агенту по одной, отчёты, отмена, повтор, сверка с зеркалом, автообновление. Новых команд хаб сам не создаёт: повтор после failed — новый пакет по запросу человека. Команду, выданную агенту и оставшуюся без отчёта, хаб довыдаёт при следующем claim — это доставка той же выдачи (bridge-reliability, решение 1).""" import json import logging from datetime import timedelta from . import backup as _backup # noqa: F401 (имя модуля занято параметром backup в claim) from . import commands as C from . import db, mirror, settings 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 create_batch(conn, title, author, cmds) -> dict: cmds = C.validate(cmds) keys = [] for c in cmds: keys += [k for k in C.referenced_keys(c) if k not in keys] if any(c["type"] != "refresh" for c in cmds) and keys: cmds.append({"type": "refresh", "params": {"keys": keys}}) est = C.estimate([c["type"] for c in cmds], settings.pacing(conn)) with db.transaction(conn): cur = conn.execute("INSERT INTO batch(title, author, created_at, estimate_seconds) VALUES (?, ?, ?, ?)", (title, author or "api", db.now(), est)) batch_id = cur.lastrowid defined = {} for seq, c in enumerate(cmds, 1): deps = [defined[k] for k in C.temp_keys_used(c) 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 c["type"] == "ensure_issue": defined[c["params"]["temp_key"]] = cur.lastrowid return {"id": batch_id, "estimate_seconds": est, "count": len(cmds)} 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,))] pacing = settings.pacing(conn) 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})} d["remaining_seconds"] = C.estimate([c["type"] for c in cmds if c["status"] in ("queued", "claimed")], pacing) if with_commands: d["commands"] = cmds return d def list_batches(conn, limit=20) -> list[dict]: ids = [r["id"] for r in conn.execute("SELECT id FROM batch ORDER BY id DESC LIMIT ?", (int(limit),))] return [batch_view(conn, i, with_commands=False) for i in ids] # --- выдача ----------------------------------------------------------------- def _mapping(conn, batch_id) -> dict: """temp_key → настоящий ключ по завершённым ensure_issue пакета.""" rows = conn.execute("SELECT json_extract(params, '$.temp_key') AS t, json_extract(result, '$.key') AS k FROM command " "WHERE batch_id = ? AND type = 'ensure_issue' AND status IN ('done', 'confirmed_by_state')", (batch_id,)).fetchall() return {r["t"]: r["k"] for r in rows if r["k"]} def _definer_status(conn, batch_id, temp_key): r = conn.execute("SELECT status FROM command WHERE batch_id = ? AND type = 'ensure_issue' " "AND json_extract(params, '$.temp_key') = ?", (batch_id, temp_key)).fetchone() return r["status"] if r else "failed" def _next(conn) -> dict | None: rows = conn.execute("SELECT c.* FROM command c JOIN batch b ON b.id = c.batch_id " "WHERE c.status = 'queued' AND b.cancelled_at IS NULL ORDER BY b.created_at, b.id, c.seq").fetchall() for row in rows: cmd = _cmd(row) needed = C.temp_keys_used(cmd) if needed: mapping = _mapping(conn, cmd["batch_id"]) unresolved = [k for k in needed if k not in mapping] if any(_definer_status(conn, cmd["batch_id"], k) not in TERMINAL for k in unresolved): return None # создание ещё не завершено — ждём (одна в полёте) cmd["params"] = C.substitute(cmd["type"], cmd["params"], mapping) if cmd["type"] == "refresh": cmd["params"]["keys"] = [k for k in cmd["params"].get("keys", []) if not C.is_temp(k)] unresolved = [] if cmd["params"]["keys"] else unresolved if unresolved: conn.execute("UPDATE command SET status = 'skipped', finished_at = ?, error = ? WHERE id = ?", (db.now(), f"задача {unresolved[0]} не создана", cmd["id"])) continue return cmd return None def _redeliver(conn, now) -> dict | None: """Самая старая взятая команда без отчёта — выдать снова. Агент приходит за новой командой, только когда своей незавершённой нет, значит выданную он не получил или потерял. После MAX_DELIVERIES выдач — failed.""" while True: row = conn.execute("SELECT * FROM command WHERE status = 'claimed' ORDER BY claimed_at, id LIMIT 1").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"])) conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1 AND current_command_id = ?", (cmd["id"],)) log.warning("команда %s: агент не отчитался после %s выдач — failed", cmd["id"], MAX_DELIVERIES) def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=None, backup=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_jira_call_at = COALESCE(?, last_jira_call_at) WHERE id = 1", (now, None if agent_version is None else str(agent_version), settings_version, last_jira_call_at)) cmd = _redeliver(conn, now) if cmd is not None: pause = int(settings.pacing(conn).get(cmd["type"], 0)) conn.execute("UPDATE command SET claimed_at = ?, pause_seconds = ?, deliveries = deliveries + 1 WHERE id = ?", (now, pause, cmd["id"])) conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],)) log.info("команда %s выдана снова (выдача %s)", cmd["id"], cmd["deliveries"] + 1) else: cmd = _next(conn) if cmd is None: conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1") return None pause = int(settings.pacing(conn).get(cmd["type"], 0)) first = conn.execute("SELECT COUNT(*) AS n FROM command WHERE batch_id = ? AND status != 'queued'", (cmd["batch_id"],)).fetchone()["n"] == 0 has_writes = conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type != 'refresh' LIMIT 1", (cmd["batch_id"],)).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 id = 1", (cmd["id"],)) if cmd["type"] == "refresh": p = cmd["params"] conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, ?, ?, ?)", (now, "jql" if p.get("jql") else "keys", p.get("jql") or ",".join(p.get("keys", [])), cmd["id"])) 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": cmd["params"], "pause_seconds": pause, "batch_id": cmd["batch_id"]} # --- отчёты ----------------------------------------------------------------- def is_full_refresh(conn, command_id) -> bool: """Команда — refresh по базовой JQL (полный прогон): только он вправе сбрасывать флаг neighbor.""" if not command_id: return False cmd = get_command(conn, command_id) return bool(cmd and cmd["type"] == "refresh" and cmd["params"].get("jql") and cmd["params"]["jql"] == settings.get(conn, "base_jql")) def record_state(conn, command_id, issues_count, links_count) -> None: conn.execute("UPDATE sync_run SET issues_count = issues_count + ?, links_count = links_count + ? WHERE command_id = ?", (int(issues_count), int(links_count), command_id)) def report_result(conn, command_id, status, result=None, error=None, jira_calls=0, rate_limited=0) -> dict: if status not in ("done", "failed"): raise ValueError("status — done или failed") error = None if error is None else str(error)[:ERROR_LIMIT] 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} не выдана") if cmd["status"] in TERMINAL: # поздний отчёт о созданной задаче по команде, которую хаб уже закрыл (force-отмена — M-5, предел выдач — # M-4 #12): ключ не теряем, статус не трогаем if (cmd["status"] in ("cancelled", "failed") and cmd["type"] == "ensure_issue" and status == "done" and result and result.get("key") and not cmd["result"]): conn.execute("UPDATE command SET result = ?, jira_calls = ?, rate_limited = ? WHERE id = ?", (_dump(result), int(jira_calls or 0), int(rate_limited or 0), command_id)) mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"]) log.warning("команда %s уже %s, но агент создал %s — ключ записан в результат", command_id, cmd["status"], result["key"]) return get_command(conn, command_id) return cmd # идемпотентно; cancelled не затирается now = db.now() 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 id = 1 AND current_command_id = ?", (now, command_id)) if cmd["type"] == "ensure_issue" and status == "done" and result and result.get("key"): mirror.rename_note(conn, cmd["params"]["temp_key"], result["key"]) if cmd["type"] == "refresh": conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?", (now, int(jira_calls or 0), int(rate_limited or 0), command_id)) if status == "done" and is_full_refresh(conn, command_id): # полный прогон не обновил задачу — в базовую JQL она больше не входит: сосед. Пустой прогон (0 задач) — # почти всегда сбой выборки, а не «все задачи ушли»: соседей не трогаем (M-12) run = conn.execute("SELECT started_at, issues_count FROM sync_run WHERE command_id = ?", (command_id,)).fetchone() if run and run["issues_count"] > 0: conn.execute("UPDATE issue SET neighbor = 1 WHERE fetched_at < ?", (run["started_at"],)) if cmd["type"] == "refresh" and status == "done": reconcile(conn) return get_command(conn, command_id) def reconcile(conn) -> list[int]: """Взятые команды без ответа, чей результат уже виден в зеркале → confirmed_by_state.""" link_types = settings.get(conn, "link_types") or {} changed = [] for row in conn.execute("SELECT * FROM command WHERE status = 'claimed'").fetchall(): cmd = _cmd(row) ok, result = C.check_confirmed(conn, cmd, link_types) 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() conn.execute("UPDATE batch SET cancelled_at = ? WHERE id = ? AND cancelled_at IS NULL", (now, batch_id)) conn.execute("UPDATE command SET status = 'cancelled', finished_at = ? WHERE batch_id = ? AND status = 'queued'", (now, batch_id)) return batch_view(conn, batch_id) 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 id = 1 AND current_command_id = ?", (command_id,)) return get_command(conn, command_id) def retry_command(conn, command_id, author) -> dict: cmd = get_command(conn, command_id) if cmd is None: raise KeyError(command_id) return create_batch(conn, f"Повтор команды {command_id}", author, [{"type": cmd["type"], "params": cmd["params"]}]) # --- агент и автообновление ---------------------------------------------------- def agent_view(conn) -> dict: d = dict(conn.execute("SELECT * FROM agent_status WHERE id = 1").fetchone()) now = db.utcnow() d["online"] = bool(d["last_seen_at"]) and now - db.parse(d["last_seen_at"]) < timedelta(minutes=ONLINE_MINUTES) min_version = int(settings.get(conn, "min_agent_version") or 0) d["warning"] = "обновите агента" if d["agent_version"] and int(d["agent_version"]) < min_version else None d["rate_limited_24h"] = conn.execute("SELECT COALESCE(SUM(rate_limited), 0) AS n FROM command WHERE finished_at >= ?", ((now - timedelta(days=1)).isoformat(),)).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 ORDER BY id")] return d def maybe_auto_refresh(conn) -> dict | None: minutes = int(settings.get(conn, "auto_refresh_minutes") or 0) if minutes <= 0 or not agent_view(conn)["online"]: return None if conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone(): return None last = conn.execute("SELECT MAX(started_at) AS t FROM sync_run").fetchone()["t"] if last and db.utcnow() - db.parse(last) < timedelta(minutes=minutes): return None return create_batch(conn, "Автообновление", "hub", [{"type": "refresh", "params": {"jql": settings.get(conn, "base_jql")}}])