#!/usr/bin/env python3 """jira-agent — клиент моста Jira для виртуалки с доступом к Jira банка. Опрашивает хаб (jira-hub) на Mac mini, выполняет команды «желаемый результат» в Jira по правилу «проверь, потом сделай», ведёт журнал на диске и возвращает хабу состояние задач. Только стандартная библиотека, Python 3.9. Спека: jira-hub/openspec/specs/agent/spec.md. Связь с хабом — HTTP (`hub_url = http://…`) или общая папка (`hub_url = folder:<путь>`), если сеть закрыта VPN. python3 agent.py check # доступ к хабу и Jira, без записи python3 agent.py run # основной цикл (по умолчанию) python3 agent.py selftest # самопроверка новой версии перед обновлением: версия + check; код 0 или 3 python3 agent.py prune # почистить journal.jsonl от подтверждённых записей старше 30 дней Самообновление — через запускатель run.py (спека agent, «Самообновление агента»). """ import argparse import base64 import fcntl import hashlib import http.client import json import os import re import signal import ssl import subprocess import sys import time import urllib.error import urllib.parse import urllib.request import uuid from datetime import datetime, timezone from typing import Any, Dict, List, Optional AGENT_VERSION = 5 EXIT_UPDATE = 75 # «обновление готово»: run.py подменит agent.py на agent.py.new HERE = os.path.dirname(os.path.abspath(__file__)) API = "/rest/api/2" AGILE = "/rest/agile/1.0" def now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() def log(msg: str) -> None: print("[%s] %s" % (datetime.now().strftime("%H:%M:%S"), msg), flush=True) def load_config(path: str) -> dict: try: with open(path, encoding="utf-8") as f: cfg = json.load(f) except FileNotFoundError: sys.exit("нет файла настроек %s — скопируйте agent.example.json в agent.json и заполните" % path) for key in ("hub_url", "hub_token", "jira"): if not cfg.get(key): sys.exit("в %s нет обязательного поля %s" % (path, key)) if not cfg["jira"].get("base_url"): sys.exit("в %s нет jira.base_url" % path) if cfg["hub_url"].startswith("folder:") and not os.path.isabs(cfg["hub_url"][len("folder:"):]): sys.exit("hub_url: после folder: нужен абсолютный путь общей папки, например " "folder:/Volumes/My Shared Files/RSHB/jira-bridge") cfg.setdefault("timeout", 60) cfg.setdefault("poll_seconds", 5) cfg.setdefault("verify_ssl", True) return cfg # --------------------------------------------------------------------------- # Самообновление: update.json рядом с agent.py (спека agent «Самообновление агента», design 7) # --------------------------------------------------------------------------- def read_update_state(path: str) -> Optional[dict]: """Состояние обновления; нет файла или мусор — None.""" try: with open(path, encoding="utf-8") as f: data = json.load(f) except FileNotFoundError: return None except (OSError, ValueError): log("update.json не читается — считаю, что состояния нет") return None return data if isinstance(data, dict) else None def write_update_state(path: str, state: str, from_version, to_version, sha256, error=None) -> dict: data = {"state": state, "from_version": from_version, "to_version": to_version, "sha256": sha256, "error": error, "at": now_iso()} tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) os.replace(tmp, path) return data # --------------------------------------------------------------------------- # HTTP # --------------------------------------------------------------------------- class HttpError(Exception): def __init__(self, code: int, text: str, headers=None): super().__init__("HTTP %d: %s" % (code, text[:300])) self.code, self.text, self.headers = code, text, headers or {} class Http: def __init__(self, base: str, headers: Dict[str, str], timeout: float = 60, ctx=None): self.base, self.headers, self.timeout, self.ctx = base.rstrip("/"), headers, timeout, ctx def __repr__(self): return "<%s %s>" % (self.__class__.__name__, self.base) def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any: url = self.base + path if params: url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, method=method) for k, v in self.headers.items(): req.add_header(k, v) req.add_header("Accept", "application/json") if data is not None: req.add_header("Content-Type", "application/json; charset=utf-8") try: with urllib.request.urlopen(req, timeout=timeout or self.timeout, context=self.ctx) as resp: raw = resp.read() status = resp.status except urllib.error.HTTPError as e: self._touch() try: headers = {k.lower(): v for k, v in e.headers.items()} text = e.read().decode("utf-8", "replace") except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err: raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None finally: e.close() raise HttpError(e.code, text, headers) from None except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e: raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None self._touch() try: return json.loads(raw.decode("utf-8")) if raw else {} except ValueError as e: # включая UnicodeDecodeError и JSONDecodeError raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, raw[:200].decode("utf-8", "replace"))) from None def _touch(self) -> None: """Вызывается на каждый полученный HTTP-ответ (любой код).""" class JiraClient(Http): """Все вызовы Jira идут здесь: счётчики вызовов и 429/503, время последнего вызова, повтор по Retry-After.""" def __init__(self, cfg: dict): j = cfg["jira"] auth = j.get("auth") or {} kind = auth.get("type") or ("bearer" if auth.get("token") else "basic") if kind == "bearer": if not auth.get("token"): sys.exit("jira.auth.token пуст") header = "Bearer " + auth["token"] else: if not (auth.get("user") and auth.get("password")): sys.exit("jira.auth.user / jira.auth.password пусты") header = "Basic " + base64.b64encode(("%s:%s" % (auth["user"], auth["password"])).encode()).decode() ctx = None if not cfg.get("verify_ssl", True): ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE elif cfg.get("ca_bundle"): ctx = ssl.create_default_context(cafile=cfg["ca_bundle"]) super().__init__(j["base_url"], {"Authorization": header, "X-Atlassian-Token": "no-check"}, cfg.get("timeout", 60), ctx) self.calls = 0 self.rate_limited = 0 self.last_call_at = None # type: Optional[float] (time.monotonic) self.last_call_iso = None # type: Optional[str] self.sleep = time.sleep def _touch(self) -> None: self.calls += 1 self.last_call_at, self.last_call_iso = time.monotonic(), now_iso() def reset_counters(self) -> None: self.calls, self.rate_limited = 0, 0 def call(self, method, path, body=None, params=None): for attempt in range(1, 6): try: return super().call(method, path, body, params) except HttpError as e: if e.code in (429, 503) and attempt < 5: wait = 60 * attempt try: wait = max(0, int(e.headers.get("retry-after"))) except (TypeError, ValueError): pass self.rate_limited += 1 log(" ! HTTP %d, жду %d с (попытка %d/5)" % (e.code, wait, attempt)) self.sleep(wait) continue if e.code in (429, 503): self.rate_limited += 1 raise def get(self, path, **params): return self.call("GET", path, params=params or None) def post(self, path, body): return self.call("POST", path, body) def put(self, path, body): return self.call("PUT", path, body) def delete(self, path): return self.call("DELETE", path) class FolderTransport: """Обмен с хабом через общую папку (hub_url = folder:<путь>), когда сеть до хаба закрыта (VPN на виртуалке). Запрос — файл req/.json, ответ кладёт реле хаба в resp/.json. Ошибки те же, что у Http.call: нет ответа к сроку или нет папки — ConnectionError, код не 2xx — HttpError. Спека: specs/agent, folder-transport.""" POLL = 0.1 BAD_JSON_WAIT = 2.0 def __init__(self, root: str, authorization: str): self.root, self.authorization = root, authorization def __repr__(self): return "" % self.root def call(self, method: str, path: str, body=None, timeout: float = 60) -> Any: if not os.path.isdir(self.root): raise ConnectionError("нет папки %s" % self.root) req_dir, resp_dir = os.path.join(self.root, "req"), os.path.join(self.root, "resp") rid = uuid.uuid4().hex try: os.makedirs(req_dir, exist_ok=True) os.makedirs(resp_dir, exist_ok=True) deadline = time.time() + timeout self._write(req_dir, rid + ".json", { "id": rid, "method": method, "path": path, "body": body, "authorization": self.authorization, "deadline": datetime.fromtimestamp(deadline, timezone.utc).isoformat()}) except OSError as e: raise ConnectionError("папка %s недоступна: %s" % (self.root, e)) from None resp_path = os.path.join(resp_dir, rid + ".json") end = time.monotonic() + timeout while time.monotonic() < end: if os.path.exists(resp_path): return self._take(resp_path) time.sleep(self.POLL) try: os.remove(os.path.join(req_dir, rid + ".json")) except OSError: pass raise ConnectionError("не ответил через папку %s за %.0f с" % (self.root, timeout)) @staticmethod def _write(directory: str, name: str, payload: dict) -> None: tmp = os.path.join(directory, "." + name + ".tmp") with open(tmp, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False) f.flush() os.fsync(f.fileno()) os.replace(tmp, os.path.join(directory, name)) def _take(self, path: str) -> Any: end = time.monotonic() + self.BAD_JSON_WAIT while True: try: with open(path, encoding="utf-8") as f: resp = json.load(f) if not isinstance(resp, dict): raise ValueError("ответ не объект") status, body = resp.get("status"), resp.get("body") if type(status) is not int or not (body is None or isinstance(body, str)): raise ValueError("неверные типы status/body") # bool — подкласс int, поэтому type(), не isinstance (M-2 agent#8) break except (ValueError, OSError): if time.monotonic() >= end: raise ConnectionError("нечитаемый ответ хаба в %s" % path) from None time.sleep(self.POLL) try: os.remove(path) except OSError: pass status, text = int(resp.get("status") or 0), resp.get("body") or "" if not 200 <= status < 300: raise HttpError(status, text) try: return json.loads(text) if text else {} except ValueError as e: raise HttpError(status, "ответ не JSON: %s; начало: %s" % (e, text[:200])) from None class HubClient(Http): def __init__(self, cfg: dict): auth = "Bearer " + cfg["hub_token"] super().__init__(cfg["hub_url"], {"Authorization": auth}, cfg.get("timeout", 60)) url = cfg["hub_url"] self.folder = FolderTransport(url[len("folder:"):], auth) if url.startswith("folder:") else None def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any: if self.folder is None: return super().call(method, path, body, params, timeout) if params: path += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) return self.folder.call(method, path, body, timeout or self.timeout) def claim(self, agent_version, settings_version, last_jira_call_at, update_state=None, launcher_version=None) -> dict: return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version, "last_jira_call_at": last_jira_call_at, "update_state": update_state, "launcher_version": launcher_version}, timeout=max(self.timeout, 45)) def release(self) -> dict: return self.call("GET", "/agent/release") def settings(self) -> dict: return self.call("GET", "/agent/settings") def command(self, command_id) -> dict: return self.call("GET", "/agent/commands/%d" % int(command_id)) def result(self, command_id, payload: dict) -> dict: return self.call("POST", "/agent/commands/%d/result" % int(command_id), payload) def state(self, payload: dict) -> dict: return self.call("POST", "/agent/state", payload) def known_keys(self) -> List[str]: return self.call("GET", "/agent/known-keys")["keys"] # --------------------------------------------------------------------------- # Журнал: только дозапись; started → finished → acked по id команды # --------------------------------------------------------------------------- class Journal: def __init__(self, path: str): self.path = path def _torn_tail(self) -> bool: """Файл не пуст и не заканчивается переводом строки (оборванная запись).""" try: if os.path.getsize(self.path) == 0: return False except FileNotFoundError: return False with open(self.path, "rb") as f: f.seek(-1, os.SEEK_END) return f.read(1) != b"\n" def append(self, record: dict) -> None: record = dict(record) record.setdefault("at", now_iso()) data = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8") if self._torn_tail(): data = b"\n" + data fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600) with os.fdopen(fd, "ab") as f: f.write(data) f.flush() os.fsync(f.fileno()) def _scan(self): """Возвращает (годные записи, номера битых строк). Годная — dict с command_id.""" good, bad = [], [] if not os.path.exists(self.path): return good, bad with open(self.path, "rb") as f: for n, raw in enumerate(f, 1): line = raw.strip() if not line: continue try: r = json.loads(line.decode("utf-8")) except ValueError: r = None if isinstance(r, dict) and "command_id" in r: good.append(r) else: bad.append(n) return good, bad def records(self) -> List[dict]: good, bad = self._scan() for n in bad: log("журнал: строка %d не читается, пропускаю" % n) return good def last_at(self) -> Optional[datetime]: """Самое позднее время записи журнала (UTC). Записи без времени или с непонятным временем пропускаются.""" latest = None for r in self.records(): try: t = datetime.fromisoformat(r["at"]) except (KeyError, TypeError, ValueError): continue if t.tzinfo is None: continue if latest is None or t > latest: latest = t return latest def claimed(self, command_id) -> None: """Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту.""" self.append({"kind": "claimed", "command_id": int(command_id)}) def started(self, command_id) -> None: self.append({"kind": "started", "command_id": int(command_id)}) def finished(self, command_id, status, result, error, jira_calls, rate_limited) -> None: self.append({"kind": "finished", "command_id": int(command_id), "status": status, "result": result, "error": error, "jira_calls": int(jira_calls or 0), "rate_limited": int(rate_limited or 0)}) def acked(self, command_id) -> None: self.append({"kind": "acked", "command_id": int(command_id)}) def unacked_finished(self) -> List[dict]: last, acked = {}, set() for r in self.records(): if r.get("kind") == "finished": last[r["command_id"]] = r acked.discard(r["command_id"]) elif r.get("kind") == "acked": acked.add(r["command_id"]) return [r for cid, r in last.items() if cid not in acked] def unfinished_started(self) -> List[int]: """Порядковый учёт: claimed/started открывают id, finished/acked закрывает; порядок — первое открытие.""" opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия for r in self.records(): kind, cid = r.get("kind"), r["command_id"] if kind in ("claimed", "started"): opened[cid] = True elif kind in ("finished", "acked") and cid in opened: opened[cid] = False return [cid for cid, is_open in opened.items() if is_open] def prune(self, days: int = 30) -> int: """Удалить записи id до его последнего acked, если тот старше days дней (возраст — по acked) и после него у id нет started/finished. Возвращает число удалённых годных записей; битые строки отбрасываются при перезаписи и учитываются в логе.""" good, bad = self._scan() if bad: log("журнал: удаляю битых строк %d" % len(bad)) cutoff = datetime.now(timezone.utc).timestamp() - days * 86400 last_ack, last_open = {}, {} for i, r in enumerate(good): cid = r["command_id"] if r.get("kind") == "acked": last_ack[cid] = i elif r.get("kind") in ("claimed", "started", "finished"): last_open[cid] = i drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется for cid, i in last_ack.items(): if last_open.get(cid, -1) > i: continue try: ts = _parse_at(good[i].get("at")) except (ValueError, TypeError): continue if ts < cutoff: drop_upto[cid] = i keep = [r for i, r in enumerate(good) if not (r["command_id"] in drop_upto and i <= drop_upto[r["command_id"]])] tmp = self.path + ".tmp" try: os.remove(tmp) except FileNotFoundError: pass fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "wb") as f: for r in keep: f.write((json.dumps(r, ensure_ascii=False) + "\n").encode("utf-8")) f.flush() os.fsync(f.fileno()) os.replace(tmp, self.path) return len(good) - len(keep) def _parse_at(value) -> float: """Метка времени журнала в эпоху; наивное время трактуется как UTC.""" dt = datetime.fromisoformat(value) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.timestamp() # --------------------------------------------------------------------------- # Исполнитель: команда — желаемый результат; прочитать → сравнить → записать разницу # --------------------------------------------------------------------------- def _date(value) -> Optional[str]: """Даты Jira (duedate — YYYY-MM-DD, кастомные — с временем) сравниваем по дню.""" return (value or "")[:10] or None def _nl(value) -> str: r"""Текст для сравнения: Jira хранит правленное в UI с \r\n, команды приходят с \n (T3.9, проверено 13.10.2026).""" return (value or "").replace("\r\n", "\n") class ConfigGap(Exception): """В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле.""" class Executor: def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None): self.jira, self.s, self.hub = jira, settings, hub self.sleep = time.sleep self._symmetric = None self._sprints = None def run(self, cmd: dict) -> dict: self.jira.reset_counters() self._sprints = None handler = getattr(self, "do_" + str(cmd.get("type")), None) if handler is None: return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type")) try: return handler(cmd.get("params") or {}, cmd) except ConfigGap as e: return self._out("failed", error=str(e)) except HttpError as e: out = self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500])) out["retryable"] = e.code >= 500 # исход неизвестен: Runner повторит через проверку (R-10) return out except ConnectionError as e: out = self._out("failed", error=str(e)) out["retryable"] = True return out def _out(self, status, result=None, error=None) -> dict: return {"status": status, "result": result, "error": error, "jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited} def gap(self, cmd_type: str) -> None: """Зазор между записями внутри одной команды — по политике хаба для этого типа.""" seconds = int((self.s.get("pacing") or {}).get(cmd_type, 0) or 0) if seconds > 0: log(" пауза %d с" % seconds) self.sleep(seconds) def fid(self, logical: str) -> str: return (self.s.get("field_ids") or {}).get(logical, logical) def has_field(self, logical: str) -> bool: return logical in ("summary", "description", "priority", "duedate", "labels") or bool((self.s.get("field_ids") or {}).get(logical)) def priority_value(self, name: str) -> dict: mapped = (self.s.get("priority_map") or {}).get(name, name) return {self.s.get("priority_value_key") or "name": mapped} def issue_type_ref(self, kind: str) -> dict: it = (self.s.get("issue_types") or {}).get(kind) or {} return {"id": str(it["id"])} if it.get("id") else {"name": it.get("name")} def epic_field_id(self) -> str: ep = self.s.get("epic_link") or {} if ep.get("mode", "field") != "field": raise ConfigGap("epic_key передан, но epic_link.mode = %r (поддержан только «field»)" % ep.get("mode")) if not ep.get("field_id"): raise ConfigGap("epic_key передан, но не настроено epic_link.field_id") return ep["field_id"] def epic_fields(self, epic_key) -> dict: return {self.epic_field_id(): epic_key} if epic_key else {} def need_field(self, logical: str) -> str: if not self.has_field(logical): raise ConfigGap("поле передано, но не настроено field_ids.%s" % logical) return self.fid(logical) # --- ensure_issue ---------------------------------------------------------- def do_ensure_issue(self, p: dict, cmd: dict) -> dict: summary = (p["summary"] or "").strip() jql = "project = %s AND reporter = currentUser() AND created >= -7d ORDER BY created DESC" % self.s["project_key"] start = 0 while start < 1000: found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200, startAt=start) issues = found.get("issues", []) for it in issues: if (it.get("fields", {}).get("summary") or "").strip() == summary: return self._out("done", {"key": it["key"], "existed": True}) start += len(issues) if not issues or start >= int(found.get("total") or 0): break fields = {"project": {"key": self.s["project_key"]}, "issuetype": self.issue_type_ref(p["issue_type"]), "summary": p["summary"]} if p.get("description") is not None: fields[self.fid("description")] = p["description"] if p.get("priority"): fields[self.fid("priority")] = self.priority_value(p["priority"]) if p.get("duedate"): fields[self.fid("duedate")] = p["duedate"] for logical in ("planned_start", "planned_end"): if p.get(logical): fields[self.need_field(logical)] = p[logical] if p.get("labels") is not None: fields[self.fid("labels")] = list(p["labels"]) if p.get("assignee"): fields["assignee"] = {"name": p["assignee"]} fields.update(self.epic_fields(p.get("epic_key"))) if p["issue_type"] == "epic" and self.s.get("epic_name_field_id"): fields[self.s["epic_name_field_id"]] = p["summary"] created = self.jira.post(API + "/issue", {"fields": fields}) return self._out("done", {"key": created["key"], "created": True}) # --- ensure_fields --------------------------------------------------------- def do_ensure_fields(self, p: dict, cmd: dict) -> dict: key = p["key"] cur = self.jira.get(API + "/issue/%s" % key)["fields"] fields, changed = {}, [] if "summary" in p and (cur.get("summary") or "") != (p["summary"] or ""): fields["summary"] = p["summary"] changed.append("summary") if "description" in p and _nl(cur.get(self.fid("description"))) != _nl(p["description"]): fields[self.fid("description")] = p["description"] changed.append("description") if "priority" in p: want = self.priority_value(p["priority"]) vkey = list(want)[0] have = (cur.get(self.fid("priority")) or {}).get(vkey) if (None if have is None else str(have)) != str(want[vkey]): fields[self.fid("priority")] = want changed.append("priority") for logical in ("duedate", "planned_start", "planned_end"): if logical not in p: continue if logical != "duedate" and p[logical] and not self.has_field(logical): self.need_field(logical) if self.has_field(logical) and _date(cur.get(self.fid(logical))) != _date(p[logical]): fields[self.fid(logical)] = p[logical] changed.append(logical) if "epic_key" in p and (p["epic_key"] or (self.s.get("epic_link") or {}).get("field_id")): fid = self.epic_field_id() if (cur.get(fid) or None) != (p["epic_key"] or None): fields[fid] = p["epic_key"] changed.append("epic_key") if "assignee" in p and ((cur.get("assignee") or {}).get("name") or None) != (p["assignee"] or None): fields["assignee"] = {"name": p["assignee"]} if p["assignee"] else None changed.append("assignee") if "labels_add" in p or "labels_remove" in p: labels = list(cur.get(self.fid("labels")) or []) remove = set(p.get("labels_remove") or []) new = [l for l in labels if l not in remove] + [l for l in (p.get("labels_add") or []) if l not in labels] if new != labels: fields[self.fid("labels")] = new changed.append("labels") if not fields: return self._out("done", {"changed": []}) self.jira.put(API + "/issue/%s" % key, {"fields": fields}) return self._out("done", {"changed": changed}) # --- ensure_comment -------------------------------------------------------- def do_ensure_comment(self, p: dict, cmd: dict) -> dict: key, marker, body = p["key"], p["marker"], p["body"] heads = (marker, "*%s.*" % marker) # маркер — только в начале тела (формат агента: «*маркер.* текст») if not body.startswith(heads): body = "*%s.* %s" % (marker, body) existing, start = None, 0 while existing is None: page = self.jira.get(API + "/issue/%s/comment" % key, startAt=start, maxResults=100) comments = page.get("comments", []) for c in comments: if (c.get("body") or "").startswith(heads): existing = c break start += len(comments) if not comments or page.get("total") is None or start >= int(page["total"]): break if existing and _nl(existing.get("body")) == _nl(body): return self._out("done", {"comment_id": existing["id"], "changed": False}) if existing: self.jira.put(API + "/issue/%s/comment/%s" % (key, existing["id"]), {"body": body}) return self._out("done", {"comment_id": existing["id"], "changed": True}) c = self.jira.post(API + "/issue/%s/comment" % key, {"body": body}) return self._out("done", {"comment_id": c.get("id"), "created": True}) # --- ensure_status --------------------------------------------------------- @staticmethod def _pick_transition(trs: list, step: str) -> Optional[dict]: """Переход, ведущий в статус step. При неоднозначности предпочитается он; имя перехода — запасной вариант.""" for t in trs: if (t.get("to") or {}).get("name") == step: return t for t in trs: if t.get("name") == step: return t return None def do_ensure_status(self, p: dict, cmd: dict) -> dict: key, target = p["key"], p["target"] if isinstance(target, str): path = [target] elif isinstance(target, list) and all(isinstance(s, str) for s in target): path = list(target) else: return self._out("failed", error="target должен быть строкой или списком строк") if not path or not all(path): return self._out("failed", error="пустой путь статусов") final = path[-1] cur = self.jira.get(API + "/issue/%s" % key, fields="status")["fields"]["status"]["name"] start = cur if cur == final: return self._out("done", {"from": start, "to": final, "steps": 0}) if cur in path: path = path[path.index(cur) + 1:] steps = 0 for step in path: if cur == step: continue # повтор того же статуса подряд в пути: перехода не нужно trs = self.jira.get(API + "/issue/%s/transitions" % key).get("transitions", []) hit = self._pick_transition(trs, step) if hit is None: avail = ", ".join("%s→%s" % (t.get("name"), (t.get("to") or {}).get("name")) for t in trs) return self._out("failed", {"from": start, "reached": cur}, "нет перехода из «%s» в «%s»; доступны: %s" % (cur, step, avail or "—")) if steps: self.gap("ensure_status") self.jira.post(API + "/issue/%s/transitions" % key, {"transition": {"id": hit["id"]}}) cur = (hit.get("to") or {}).get("name") or step steps += 1 return self._out("done", {"from": start, "to": cur, "steps": steps}) # --- ensure_link ----------------------------------------------------------- def symmetric_types(self) -> set: """Типы, у которых inward == outward («Связанные»): направление не важно. Ошибка HTTP пробрасывается наружу и не кэшируется: следующая команда попробует снова.""" if self._symmetric is None: types = self.jira.get(API + "/issueLinkType").get("issueLinkTypes", []) self._symmetric = {t["name"] for t in types if t.get("inward") == t.get("outward")} return self._symmetric def do_ensure_link(self, p: dict, cmd: dict) -> dict: src, dst = p["source"], p["target"] types = self.s.get("link_types") or {} # намеренный fallback (рулинг R-5): имя, которого нет в link_types, уходит в Jira как есть lt = types.get(p["link_type"], p["link_type"]) managed = set(types.values()) | {lt} symmetric = self.symmetric_types() raw = self.jira.get(API + "/issue/%s" % src, fields="issuelinks")["fields"].get("issuelinks") or [] links = [] # (тип, другая задача, направление, id) for l in raw: t = (l.get("type") or {}).get("name") if l.get("outwardIssue"): links.append((t, l["outwardIssue"]["key"], "out", l.get("id"))) if l.get("inwardIssue"): links.append((t, l["inwardIssue"]["key"], "in", l.get("id"))) wanted = [x for x in links if x[1] == dst and x[0] == lt and (x[2] == "out" or lt in symmetric)] conflicts = [x for x in links if x[1] == dst and x[0] in managed and x not in wanted] if conflicts and not p.get("replace"): desc = ", ".join("«%s» (%s)" % (t, "обратное направление" if t == lt else "другой тип") for t, _, _, _ in conflicts) return self._out("failed", error="между %s и %s уже есть: %s; нужен replace" % (src, dst, desc)) removed = [] for t, other, direction, lid in conflicts: if not lid: return self._out("failed", {"removed": removed}, "у связи «%s» с %s нет id — снять нельзя" % (t, other)) if removed: self.gap("ensure_link") self.jira.delete(API + "/issueLink/%s" % lid) removed.append(t) if wanted: return self._out("done", {"existed": True, "removed": removed}) if removed: self.gap("ensure_link") # inwardIssue — источник (получает outward-описание «Блокирует начало…»), outwardIssue — цель self.jira.post(API + "/issueLink", {"type": {"name": lt}, "inwardIssue": {"key": src}, "outwardIssue": {"key": dst}}) return self._out("done", {"created": True, "removed": removed}) # --- ensure_sprint --------------------------------------------------------- def sprints(self) -> Dict[str, dict]: """Имя → спринт (active и future) доски board_id; без доски — пусто.""" if self._sprints is None: out, start = {}, 0 board = self.s.get("board_id") while board: page = self.jira.get(AGILE + "/board/%s/sprint" % board, state="active,future", startAt=start, maxResults=50) for sp in page.get("values", []): out[sp["name"]] = sp if page.get("isLast", True) or not page.get("values"): break start += len(page["values"]) self._sprints = out return self._sprints def do_ensure_sprint(self, p: dict, cmd: dict) -> dict: key, want = p["key"], p["sprint"] # у задачи без спринта Jira банка отвечает без ключа fields (#13) issue = self.jira.get(AGILE + "/issue/%s" % key, fields="sprint") cur = (((issue.get("fields") or {}).get("sprint")) or {}).get("name") if want == "backlog": # в бэклог доска не нужна if cur is None: return self._out("done", {"sprint": None, "changed": False}) self.jira.post(AGILE + "/backlog/issue", {"issues": [key]}) return self._out("done", {"sprint": None, "from": cur}) if not self.s.get("board_id"): return self._out("failed", error="board_id не задан в настройках хаба — спринты недоступны") if cur == want: return self._out("done", {"sprint": want, "changed": False}) sp = self.sprints().get(want) if not sp: return self._out("failed", error="нет спринта «%s» среди активных и будущих доски %s: %s" % (want, self.s["board_id"], ", ".join(sorted(self.sprints())) or "—")) self.jira.post(AGILE + "/sprint/%s/issue" % sp["id"], {"issues": [key]}) return self._out("done", {"sprint": want, "from": cur}) # --- refresh --------------------------------------------------------------- STATE_FIELDS = ["summary", "description", "status", "priority", "assignee", "reporter", "duedate", "labels", "issuetype", "created", "updated", "issuelinks"] def state_fields(self) -> str: extra = [self.fid(x) for x in ("planned_start", "planned_end") if self.has_field(x)] ep = (self.s.get("epic_link") or {}) # режим связи с эпиком — только field (mode=link не поддержан) if ep.get("mode", "field") == "field" and ep.get("field_id"): extra.append(ep["field_id"]) return ",".join(self.STATE_FIELDS + extra) def search_all(self, jql: str) -> List[dict]: issues, start = [], 0 while True: page = self.jira.get(API + "/search", jql=jql, fields=self.state_fields(), startAt=start, maxResults=100) got = page.get("issues") or [] issues += got start += len(got) if not got or start >= int(page.get("total") or 0): return issues def search_tolerant(self, keys: List[str], missing: List[str], errors: List[HttpError]) -> List[dict]: """key in (...): Jira Server отвечает 400 на несуществующий/недоступный ключ. Чанк делим пополам до одиночного ключа; одиночный с 400 — «не найден» (в missing), не ошибка. Прочие HttpError пробрасываются. Первый 400 копится в errors: вызывающий пробрасывает его, если не найден НИ ОДИН ключ (неверная конфигурация, а не удалённые задачи). Стоимость: до ~2·m·log2(n) вызовов на m отсутствующих ключей из n — принятое ограничение, рулинг R-7.""" if not keys: return [] try: return self.search_all("key in (%s)" % ",".join(keys)) except HttpError as e: if e.code != 400: raise if not errors: errors.append(e) if len(keys) == 1: missing.append(keys[0]) return [] mid = len(keys) // 2 return self.search_tolerant(keys[:mid], missing, errors) + self.search_tolerant(keys[mid:], missing, errors) def to_state(self, raw: dict) -> dict: f = raw.get("fields") or {} def name(obj): return (obj or {}).get("name") ep = self.s.get("epic_link") or {} epic = f.get(ep["field_id"]) if ep.get("mode", "field") == "field" and ep.get("field_id") else None return {"key": raw["key"], "id": raw.get("id"), "issue_type": name(f.get("issuetype")), "summary": f.get("summary"), "description": f.get("description"), "status": name(f.get("status")), "priority": name(f.get("priority")), "assignee": name(f.get("assignee")), "reporter": name(f.get("reporter")), "duedate": _date(f.get("duedate")), "planned_start": _date(f.get(self.fid("planned_start"))) if self.has_field("planned_start") else None, "planned_end": _date(f.get(self.fid("planned_end"))) if self.has_field("planned_end") else None, "epic_key": epic, "labels": list(f.get("labels") or []), "sprint": None, "created": f.get("created"), "jira_updated": f.get("updated")} def links_from(self, raw: dict) -> List[dict]: key, out = raw["key"], [] for l in (raw.get("fields") or {}).get("issuelinks") or []: t = (l.get("type") or {}).get("name") if l.get("outwardIssue"): out.append({"source_key": key, "target_key": l["outwardIssue"]["key"], "link_type": t}) if l.get("inwardIssue"): out.append({"source_key": l["inwardIssue"]["key"], "target_key": key, "link_type": t}) return out def sprint_names(self, keys: List[str]) -> Dict[str, str]: # Источник спринта — Agile по активным и будущим спринтам доски (R-6): id поля спринта в настройках хаба нет, # поэтому у задач закрытых спринтов sprint = None (принятое ограничение). names = {} if not keys or not self.s.get("board_id"): return names for name, sp in self.sprints().items(): for i in range(0, len(keys), 100): chunk = keys[i:i + 100] start = 0 while True: page = self.jira.get(AGILE + "/sprint/%s/issue" % sp["id"], jql="key in (%s)" % ",".join(chunk), fields="key", startAt=start, maxResults=50) got = page.get("issues") or [] for it in got: names[it["key"]] = name start += len(got) if not got or start >= int(page.get("total") or 0): break return names def do_refresh(self, p: dict, cmd: dict) -> dict: missing: List[str] = [] errors: List[HttpError] = [] if p.get("jql"): raws = self.search_all(p["jql"]) else: keys = list(dict.fromkeys(k for k in (p.get("keys") or []) if k)) if not keys: return self._out("done", {"issues": 0, "neighbors": 0, "pages": 0}) raws = [] for i in range(0, len(keys), 100): raws += self.search_tolerant(keys[i:i + 100], missing, errors) if not raws and errors: # ни один ключ не найден: это конфигурация/запрос, а не удалённые задачи raise errors[0] fetched = {r["key"] for r in raws} known = set(self.hub.known_keys()) if self.hub else set() linked = set() for r in raws: for l in self.links_from(r): linked.add(l["source_key"]) linked.add(l["target_key"]) absent = sorted(k for k in linked if k not in fetched and k not in known) neighbors = [] for i in range(0, len(absent), 100): neighbors += self.search_tolerant(absent[i:i + 100], missing, errors) everything = raws + neighbors names = self.sprint_names([r["key"] for r in everything]) issues, links, seen = [], [], set() for r in everything: st = self.to_state(r) st["sprint"] = names.get(r["key"]) issues.append(st) for l in self.links_from(r): sig = (l["source_key"], l["target_key"], l["link_type"]) if sig not in seen: seen.add(sig) links.append(l) neighbor_keys = [r["key"] for r in neighbors] pages = 0 for i in range(0, len(issues), 100): chunk = issues[i:i + 100] chunk_keys = {x["key"] for x in chunk} pages += 1 self.hub.state({"command_id": cmd.get("id"), "agent_version": AGENT_VERSION, "issues": chunk, "links": [l for l in links if l["source_key"] in chunk_keys or l["target_key"] in chunk_keys], "neighbor_keys": [k for k in neighbor_keys if k in chunk_keys], "page": pages, "last_page": i + 100 >= len(issues)}) result = {"issues": len(raws), "neighbors": len(neighbors), "pages": pages} if missing: result["missing"] = list(dict.fromkeys(missing))[:50] return self._out("done", result) # --------------------------------------------------------------------------- # Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте # --------------------------------------------------------------------------- class Updater: """Самообновление агента (спека agent «Самообновление агента», design 7–12). Работает только под run.py.""" def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True, config_path: Optional[str] = None): self.hub, self.dir, self.launcher_version, self.enabled = hub, base_dir, launcher_version, enabled self.config_path = config_path or os.path.join(base_dir, "agent.json") self.state_path = os.path.join(base_dir, "update.json") self.new_path = os.path.join(base_dir, "agent.py.new") self.agent_path = os.path.join(base_dir, "agent.py") def state(self) -> Optional[dict]: return read_update_state(self.state_path) def confirm(self) -> None: """Первый успешный claim новой версии: trial этой версии → ok (запускатель перестаёт ждать отката).""" st = self.state() if st and st.get("state") == "trial" and st.get("to_version") == AGENT_VERSION: write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256")) log("обновление до v%d подтверждено" % AGENT_VERSION) class Runner: RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Jira / 5xx (R-10) RETRY_MIN_WAIT = 30 # секунд до повтора def __init__(self, cfg: dict, hub: HubClient, jira: JiraClient, journal: Journal): self.cfg, self.hub, self.jira, self.journal = cfg, hub, jira, journal self.settings = None # type: Optional[dict] self.executor = None # type: Optional[Executor] self.stop = False self.sleep = time.sleep self.idle_limit = None # type: Optional[int] self.updater = None # type: Optional[Updater] self.exit_code = 0 # EXIT_UPDATE — выход для подмены версии @property def poll(self) -> float: return max(float(self.cfg.get("poll_seconds", 5) or 0), 0) def hub_wait(self, e: Exception, base: float) -> float: """Залогировать сбой хаба и вернуть, сколько ждать. 401/403 — токен неверен: повторять часто бессмысленно.""" if isinstance(e, HttpError) and e.code in (401, 403): log("хаб: HTTP %d — неверный hub_token в agent.json, повтор через 60 с" % e.code) return 60 log("хаб: %s, повтор через %d с" % (e, base)) return base def load_settings(self) -> None: self.settings = self.hub.settings() self.executor = Executor(self.jira, self.settings, self.hub) self.executor.sleep = self.sleep log("настройки хаба v%s (проект %s)" % (self.settings.get("settings_version"), self.settings.get("project_key"))) def restore_pace(self) -> None: """После перезапуска время последнего вызова Jira неизвестно (живёт в памяти). Берём самое позднее время записи журнала: finished пишется сразу после последнего вызова команды, оценка не раньше настоящего вызова (M-9). Время в будущем (часы ушли назад) — считаем «сейчас»: пауза полная (M-6 #12). Оценка служит только паузе: last_call_iso (уходит хабу в claim) ставит лишь настоящий вызов Jira (M-7 #12).""" if self.jira.last_call_at is not None: return at = self.journal.last_at() if at is None: return ago = (datetime.now(timezone.utc) - at).total_seconds() if ago < 0: log("журнал: время последней записи в будущем (%s) — пауза перед первой командой полная" % at.isoformat()) ago = 0 self.jira.last_call_at = time.monotonic() - ago log("темп: последний вызов Jira не позже %s (по журналу)" % at.astimezone(timezone.utc).replace(microsecond=0).isoformat()) def send_result(self, rec: dict) -> bool: """Отчёт хабу; повторяется каждые 10 с, пока хаб не подтвердит. 404/409/422 — хаб отверг (команда отменена или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка.""" cid = rec["command_id"] payload = {"status": rec["status"], "result": rec.get("result"), "error": rec.get("error"), "jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0), "agent_version": AGENT_VERSION} while True: try: self.hub.result(cid, payload) self.journal.acked(cid) return True except HttpError as e: if e.code in (404, 409, 422): log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e)) self.journal.acked(cid) return True wait = self.hub_wait(e, 10) except ConnectionError as e: wait = self.hub_wait(e, 10) if self.stop: log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid) return False self.sleep(wait) def wait_seconds(self, seconds) -> bool: """Спать заданное время шагами по секунде; False — остановлены во время ожидания.""" remaining = float(seconds or 0) if remaining > 0: log(" пауза %.0f с" % remaining) while remaining > 0 and not self.stop: step = min(remaining, 1.0) self.sleep(step) remaining -= step return not self.stop def wait_pause(self, pause_seconds) -> bool: """Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы. Если вызовов Jira ещё не было — пауза не нужна.""" pause = float(pause_seconds or 0) if pause > 0 and self.jira.last_call_at is not None: return self.wait_seconds(pause - (time.monotonic() - self.jira.last_call_at)) return not self.stop def run_once(self, cmd: dict) -> dict: try: return self.executor.run(cmd) except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление log(" ! необработанная ошибка: %r" % e) return {"status": "failed", "result": None, "error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]), "jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited} def execute(self, cmd: dict, journal_claimed: bool = True) -> None: cid = cmd["id"] log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200])) if journal_claimed: self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду if not self.wait_pause(cmd.get("pause_seconds")): return self.journal.started(cid) # started — только после паузы calls = limited = attempt = 0 while True: out = self.run_once(cmd) calls += out["jira_calls"] limited += out["rate_limited"] if not out.pop("retryable", False): break if attempt >= self.RETRY_ATTEMPTS: out = {"status": "failed", "result": None, "error": "результат неизвестен (обрыв связи с Jira), повтор безопасен: %s" % out.get("error")} break attempt += 1 log(" ! обрыв связи с Jira, повтор %d/%d через проверку: %s" % (attempt, self.RETRY_ATTEMPTS, out.get("error"))) # F-1: обрыв не обновляет last_call_at, поэтому ждём фиксированно от момента отказа, а не от последнего вызова if not self.wait_seconds(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT)): return # остановка: finished не пишем, команду подхватит recover out["jira_calls"], out["rate_limited"] = calls, limited self.journal.finished(cid, out["status"], out.get("result"), out.get("error"), calls, limited) log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False))) rec = dict(out) rec["command_id"] = cid self.send_result(rec) def fetch_command(self, cid): """Команда из хаба по id; None — хаб такой не знает; при недоступности хаба ждём и повторяем.""" while True: try: return self.hub.command(cid) except HttpError as e: if e.code in (404, 409, 422): log("#%s: хаб ответил %s" % (cid, e)) return None wait = self.hub_wait(e, 10) except ConnectionError as e: wait = self.hub_wait(e, 10) if self.stop: raise InterruptedError(cid) self.sleep(wait) def recover(self) -> None: for rec in self.journal.unacked_finished(): log("досылаю отчёт по #%s" % rec["command_id"]) self.send_result(rec) for cid in self.journal.unfinished_started(): if self.stop: return try: cmd = self.fetch_command(cid) except InterruptedError: return if cmd is None: self.journal.acked(cid) continue if cmd.get("status") != "claimed": log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status"))) self.journal.acked(cid) continue log("#%s выдана до перезапуска — выполняю через проверку" % cid) self.execute(cmd, journal_claimed=False) # claimed/started для неё уже в журнале (FM-4); пауза — из хаба (M-9) def reload_settings_before(self, cmd) -> bool: """Настройки сменились: перечитать ДО выполнения. False — остановка, команда не должна выполняться (если она уже выдана хабом, фиксируем claimed — подхватит recover).""" while True: try: self.load_settings() return True except (ConnectionError, HttpError) as e: wait = self.hub_wait(e, max(self.poll, 1)) if self.stop: if cmd: self.journal.claimed(cmd["id"]) return False if not cmd: self.sleep(wait) return False self.sleep(wait) def loop(self) -> None: self.restore_pace() idle, recovered = 0, False while not self.stop: if self.settings is None: try: self.load_settings() except (ConnectionError, HttpError) as e: self.sleep(self.hub_wait(e, max(self.poll, 1))) continue if not recovered: self.recover() recovered = not self.stop continue try: upd = self.updater resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso, upd.state() if upd else None, upd.launcher_version if upd else None) except (ConnectionError, HttpError) as e: self.sleep(self.hub_wait(e, max(self.poll, 1))) continue if self.updater: self.updater.confirm() cmd = resp.get("command") if resp.get("settings_version") != self.settings.get("settings_version"): if not self.reload_settings_before(cmd): continue if cmd: idle = 0 self.execute(cmd) continue idle += 1 if self.idle_limit is not None and idle >= self.idle_limit: return self.sleep(self.poll) def acquire_lock(journal_path: str): """Исключительная блокировка <журнал>.lock: один процесс агента на журнал. Хаб довыдаёт команду без отчёта при следующем claim, поэтому второй процесс выполнял бы ту же команду одновременно с первым. None — уже занято.""" f = open(journal_path + ".lock", "a") try: fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: f.close() return None return f def cmd_run(cfg: dict, journal_path: str) -> int: lock = acquire_lock(journal_path) if lock is None: log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path) return 1 try: return _run(cfg, journal_path) finally: lock.close() def cmd_prune(journal_path: str) -> int: """Удалить подтверждённые записи старше 30 дней. Под той же блокировкой, что run: prune перезаписывает файл, и запись, добавленная работающим агентом (finished — неотправленный отчёт), пропала бы (T2.3).""" lock = acquire_lock(journal_path) if lock is None: log("агент запущен с журналом %s — prune невозможен, сначала остановите агента" % journal_path) return 1 try: removed = Journal(journal_path).prune(30) log("журнал: удалено записей %d" % removed) return 0 finally: lock.close() def _run(cfg: dict, journal_path: str) -> int: runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path)) def on_signal(signum, frame): try: if runner.stop: log("повторный сигнал — выхожу немедленно") else: log("сигнал %s: доделываю текущую команду и выхожу" % signum) except RuntimeError: # print внутри обработчика, прервавшего другой print (reentrant call) pass if runner.stop: sys.exit(130) runner.stop = True signal.signal(signal.SIGINT, on_signal) signal.signal(signal.SIGTERM, on_signal) log("jira-agent v%d: хаб %s, Jira %s, журнал %s" % (AGENT_VERSION, cfg["hub_url"], cfg["jira"]["base_url"], journal_path)) runner.loop() log("остановлен") return 0 # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def cmd_check(cfg: dict) -> int: ok = True hub = HubClient(cfg) via = "" if hub.folder is not None: hub.timeout = 10 # реле молчит дольше 10 с — хаб недоступен via = " через папку %s" % hub.folder.root try: s = hub.settings() log("хаб: доступен%s, настройки v%s, проект %s" % (via, s.get("settings_version"), s.get("project_key"))) except (ConnectionError, HttpError) as e: log("хаб: НЕДОСТУПЕН — %s" % e) ok = False try: me = JiraClient(cfg).get(API + "/myself") log("Jira: доступен, пользователь %s" % me.get("name")) except (ConnectionError, HttpError) as e: log("Jira: НЕДОСТУПЕН — %s" % e) ok = False return 0 if ok else 1 def cmd_selftest(cfg: dict) -> int: """Самопроверка версии перед обновлением: её запускает предыдущая версия. Контракт кодов MUST сохраняться: 0 — хаб и Jira доступны, 3 — что-то недоступно (временно); любой другой код — файл негоден (design 10).""" log("jira-agent v%d" % AGENT_VERSION) return 0 if cmd_check(cfg) == 0 else 3 def main(argv=None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--config", default=os.path.join(HERE, "agent.json")) ap.add_argument("--journal", default=os.path.join(HERE, "journal.jsonl")) ap.add_argument("cmd", nargs="?", default="run", choices=["run", "check", "selftest", "prune"]) args = ap.parse_args(argv) cfg = load_config(args.config) if args.cmd == "check": return cmd_check(cfg) if args.cmd == "selftest": return cmd_selftest(cfg) if args.cmd == "prune": return cmd_prune(args.journal) return cmd_run(cfg, args.journal) if __name__ == "__main__": sys.exit(main())