#!/usr/bin/env python3 """run.py — запускатель jira-agent: запускает agent.py дочерним процессом, ставит новую версию после самообновления агента, откатывает её, если она упала или не вышла на связь, и поднимает агента после падения. Только стандартная библиотека, Python 3.9. agent.py не импортирует: тот меняется, запускатель — нет. Спека: jira-hub/openspec/specs/agent/spec.md, «Запускатель run.py». python3 run.py [--config agent.json] [--journal journal.jsonl] # аргументы уходят в agent.py run Обновляется вручную (редко): остановить, заменить файл, запустить. """ import fcntl import hashlib import json import os import signal import subprocess import sys import time from datetime import datetime, timezone from typing import List, Optional LAUNCHER_VERSION = 2 EXIT_UPDATE = 75 # агент: «agent.py.new проверен, update.json = staged» EXIT_LOCKED = 73 # агент: «журнал занят другим процессом» — не сбой версии (design 6) TRIAL_SECONDS = 600 # новая версия должна записать ok за это время KILL_GRACE = 60 # после SIGTERM до SIGKILL BACKOFF_START, BACKOFF_MAX = 5, 300 HEALTHY_UPTIME = 600 # проработал столько — пауза перезапуска с начала STOP_CODES = (0, 130, -signal.SIGINT) # выход агента после Ctrl-C (-SIGINT — Ctrl-C до установки его обработчиков) def log(msg: str) -> None: print("[%s] run: %s" % (datetime.now().strftime("%H:%M:%S"), msg), flush=True) def read_state(path: str) -> Optional[dict]: try: with open(path, encoding="utf-8") as f: data = json.load(f) except (OSError, ValueError): return None return data if isinstance(data, dict) else None def write_state(path: str, state: str, base: Optional[dict], error=None) -> bool: """Атомарная запись update.json. False — не записано (диск полон): запускатель не падает (design 5).""" base = base or {} data = {"state": state, "from_version": base.get("from_version"), "to_version": base.get("to_version"), "sha256": base.get("sha256"), "error": error, "at": datetime.now(timezone.utc).replace(microsecond=0).isoformat()} tmp = path + ".tmp" try: with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) os.replace(tmp, path) except OSError as e: log("не удалось записать update.json: %s" % e) return False return True def sha256_file(path: str) -> Optional[str]: try: with open(path, "rb") as f: return hashlib.sha256(f.read()).hexdigest() except OSError: return None def remove(path: str) -> None: try: os.remove(path) except OSError: pass def acquire_lock(base_dir: str): """Один запускатель на каталог. None — уже занято.""" f = open(os.path.join(base_dir, "run.py.lock"), "a") try: fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except OSError: f.close() return None return f class Launcher: def __init__(self, base_dir: str, args: List[str]): self.dir, self.args = base_dir, list(args) self.agent = os.path.join(base_dir, "agent.py") self.new = os.path.join(base_dir, "agent.py.new") self.prev = os.path.join(base_dir, "agent.py.prev") self.state_path = os.path.join(base_dir, "update.json") self.trial_seconds, self.kill_grace = TRIAL_SECONDS, KILL_GRACE self.backoff = BACKOFF_START self.stopping = False self.interrupted = False # был Ctrl-C: выход агента с 0/130 — остановка человеком, а не сбой версии self.child = None # --- сигналы --------------------------------------------------------------- def on_sigint(self, signum, frame): """Ctrl-C в терминале приходит и агенту (общая группа): пока агент жив — решает он, запускатель только запоминает, что человек остановил (I-1). Без агента (пауза) — выход.""" self.interrupted = True if self.child is None: self.stopping = True def on_sigterm(self, signum, frame): self.stopping = True child = self.child if child is not None and hasattr(child, "send_signal"): child.send_signal(signal.SIGTERM) # --- файлы версии ---------------------------------------------------------- def reconcile(self) -> None: """Сверка update.json с файлами при старте (design 1): запускатель мог упасть посреди подмены или отката.""" for tmp in (self.new + ".tmp", self.prev + ".tmp", self.state_path + ".tmp"): remove(tmp) if not os.path.exists(self.agent) and os.path.exists(self.prev): os.replace(self.prev, self.agent) log("agent.py не найден — восстановлен из agent.py.prev") st = read_state(self.state_path) state, sha = (st or {}).get("state"), (st or {}).get("sha256") if state == "staged": if sha and sha256_file(self.agent) == sha: # подмена сделана, trial не записан (M-5) remove(self.new) write_state(self.state_path, "trial", st) log("v%s уже на месте — пробный запуск" % st.get("to_version")) elif sha and os.path.exists(self.new) and sha256_file(self.new) == sha: log("найдена подготовленная версия — ставлю") self.apply_update() else: remove(self.new) write_state(self.state_path, "rejected", st, "подготовленный файл потерян") log("подготовленная v%s потеряна — отвергнута" % st.get("to_version")) return remove(self.new) if state == "trial" and (not sha or sha256_file(self.agent) != sha): self.drop_stale_trial(st) def in_trial(self) -> bool: """Пробный период для этого запуска: trial и agent.py — именно эта версия (design 2).""" st = read_state(self.state_path) if not (st and st.get("state") == "trial"): return False if not st.get("sha256") or sha256_file(self.agent) != st.get("sha256"): self.drop_stale_trial(st) return False return True def drop_stale_trial(self, st: dict) -> None: """agent.py не та версия, что в trial (прерван откат, ручная подмена): пробного периода нет, версию не ставить (M-3).""" reason = "пробный период прерван: agent.py не v%s" % st.get("to_version") write_state(self.state_path, "rolled_back", st, reason) log(reason) def apply_update(self) -> None: st = read_state(self.state_path) if not (st and st.get("state") == "staged" and os.path.exists(self.new) and sha256_file(self.new) == st.get("sha256")): remove(self.new) write_state(self.state_path, "rejected", st, "файл изменился после проверки") log("обновление отвергнуто: файл изменился после проверки") return tmp = self.prev + ".tmp" try: with open(self.agent, "rb") as src, open(tmp, "wb") as dst: dst.write(src.read()) os.replace(tmp, self.prev) os.replace(self.new, self.agent) # agent.py существует на каждом шаге except OSError as e: self.fail_update(st, e) return if not write_state(self.state_path, "trial", st): try: os.replace(self.prev, self.agent) # без trial новая версия осталась бы без пробного периода except OSError as e: log("не удалось вернуть agent.py.prev: %s" % e) self.fail_update(st, "update.json не записан") return log("v%s → v%s: пробный запуск" % (st.get("from_version"), st.get("to_version"))) def fail_update(self, st: dict, err) -> None: """Подмена не удалась (диск полон, права): текущая версия остаётся, эта — rejected навсегда (design 4).""" remove(self.prev + ".tmp") remove(self.new) write_state(self.state_path, "rejected", st, "подмена не удалась: %s" % err) log("подмена v%s не удалась: %s" % (st.get("to_version"), err)) def rollback(self, st: Optional[dict], reason: str) -> bool: if not os.path.exists(self.prev): log("откат невозможен: нет agent.py.prev") return False os.replace(self.prev, self.agent) write_state(self.state_path, "rolled_back", st, reason) log("откат v%s → v%s: %s" % ((st or {}).get("to_version"), (st or {}).get("from_version"), reason)) return True # --- процесс агента -------------------------------------------------------- def spawn(self): env = dict(os.environ, JIRA_AGENT_LAUNCHER=str(LAUNCHER_VERSION)) return subprocess.Popen([sys.executable, "-u", self.agent, "run"] + self.args, env=env, cwd=self.dir) def terminate(self, child) -> None: child.send_signal(signal.SIGTERM) try: child.wait(self.kill_grace) except subprocess.TimeoutExpired: child.kill() child.wait() def wait_child(self, child, trial: bool): """(код выхода, «пробный период истёк», «версия подтверждена»). В пробный период ждём ok не дольше trial_seconds. Подтверждение — любое состояние update.json, кроме trial этой же версии, пока агент жив: запускатель сам update.json при живом агенте не пишет, значит это сделал подтвердивший агент (ok, затем rejected/staged для следующей версии — I-1).""" start = time.monotonic() st0 = read_state(self.state_path) if trial else None trial_sha = (st0 or {}).get("sha256") confirmed = False while True: try: return child.wait(1), False, confirmed except subprocess.TimeoutExpired: pass if trial and not confirmed: st = read_state(self.state_path) if st and not (st.get("state") == "trial" and st.get("sha256") == trial_sha): confirmed = True elif time.monotonic() - start >= self.trial_seconds: self.terminate(child) return child.returncode, True, False def pause(self, seconds: float) -> bool: """Пауза перед перезапуском; False — остановлены во время паузы.""" end = time.monotonic() + seconds while not self.stopping: left = end - time.monotonic() # одно чтение часов: остаток не уйдёт в минус между проверкой и sleep (M-1) if left <= 0: break time.sleep(min(1.0, left)) return not self.stopping def next_backoff(self, uptime: float) -> int: if uptime >= HEALTHY_UPTIME: self.backoff = BACKOFF_START wait = self.backoff self.backoff = min(self.backoff * 2, BACKOFF_MAX) return wait def run(self) -> int: self.reconcile() code = 0 while True: if self.stopping: # сигнал пришёл между запусками — нового агента не запускаем (M-2) return code trial = self.in_trial() trial_sha = (read_state(self.state_path) or {}).get("sha256") if trial else None started = time.monotonic() self.interrupted = False # Ctrl-C относится к этому запуску агента self.child = self.spawn() if self.stopping: # SIGTERM между Popen и присваиванием child до агента не дошёл self.child.send_signal(signal.SIGTERM) code, timed_out, confirmed = self.wait_child(self.child, trial) self.child = None uptime = time.monotonic() - started st = read_state(self.state_path) ok = bool(st and st.get("state") == "ok") if trial and not (ok or confirmed) and st and not (st.get("state") == "trial" and st.get("sha256") == trial_sha) \ and sha256_file(self.agent) == trial_sha: confirmed = True # ok и следующая запись агента случились между опросами (I-1) ok = ok or confirmed reason = "агент упал (код %s)" % code if timed_out and ok: # ответ claim пришёл в последнюю секунду (M-4, design 3) log("подтверждение пришло в последний момент — отката нет") self.backoff = BACKOFF_START continue if timed_out: if self.rollback(st, "не вышла на связь за 10 мин"): continue elif self.stopping or (self.interrupted and code in STOP_CODES): return code elif code == EXIT_UPDATE: self.apply_update() self.backoff = BACKOFF_START if self.interrupted: # Ctrl-C во время подготовки: подмена сделана, trial — при следующем старте return 0 continue elif code == EXIT_LOCKED: # журнал держит другой процесс — версия ни при чём (M-7, design 6) reason = "журнал занят другим процессом (агент уже запущен)" elif trial and not ok: if self.rollback(st, "код %s через %d с" % (code, uptime)): continue if self.stopping: return code wait = self.next_backoff(uptime) log("%s, перезапуск через %d с" % (reason, wait)) if not self.pause(wait): return code def main(argv=None) -> int: base = os.path.dirname(os.path.abspath(__file__)) lock = acquire_lock(base) if lock is None: log("запускатель уже работает в %s — второй не стартует" % base) return 1 try: launcher = Launcher(base, sys.argv[1:] if argv is None else argv) signal.signal(signal.SIGINT, launcher.on_sigint) signal.signal(signal.SIGTERM, launcher.on_sigterm) log("запускатель v%d: агент %s" % (LAUNCHER_VERSION, launcher.agent)) return launcher.run() finally: lock.close() if __name__ == "__main__": sys.exit(main())