diff --git a/src/run.py b/src/run.py new file mode 100644 index 0000000..70e8ceb --- /dev/null +++ b/src/run.py @@ -0,0 +1,223 @@ +#!/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 = 1 +EXIT_UPDATE = 75 # агент: «agent.py.new проверен, update.json = staged» +TRIAL_SECONDS = 600 # новая версия должна записать ok за это время +KILL_GRACE = 60 # после SIGTERM до SIGKILL +BACKOFF_START, BACKOFF_MAX = 5, 300 +HEALTHY_UPTIME = 600 # проработал столько — пауза перезапуска с начала + + +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) -> None: + 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" + with open(tmp, "w", encoding="utf-8") as f: + json.dump(data, f, ensure_ascii=False) + os.replace(tmp, path) + + +def sha256_file(path: str) -> str: + with open(path, "rb") as f: + return hashlib.sha256(f.read()).hexdigest() + + +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.child = None + + # --- сигналы --------------------------------------------------------------- + def on_sigint(self, signum, frame): + """Ctrl-C в терминале приходит и агенту (общая группа): пока агент жив — решает он. Без агента (пауза) — выход.""" + 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 recover_files(self) -> None: + 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") + if os.path.exists(self.new): + st = read_state(self.state_path) + if st and st.get("state") == "staged": + log("найдена подготовленная версия — ставлю") + self.apply_update() + else: + os.remove(self.new) + + 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")): + if os.path.exists(self.new): + os.remove(self.new) + write_state(self.state_path, "rejected", st, "файл изменился после проверки") + log("обновление отвергнуто: файл изменился после проверки") + return + tmp = self.prev + ".tmp" + 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 существует на каждом шаге + write_state(self.state_path, "trial", st) + log("v%s → v%s: пробный запуск" % (st.get("from_version"), st.get("to_version"))) + + 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.""" + start = time.monotonic() + while True: + try: + return child.wait(1), False + except subprocess.TimeoutExpired: + pass + if trial: + st = read_state(self.state_path) + if st and st.get("state") == "ok": + trial = False + elif time.monotonic() - start >= self.trial_seconds: + self.terminate(child) + return child.returncode, True + + def pause(self, seconds: float) -> bool: + """Пауза перед перезапуском; False — остановлены во время паузы.""" + end = time.monotonic() + seconds + while not self.stopping and time.monotonic() < end: + time.sleep(min(1.0, end - time.monotonic())) + 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.recover_files() + while True: + st = read_state(self.state_path) + trial = bool(st and st.get("state") == "trial") + started = time.monotonic() + self.child = self.spawn() + code, timed_out = self.wait_child(self.child, trial) + self.child = None + uptime = time.monotonic() - started + st = read_state(self.state_path) + if timed_out: + if self.rollback(st, "не вышла на связь за 10 мин"): + continue + elif code in (0, 130) or self.stopping: + return code + elif code == EXIT_UPDATE: + self.apply_update() + self.backoff = BACKOFF_START + continue + elif trial and not (st and st.get("state") == "ok"): + if self.rollback(st, "код %s через %d с" % (code, uptime)): + continue + if self.stopping: + return code + wait = self.next_backoff(uptime) + log("агент упал (код %s), перезапуск через %d с" % (code, 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()) diff --git a/tests/test_launcher.py b/tests/test_launcher.py new file mode 100644 index 0000000..bb4b995 --- /dev/null +++ b/tests/test_launcher.py @@ -0,0 +1,212 @@ +import hashlib +import json +import os +import signal +import subprocess +import sys +import tempfile +import textwrap +import unittest + +from _common import SRC + +if SRC not in sys.path: + sys.path.insert(0, SRC) +import run # noqa: E402 + +# Поддельные агенты: маленькие скрипты, которые ведут себя как agent.py по контракту (коды выхода, update.json). +STATE_HELPERS = textwrap.dedent(''' + import json, os, sys + HERE = os.path.dirname(os.path.abspath(__file__)) + STATE = os.path.join(HERE, "update.json") + def state(): + try: + return json.load(open(STATE, encoding="utf-8")) + except Exception: + return None + def write(**kw): + json.dump(kw, open(STATE, "w", encoding="utf-8")) + def mark(name): + open(os.path.join(HERE, "runs.log"), "a").write(name + "\\n") +''') + + +def fake(name, body): + return STATE_HELPERS + "mark(%r)\n" % name + textwrap.dedent(body) + + +# v5 «нашёл» новую версию: кладёт agent.py.new из next.py, пишет staged и выходит с 75; после отката/отказа — выход 0. +V5_UPDATES = fake("v5", ''' + import hashlib + st = state() + if st is None: + data = open(os.path.join(HERE, "next.py"), "rb").read() + open(os.path.join(HERE, "agent.py.new"), "wb").write(data) + write(state="staged", from_version=5, to_version=6, sha256=hashlib.sha256(data).hexdigest()) + sys.exit(75) + sys.exit(0) +''') +# v5 с подменённым после проверки файлом: sha в staged не сходится с agent.py.new. +V5_BAD_STAGE = fake("v5", ''' + st = state() + if st is None: + open(os.path.join(HERE, "agent.py.new"), "w").write("мусор") + write(state="staged", from_version=5, to_version=6, sha256="чужой") + sys.exit(75) + sys.exit(0) +''') +V6_OK_THEN_STOP = fake("v6", ''' + st = state() + write(**dict(st, state="ok")) + sys.exit(0) +''') +V6_CRASH = fake("v6", 'sys.exit(1)\n') +V6_HANG = fake("v6", 'import time\ntime.sleep(60)\n') + + +class LauncherTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.dir = self.tmp.name + self.l = run.Launcher(self.dir, ["--journal", "j.jsonl"]) + self.pauses = [] + self.l.pause = lambda s: (self.pauses.append(s), True)[1] + + def tearDown(self): + self.tmp.cleanup() + + def put(self, name, text): + with open(os.path.join(self.dir, name), "w", encoding="utf-8") as f: + f.write(text) + + def read(self, name): + with open(os.path.join(self.dir, name), encoding="utf-8") as f: + return f.read() + + def state(self): + return json.loads(self.read("update.json")) + + def runs(self): + return self.read("runs.log").split() + + def next_version(self, text): + self.put("next.py", text) + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + def stage(self, new_text, to_version=6): + self.put("agent.py.new", new_text) + sha = hashlib.sha256(new_text.encode("utf-8")).hexdigest() + self.put("update.json", json.dumps({"state": "staged", "from_version": 5, "to_version": to_version, "sha256": sha})) + return sha + + def test_update_applied_and_confirmed(self): + self.put("agent.py", V5_UPDATES) + sha = self.next_version(V6_OK_THEN_STOP) + self.assertEqual(self.l.run(), 0) + self.assertEqual(self.read("agent.py"), V6_OK_THEN_STOP) + self.assertEqual(self.read("agent.py.prev"), V5_UPDATES) + self.assertFalse(os.path.exists(os.path.join(self.dir, "agent.py.new"))) + st = self.state() + self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("ok", 5, 6, sha)) + self.assertEqual(self.runs(), ["v5", "v6"]) + + def test_crash_in_trial_rolls_back(self): + self.put("agent.py", V5_UPDATES) + sha = self.next_version(V6_CRASH) + self.assertEqual(self.l.run(), 0) + self.assertEqual(self.read("agent.py"), V5_UPDATES) + st = self.state() + self.assertEqual((st["state"], st["to_version"], st["sha256"]), ("rolled_back", 6, sha)) + self.assertRegex(st["error"], r"^код 1 через \d+ с$") + self.assertEqual(self.runs(), ["v5", "v6", "v5"]) + self.assertEqual(self.pauses, []) # откат без паузы + + def test_hang_in_trial_rolls_back(self): + self.l.trial_seconds = 1 + self.l.kill_grace = 5 + self.put("agent.py", V5_UPDATES) + self.next_version(V6_HANG) + self.assertEqual(self.l.run(), 0) + self.assertEqual(self.read("agent.py"), V5_UPDATES) + self.assertEqual(self.state()["error"], "не вышла на связь за 10 мин") + + def test_crash_after_ok_restarts_with_backoff(self): + self.put("update.json", json.dumps({"state": "ok", "from_version": 5, "to_version": 6, "sha256": "s"})) + self.put("agent.py", fake("v6", ''' + n = len(open(os.path.join(HERE, "runs.log")).read().split()) + sys.exit(1 if n <= 2 else 0) + ''')) + self.assertEqual(self.l.run(), 0) + self.assertEqual(self.pauses, [5, 10]) + self.assertEqual(self.state()["state"], "ok") # отката нет + + def test_backoff_caps_and_resets(self): + self.assertEqual([self.l.next_backoff(10) for _ in range(8)], [5, 10, 20, 40, 80, 160, 300, 300]) + self.assertEqual(self.l.next_backoff(600), 5) # проработал 10 минут — с начала + + def test_75_without_staged_rejected(self): + self.put("agent.py", V5_BAD_STAGE) + self.assertEqual(self.l.run(), 0) + self.assertEqual(self.read("agent.py"), V5_BAD_STAGE) + self.assertEqual(self.runs(), ["v5", "v5"]) + self.assertFalse(os.path.exists(os.path.join(self.dir, "agent.py.new"))) + st = self.state() + self.assertEqual((st["state"], st["error"]), ("rejected", "файл изменился после проверки")) + + def test_staged_at_start_is_applied(self): + self.put("agent.py", fake("v5", "sys.exit(0)\n")) + self.stage(V6_OK_THEN_STOP) + self.assertEqual(self.l.run(), 0) + self.assertEqual(self.runs(), ["v6"]) + self.assertEqual(self.state()["state"], "ok") + + def test_stray_new_removed_at_start(self): + self.put("agent.py", fake("v5", "sys.exit(0)\n")) + self.put("agent.py.new", "остаток") + self.assertEqual(self.l.run(), 0) + self.assertFalse(os.path.exists(os.path.join(self.dir, "agent.py.new"))) + + def test_missing_agent_restored_from_prev(self): + self.put("agent.py.prev", fake("v5", "sys.exit(0)\n")) + self.assertEqual(self.l.run(), 0) + self.assertEqual(self.runs(), ["v5"]) + + def test_garbage_update_json_is_no_state(self): + self.put("update.json", '{"state": "tri') + self.put("agent.py", fake("v5", "sys.exit(0)\n")) + self.assertEqual(self.l.run(), 0) + + def test_env_and_args_passed(self): + self.put("agent.py", fake("v5", ''' + open(os.path.join(HERE, "env.txt"), "w").write(os.environ.get("JIRA_AGENT_LAUNCHER", "") + " " + " ".join(sys.argv[1:])) + sys.exit(130) + ''')) + self.assertEqual(self.l.run(), 130) + self.assertEqual(self.read("env.txt"), "1 run --journal j.jsonl") + + def test_sigint_without_child_stops(self): + self.l.on_sigint(signal.SIGINT, None) + self.assertTrue(self.l.stopping) + + def test_sigint_with_child_ignored(self): + self.l.child = object() + self.l.on_sigint(signal.SIGINT, None) + self.assertFalse(self.l.stopping) + + def test_second_launcher_refused(self): + with open(os.path.join(SRC, "run.py"), encoding="utf-8") as f: + self.put("run.py", f.read()) + self.put("agent.py", fake("v5", "sys.exit(0)\n")) + lock = run.acquire_lock(self.dir) + try: + p = subprocess.run([sys.executable, os.path.join(self.dir, "run.py")], stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, timeout=30) + finally: + lock.close() + self.assertEqual(p.returncode, 1) + self.assertIn("запускатель уже работает", p.stdout.decode("utf-8")) + self.assertFalse(os.path.exists(os.path.join(self.dir, "runs.log"))) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_py39.py b/tests/test_py39.py index c63eee3..03c76c9 100644 --- a/tests/test_py39.py +++ b/tests/test_py39.py @@ -10,21 +10,24 @@ from _common import SRC class Py39Test(unittest.TestCase): - def test_agent_parses_as_python_39(self): - with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f: - ast.parse(f.read(), feature_version=(3, 9)) + FILES = ("agent.py", "run.py") + + def test_parses_as_python_39(self): + for name in self.FILES: + with self.subTest(name), open(os.path.join(SRC, name), encoding="utf-8") as f: + ast.parse(f.read(), feature_version=(3, 9)) def test_no_third_party_imports(self): - with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f: - tree = ast.parse(f.read()) allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "http", "sys", "time", "urllib", "datetime", "typing", "uuid", "fcntl", "hashlib", "re", "subprocess"} - for node in ast.walk(tree): - if isinstance(node, ast.Import): - for a in node.names: - self.assertIn(a.name.split(".")[0], allowed, a.name) - elif isinstance(node, ast.ImportFrom): - self.assertIn((node.module or "").split(".")[0], allowed, node.module) - + for name in self.FILES: + with self.subTest(name), open(os.path.join(SRC, name), encoding="utf-8") as f: + tree = ast.parse(f.read()) + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for a in node.names: + self.assertIn(a.name.split(".")[0], allowed, "%s: %s" % (name, a.name)) + elif isinstance(node, ast.ImportFrom): + self.assertIn((node.module or "").split(".")[0], allowed, "%s: %s" % (name, node.module)) if __name__ == "__main__": unittest.main()