From 3aa882392996ed2162dad54c7e03231a6644f525 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 14 Oct 2026 22:22:32 +0400 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=D1=81=D0=B0=D0=BC=D0=BE=D0=BE?= =?UTF-8?q?=D0=B1=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=B2?= =?UTF-8?q?=20=D0=BF=D0=B0=D1=83=D0=B7=D0=B5=20=E2=80=94=20=D1=81=D0=BA?= =?UTF-8?q?=D0=B0=D1=87=D0=B8=D0=B2=D0=B0=D0=BD=D0=B8=D0=B5,=20sha256,=20s?= =?UTF-8?q?elftest,=20=D0=B2=D1=8B=D1=85=D0=BE=D0=B4=2075=20(#19)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- src/agent.py | 129 ++++++++++++++++++++++++++++++++-- tests/test_runner.py | 21 ++++++ tests/test_update.py | 160 ++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 303 insertions(+), 7 deletions(-) diff --git a/src/agent.py b/src/agent.py index f93bc83..1669b30 100644 --- a/src/agent.py +++ b/src/agent.py @@ -950,6 +950,10 @@ class Executor: # --------------------------------------------------------------------------- class Updater: """Самообновление агента (спека agent «Самообновление агента», design 7–12). Работает только под run.py.""" + RETRY_SECONDS = 300 # временный сбой подготовки — повтор не раньше (R-8) + HINT_SECONDS = 3600 # подсказка «запускайте через run.py» — не чаще (R-10) + SELFTEST_TIMEOUT = 120 + VERSION_RE = re.compile(r"^AGENT_VERSION = (\d+)[ \t]*\r?$", re.M) def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True, config_path: Optional[str] = None): @@ -958,6 +962,11 @@ class Updater: 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") + self.retry_after = {} # (version, sha256) -> clock(), раньше которого не повторять + self.hint_at = None # type: Optional[float] + self.clock = time.monotonic + self.should_stop = lambda: False + self.run_selftest = self._run_selftest def state(self) -> Optional[dict]: return read_update_state(self.state_path) @@ -969,6 +978,101 @@ class Updater: write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256")) log("обновление до v%d подтверждено" % AGENT_VERSION) + def check(self, release) -> bool: + """Вызывается после claim без команды. True — agent.py.new проверен, update.json = staged: run выходит с 75.""" + if not self.enabled or not isinstance(release, dict): + return False + try: + version, sha = int(release.get("version")), str(release.get("sha256")) + except (TypeError, ValueError): + return False + if version <= AGENT_VERSION: + return False + st = self.state() + if st and st.get("state") in ("rejected", "rolled_back") and st.get("to_version") == version \ + and st.get("sha256") == sha: + return False + if self.launcher_version is None: + now = self.clock() + if self.hint_at is None or now - self.hint_at >= self.HINT_SECONDS: + self.hint_at = now + log("доступна v%d; для самообновления запускайте через run.py" % version) + return False + until = self.retry_after.get((version, sha)) + if until is not None and self.clock() < until: + return False + return self.prepare(version, sha) + + def prepare(self, version: int, sha: str) -> bool: + log("доступна v%d — готовлю обновление" % version) + try: + rel = self.hub.release() + except (ConnectionError, HttpError) as e: + return self._transient(version, sha, "скачивание: %s" % e) + content = rel.get("content") if isinstance(rel, dict) else None + if not isinstance(content, str) or rel.get("version") != version or rel.get("sha256") != sha: + return self._transient(version, sha, "публикация сменилась между запросами") + data = content.encode("utf-8") + if hashlib.sha256(data).hexdigest() != sha: + return self._reject(version, sha, "sha256 не сходится") + try: + compile(content, "agent.py.new", "exec") + except (SyntaxError, ValueError) as e: + return self._reject(version, sha, "не компилируется: %s" % e) + if self.VERSION_RE.findall(content) != [str(version)]: + return self._reject(version, sha, "AGENT_VERSION в файле не %d" % version) + tmp = self.new_path + ".tmp" + with open(tmp, "wb") as f: + f.write(data) + try: + os.chmod(tmp, os.stat(self.agent_path).st_mode & 0o777) + except OSError: + pass + os.replace(tmp, self.new_path) + if self.should_stop(): + self._drop_new() + return False + code, out = self.run_selftest(self.new_path) + if self.should_stop(): # Ctrl-C уронил и selftest — это не отказ версии + self._drop_new() + return False + if code == 0: + write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha) + log("v%d готова, перезапуск через run.py" % version) + return True + if code is None: + return self._transient(version, sha, "selftest: таймаут %d с" % self.SELFTEST_TIMEOUT) + if code == 3: + return self._transient(version, sha, "selftest: хаб или Jira недоступны") + return self._reject(version, sha, "selftest: код %s: %s" % (code, out[-500:])) + + def _run_selftest(self, path: str): + try: + p = subprocess.run([sys.executable, path, "--config", self.config_path, "selftest"], cwd=self.dir, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=self.SELFTEST_TIMEOUT) + except subprocess.TimeoutExpired: + return None, "" + return p.returncode, p.stdout.decode("utf-8", "replace") + + def _drop_new(self) -> None: + for path in (self.new_path, self.new_path + ".tmp"): + try: + os.remove(path) + except FileNotFoundError: + pass + + def _transient(self, version, sha, reason) -> bool: + self._drop_new() + self.retry_after[(version, sha)] = self.clock() + self.RETRY_SECONDS + log("обновление до v%d отложено на %d мин: %s" % (version, self.RETRY_SECONDS // 60, reason)) + return False + + def _reject(self, version, sha, reason) -> bool: + self._drop_new() + write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason) + log("v%d отвергнута: %s" % (version, reason)) + return False + class Runner: RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Jira / 5xx (R-10) @@ -1190,6 +1294,9 @@ class Runner: idle = 0 self.execute(cmd) continue + if self.updater and not self.stop and self.updater.check(resp.get("agent_release")): + self.exit_code = EXIT_UPDATE + return idle += 1 if self.idle_limit is not None and idle >= self.idle_limit: return @@ -1208,13 +1315,13 @@ def acquire_lock(journal_path: str): return f -def cmd_run(cfg: dict, journal_path: str) -> int: +def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> int: lock = acquire_lock(journal_path) if lock is None: log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path) return 1 try: - return _run(cfg, journal_path) + return _run(cfg, journal_path, config_path) finally: lock.close() @@ -1234,8 +1341,18 @@ def cmd_prune(journal_path: str) -> int: lock.close() -def _run(cfg: dict, journal_path: str) -> int: +def launcher_version() -> Optional[int]: + """Версия запускателя из окружения (run.py ставит JIRA_AGENT_LAUNCHER); без запускателя — None.""" + try: + return int(os.environ["JIRA_AGENT_LAUNCHER"]) + except (KeyError, ValueError): + return None + + +def _run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> int: runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path)) + runner.updater = Updater(runner.hub, HERE, launcher_version(), cfg.get("auto_update", True) is not False, config_path) + runner.updater.should_stop = lambda: runner.stop def on_signal(signum, frame): try: @@ -1253,8 +1370,8 @@ def _run(cfg: dict, journal_path: str) -> int: 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 + log("выход для обновления (код %d)" % EXIT_UPDATE if runner.exit_code == EXIT_UPDATE else "остановлен") + return runner.exit_code # --------------------------------------------------------------------------- @@ -1302,7 +1419,7 @@ def main(argv=None) -> int: return cmd_selftest(cfg) if args.cmd == "prune": return cmd_prune(args.journal) - return cmd_run(cfg, args.journal) + return cmd_run(cfg, args.journal, args.config) if __name__ == "__main__": diff --git a/tests/test_runner.py b/tests/test_runner.py index 65c9ceb..c573423 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -66,6 +66,27 @@ class RunnerTest(unittest.TestCase): self.assertIsNone(self.hub.claim_bodies[0]["update_state"]) self.assertIsNone(self.hub.claim_bodies[0]["launcher_version"]) + def test_idle_with_release_exits_for_update(self): + rel = {"version": 6, "sha256": "s", "content": "x"} + self.hub.release = rel + self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1) + self.runner.updater.check = lambda r: r == {"version": 6, "sha256": "s"} + self.runner.idle_limit = 5 + self.runner.loop() + self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE) + self.assertEqual(self.hub.claims, 1) + + def test_command_first_then_update(self): + self.hub.release = {"version": 6, "sha256": "s", "content": "x"} + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1) + seen = [] + self.runner.updater.check = lambda r: (seen.append(r), True)[1] + self.runner.loop() + self.assertEqual(self.hub.results[1]["status"], "done") # команда выполнена и подтверждена + self.assertEqual(len(seen), 1) # проверка — только после claim без команды + self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE) + def test_restart_pace_from_old_finished(self): at = (datetime.now(timezone.utc) - timedelta(seconds=10)).replace(microsecond=0).isoformat() self.journal.append({"kind": "finished", "command_id": 99, "status": "done", "result": None, "error": None, diff --git a/tests/test_update.py b/tests/test_update.py index e392878..88c95d8 100644 --- a/tests/test_update.py +++ b/tests/test_update.py @@ -1,9 +1,11 @@ +import hashlib import json import os import tempfile import unittest -from _common import agent +from _common import SETTINGS, agent +from fake_hub import FakeHub class UpdateStateTest(unittest.TestCase): @@ -49,5 +51,161 @@ class UpdateStateTest(unittest.TestCase): self.assertEqual(u.state()["state"], "rolled_back") +def make_release(version, body="print('ok')\n"): + content = "AGENT_VERSION = %d\n%s" % (version, body) + return {"version": version, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), "content": content} + + +class UpdaterCheckTest(unittest.TestCase): + def setUp(self): + self.dir = tempfile.TemporaryDirectory() + self.hub = FakeHub(SETTINGS) + cfg = {"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5} + self.u = agent.Updater(agent.HubClient(cfg), self.dir.name, 1) + self.selftests = [] + self.code = 0 + self.u.run_selftest = lambda path: (self.selftests.append(path), (self.code, "вывод selftest"))[1] + self.now = [1000.0] + self.u.clock = lambda: self.now[0] + + def tearDown(self): + self.hub.stop() + self.dir.cleanup() + + def ref(self, rel): + return {"version": rel["version"], "sha256": rel["sha256"]} + + def test_good_version_is_staged(self): + rel = make_release(6) + self.hub.release = rel + self.assertTrue(self.u.check(self.ref(rel))) + with open(self.u.new_path, encoding="utf-8") as f: + self.assertEqual(f.read(), rel["content"]) + st = self.u.state() + self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("staged", 5, 6, rel["sha256"])) + self.assertEqual(self.selftests, [self.u.new_path]) + + def test_not_newer_or_disabled_or_no_release(self): + self.hub.release = make_release(5) + self.assertFalse(self.u.check(self.ref(self.hub.release))) + self.assertFalse(self.u.check(None)) + self.hub.release = make_release(6) + self.u.enabled = False + self.assertFalse(self.u.check(self.ref(self.hub.release))) + self.assertEqual(self.hub.release_requests, 0) + + def test_sha_mismatch_rejected_once(self): + rel = make_release(6) + self.hub.release = dict(rel, content=rel["content"] + "# подмена\n") + ref = self.ref(rel) + self.hub.release["sha256"] = rel["sha256"] + self.assertFalse(self.u.check(ref)) + st = self.u.state() + self.assertEqual((st["state"], st["to_version"], st["error"]), ("rejected", 6, "sha256 не сходится")) + self.assertFalse(os.path.exists(self.u.new_path)) + self.assertFalse(self.u.check(ref)) + self.assertEqual(self.hub.release_requests, 1) # отвергнутую версию больше не качает + + def test_syntax_error_and_wrong_version_rejected(self): + bad = make_release(6, body="def (:\n") + self.hub.release = bad + self.assertFalse(self.u.check(self.ref(bad))) + self.assertIn("не компилируется", self.u.state()["error"]) + wrong = make_release(7) + wrong = dict(wrong, version=8) + self.hub.release = wrong + self.assertFalse(self.u.check(self.ref(wrong))) + self.assertIn("AGENT_VERSION", self.u.state()["error"]) + self.assertEqual(self.selftests, []) + + def test_selftest_bad_code_rejected(self): + self.hub.release = make_release(6) + self.code = 1 + self.assertFalse(self.u.check(self.ref(self.hub.release))) + st = self.u.state() + self.assertEqual(st["state"], "rejected") + self.assertTrue(st["error"].startswith("selftest: код 1")) + self.assertIn("вывод selftest", st["error"]) + self.assertFalse(os.path.exists(self.u.new_path)) + + def test_selftest_unavailable_is_transient(self): + self.hub.release = make_release(6) + self.code = 3 + ref = self.ref(self.hub.release) + self.assertFalse(self.u.check(ref)) + self.assertIsNone(self.u.state()) # update.json не тронут + self.assertFalse(os.path.exists(self.u.new_path)) + self.now[0] += 299 + self.assertFalse(self.u.check(ref)) + self.assertEqual(len(self.selftests), 1) # раньше 5 минут не повторяет + self.now[0] += 2 + self.code = 0 + self.assertTrue(self.u.check(ref)) + + def test_selftest_timeout_is_transient(self): + self.hub.release = make_release(6) + self.code = None + self.assertFalse(self.u.check(self.ref(self.hub.release))) + self.assertIsNone(self.u.state()) + + def test_download_failure_is_transient(self): + rel = make_release(6) # в claim есть, а /agent/release — 404 + self.assertFalse(self.u.check(self.ref(rel))) + self.assertIsNone(self.u.state()) + + def test_release_changed_between_requests_is_transient(self): + old, new = make_release(6), make_release(7) + self.hub.release = new + self.assertFalse(self.u.check(self.ref(old))) + self.assertIsNone(self.u.state()) + self.assertEqual(self.selftests, []) + + def test_rolled_back_same_sha_not_retried(self): + rel = make_release(6) + self.hub.release = rel + agent.write_update_state(self.u.state_path, "rolled_back", 5, 6, rel["sha256"], "код 1 через 3 с") + self.assertFalse(self.u.check(self.ref(rel))) + fixed = make_release(6, body="print('fixed')\n") + self.hub.release = fixed + self.assertTrue(self.u.check(self.ref(fixed))) # тот же номер, другой sha — пробуем + + def test_without_launcher_only_hint(self): + self.u.launcher_version = None + self.hub.release = make_release(6) + logs = [] + orig = agent.log + agent.log = logs.append + try: + self.assertFalse(self.u.check(self.ref(self.hub.release))) + self.assertFalse(self.u.check(self.ref(self.hub.release))) + self.now[0] += 3600 + self.assertFalse(self.u.check(self.ref(self.hub.release))) + finally: + agent.log = orig + self.assertEqual(logs.count("доступна v6; для самообновления запускайте через run.py"), 2) + self.assertEqual(self.hub.release_requests, 0) + + def test_stop_during_selftest_is_not_reject(self): + self.hub.release = make_release(6) + stop = [False] + self.u.should_stop = lambda: stop[0] + + def selftest(path): + stop[0] = True # Ctrl-C пришёл и selftest упал с 1 + return 1, "KeyboardInterrupt" + self.u.run_selftest = selftest + self.assertFalse(self.u.check(self.ref(self.hub.release))) + self.assertIsNone(self.u.state()) + self.assertFalse(os.path.exists(self.u.new_path)) + + def test_real_selftest_runs_new_file(self): + u = agent.Updater(None, self.dir.name, 1, config_path=os.path.join(self.dir.name, "agent.json")) + path = os.path.join(self.dir.name, "agent.py.new") + with open(path, "w", encoding="utf-8") as f: + f.write("import sys\nprint('args', sys.argv[1:])\nsys.exit(3)\n") + code, out = u._run_selftest(path) + self.assertEqual(code, 3) + self.assertIn("'--config', '%s', 'selftest'" % os.path.join(self.dir.name, "agent.json"), out) + if __name__ == "__main__": unittest.main()