From ea1ab864dae4ea404c582d80f166089569bcc92b Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Mon, 12 Oct 2026 00:06:10 +0400 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=D1=86=D0=B8=D0=BA=D0=BB=20claim?= =?UTF-8?q?=20=E2=86=92=20=D0=BF=D0=B0=D1=83=D0=B7=D0=B0=20=E2=86=92=20?= =?UTF-8?q?=D0=B6=D1=83=D1=80=D0=BD=D0=B0=D0=BB=20=E2=86=92=20=D0=BE=D1=82?= =?UTF-8?q?=D1=87=D1=91=D1=82,=20=D0=B2=D0=BE=D1=81=D1=81=D1=82=D0=B0?= =?UTF-8?q?=D0=BD=D0=BE=D0=B2=D0=BB=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=BF=D0=BE?= =?UTF-8?q?=D1=81=D0=BB=D0=B5=20=D0=BF=D0=B5=D1=80=D0=B5=D0=B7=D0=B0=D0=BF?= =?UTF-8?q?=D1=83=D1=81=D0=BA=D0=B0,=20run?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o --- src/agent.py | 172 ++++++++++++++++++++++++++++++- tests/test_runner.py | 238 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 408 insertions(+), 2 deletions(-) create mode 100644 tests/test_runner.py diff --git a/src/agent.py b/src/agent.py index d9a61e2..bd2fd91 100644 --- a/src/agent.py +++ b/src/agent.py @@ -750,6 +750,175 @@ class Executor: return self._out("done", result) +# --------------------------------------------------------------------------- +# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте +# --------------------------------------------------------------------------- +class Runner: + 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] + + @property + def poll(self) -> float: + return max(float(self.cfg.get("poll_seconds", 5) or 0), 0) + + 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 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)} + 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 + log("хаб: %s, повтор через 10 с" % e) + except ConnectionError as e: + log("хаб недоступен: %s, повтор через 10 с" % e) + if self.stop: + log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid) + return False + self.sleep(10) + + def wait_pause(self, pause_seconds) -> bool: + """Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы.""" + pause = float(pause_seconds or 0) + if pause > 0 and self.jira.last_call_at is not None: + remaining = pause - (time.monotonic() - self.jira.last_call_at) + 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 execute(self, cmd: dict) -> None: + log("#%s %s %s" % (cmd["id"], cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200])) + if not self.wait_pause(cmd.get("pause_seconds")): + return + self.journal.started(cmd["id"]) # started — только после паузы + try: + out = self.executor.run(cmd) + except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление + log(" ! необработанная ошибка: %r" % e) + out = {"status": "failed", "result": None, + "error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]), + "jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited} + self.journal.finished(cmd["id"], out["status"], out.get("result"), out.get("error"), out["jira_calls"], out["rate_limited"]) + log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False))) + rec = dict(out) + rec["command_id"] = cmd["id"] + 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 + log("хаб: %s, повтор через 10 с" % e) + except ConnectionError as e: + log("хаб недоступен: %s, повтор через 10 с" % e) + if self.stop: + raise InterruptedError(cid) + self.sleep(10) + + 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) + cmd["pause_seconds"] = 0 + self.execute(cmd) + + def loop(self) -> None: + idle, recovered = 0, False + while not self.stop: + if self.settings is None: + try: + self.load_settings() + except (ConnectionError, HttpError) as e: + log("хаб: %s" % e) + self.sleep(self.poll) + continue + if not recovered: + self.recover() + recovered = not self.stop + continue + try: + resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso) + except (ConnectionError, HttpError) as e: + log("хаб: %s" % e) + self.sleep(self.poll) + continue + if resp.get("settings_version") != self.settings.get("settings_version"): + try: + self.load_settings() # настройки — до выполнения команды + except (ConnectionError, HttpError) as e: + log("хаб: не перечитать настройки: %s" % e) + cmd = resp.get("command") + 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 cmd_run(cfg: dict, journal_path: str) -> int: + runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path)) + + def on_signal(signum, frame): + if runner.stop: + log("повторный сигнал — выхожу немедленно") + sys.exit(130) + runner.stop = True + log("сигнал %s: доделываю текущую команду и выхожу" % signum) + + 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 # --------------------------------------------------------------------------- @@ -783,8 +952,7 @@ def main(argv=None) -> int: removed = Journal(args.journal).prune(30) log("журнал: удалено записей %d" % removed) return 0 - log("команда %s ещё не реализована" % args.cmd) - return 2 + return cmd_run(cfg, args.journal) if __name__ == "__main__": diff --git a/tests/test_runner.py b/tests/test_runner.py new file mode 100644 index 0000000..f62d905 --- /dev/null +++ b/tests/test_runner.py @@ -0,0 +1,238 @@ +import os +import tempfile +import time +import unittest + +from _common import SETTINGS, agent, jira_cfg, no_sleep +from fake_hub import FakeHub +from fake_jira import FakeJira + + +class RunnerTest(unittest.TestCase): + def setUp(self): + self.fake = FakeJira() + self.hub = FakeHub(SETTINGS) + self.dir = tempfile.TemporaryDirectory() + self.journal = agent.Journal(os.path.join(self.dir.name, "journal.jsonl")) + cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token) + self.runner = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), self.journal) + self.sleeps = [] + self.runner.sleep = self.sleeps.append + self.runner.jira.sleep = no_sleep + self.runner.idle_limit = 1 + self.fake.add("AURORA-1", duedate="2026-10-20") + + def tearDown(self): + self.fake.stop() + self.hub.stop() + self.dir.cleanup() + + def kinds(self): + return [(r["kind"], r["command_id"]) for r in self.journal.records()] + + def claim_in_flight(self, ctype, params, pause=0): + cid = self.hub.enqueue(ctype, params, pause=pause) + self.hub.queue.remove(cid) + self.hub.commands[cid]["status"] = "claimed" + self.hub.in_flight = cid + return cid + + def test_processes_queue(self): + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.hub.enqueue("refresh", {"keys": ["AURORA-1"]}) + self.runner.loop() + self.assertEqual(self.hub.results[1]["status"], "done") + self.assertEqual(self.hub.results[1]["result"], {"changed": ["duedate"]}) + self.assertEqual(self.hub.results[1]["jira_calls"], 2) + self.assertEqual(self.hub.results[2]["status"], "done") + self.assertEqual(len(self.hub.states), 1) + self.assertEqual(self.hub.states[0]["issues"][0]["duedate"], "2026-10-23") + self.assertEqual(self.kinds(), [("started", 1), ("finished", 1), ("acked", 1), ("started", 2), ("finished", 2), ("acked", 2)]) + self.assertEqual(self.sleeps, []) # пауз не было: last_call_at пуст, потом pause 0 + + def test_pause_waits_from_last_jira_call(self): + self.runner.jira.last_call_at = time.monotonic() + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=3) + self.runner.loop() + self.assertEqual(len(self.sleeps), 3) + self.assertAlmostEqual(sum(self.sleeps), 3, delta=0.1) + self.assertEqual(self.hub.results[1]["status"], "done") + + def test_result_retried_when_hub_down(self): + self.hub.fail_results = 2 + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.runner.loop() + self.assertEqual(self.hub.results[1]["status"], "done") + self.assertEqual(self.sleeps, [10, 10]) + self.assertEqual(self.journal.unacked_finished(), []) + + def test_result_retried_on_connection_error(self): + calls = [] + real = self.runner.hub.result + + def flaky(cid, payload): + calls.append(1) + if len(calls) < 3: + raise ConnectionError("нет связи") + return real(cid, payload) + + self.runner.hub.result = flaky + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.runner.loop() + self.assertEqual(self.sleeps, [10, 10]) + self.assertEqual(self.hub.results[1]["status"], "done") + + def test_result_rejected_is_acked(self): + for cid, code in ((1, 404), (2, 409), (3, 422)): + self.hub.reject_results = {cid: code} + self.journal.finished(cid, "done", {}, None, 0, 0) + self.assertTrue(self.runner.send_result(self.journal.unacked_finished()[0])) + self.assertEqual(self.journal.unacked_finished(), []) + + def test_result_rejected_in_loop(self): + self.hub.reject_results = {1: 404} + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.runner.loop() + self.assertNotIn(1, self.hub.results) + self.assertEqual(self.journal.unacked_finished(), []) + self.assertEqual(self.kinds()[-1], ("acked", 1)) + + def test_recover_unfinished_started(self): + cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"}, pause=99) + self.journal.started(cid) + self.runner.loop() + self.assertEqual(self.hub.results[cid]["result"], {"changed": []}) # уже так в Jira — записи нет + self.assertEqual(self.journal.unfinished_started(), []) + self.assertEqual(self.sleeps, []) # пауза при восстановлении не применяется + self.assertEqual([r[0] for r in self.fake.writes()], []) + + def test_recover_ensure_issue_finds_duplicate(self): + self.fake.add("AURORA-7", reporter="me", summary="Новая задача") + before = len(self.fake.issues) + cid = self.claim_in_flight("ensure_issue", {"temp_key": "new:1", "issue_type": "task", "summary": "Новая задача"}) + self.journal.started(cid) + self.runner.loop() + self.assertEqual(self.hub.results[cid]["status"], "done") + self.assertEqual(self.hub.results[cid]["result"], {"key": "AURORA-7", "existed": True}) + self.assertEqual(len(self.fake.issues), before) + self.assertEqual(self.fake.writes(), []) + + def test_recover_skips_command_not_claimed_anymore(self): + cid = self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.hub.queue.remove(cid) + self.hub.commands[cid]["status"] = "cancelled" + self.journal.started(cid) + self.runner.loop() + self.assertNotIn(cid, self.hub.results) + self.assertEqual(self.kinds(), [("started", cid), ("acked", cid)]) + + def test_recover_unacked_finished(self): + cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.journal.started(cid) + self.journal.finished(cid, "done", {"changed": ["duedate"]}, None, 2, 0) + self.hub.enqueue("refresh", {"keys": ["AURORA-1"]}) + self.runner.loop() + self.assertEqual(self.hub.results[cid]["result"], {"changed": ["duedate"]}) + self.assertEqual(self.hub.results[2]["status"], "done") + self.assertEqual(self.kinds()[:3], [("started", 1), ("finished", 1), ("acked", 1)]) + + def test_stop_during_pause(self): + self.runner.jira.last_call_at = time.monotonic() + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=5) + self.runner.sleep = lambda s: setattr(self.runner, "stop", True) + self.runner.loop() + self.assertEqual(self.journal.records(), []) + self.assertEqual(self.hub.results, {}) + self.assertEqual(self.hub.commands[1]["status"], "claimed") + + def test_stop_during_execution_finishes_and_reports(self): + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.hub.enqueue("refresh", {"keys": ["AURORA-1"]}) + self.runner.load_settings() + run = self.runner.executor.run + + def run_and_signal(cmd): + self.runner.stop = True # сигнал пришёл во время выполнения + return run(cmd) + + self.runner.executor.run = run_and_signal + self.runner.loop() + self.assertEqual(self.hub.results[1]["status"], "done") + self.assertNotIn(2, self.hub.results) # вторая команда не начата + self.assertEqual(self.kinds(), [("started", 1), ("finished", 1), ("acked", 1)]) + + def test_stop_during_execution_hub_down_single_attempt(self): + self.hub.fail_results = 5 + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + self.runner.load_settings() + run = self.runner.executor.run + self.runner.executor.run = lambda cmd: (setattr(self.runner, "stop", True), run(cmd))[1] + self.runner.loop() + self.assertEqual(self.sleeps, []) + self.assertEqual([r["command_id"] for r in self.journal.unacked_finished()], [1]) + + def test_unexpected_exception_becomes_failed(self): + self.runner.load_settings() + + def boom(cmd): + raise KeyError("summary") + + self.runner.executor.run = boom + self.hub.enqueue("ensure_issue", {"temp_key": "new:1", "issue_type": "task"}) + self.runner.loop() + self.assertEqual(self.hub.results[1]["status"], "failed") + self.assertIn("KeyError", self.hub.results[1]["error"]) + self.assertEqual(self.journal.unfinished_started(), []) + self.assertEqual(self.journal.unacked_finished(), []) + + def test_recover_waits_when_hub_down(self): + cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"}) + self.journal.started(cid) + real = self.runner.hub.command + calls = [] + + def flaky(i): + calls.append(1) + if len(calls) == 1: + raise ConnectionError("нет связи") + return real(i) + + self.runner.hub.command = flaky + self.runner.loop() + self.assertEqual(self.sleeps, [10]) + self.assertEqual(self.hub.results[cid]["status"], "done") + + def test_settings_reload_on_version_change(self): + self.runner.load_settings() + self.runner.settings["settings_version"] = 0 + self.runner.loop() + self.assertEqual(self.runner.settings["settings_version"], 1) + + def test_settings_reloaded_before_command(self): + self.runner.load_settings() + self.hub.settings["settings_version"] = 2 + self.hub.settings["pacing"] = dict(self.hub.settings["pacing"], ensure_fields=0) + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) + seen = [] + real = self.runner.execute + self.runner.execute = lambda cmd: (seen.append(self.runner.settings["settings_version"]), real(cmd)) + self.runner.loop() + self.assertEqual(seen, [2]) + + def test_hub_down_at_start_keeps_polling(self): + calls = [] + + def hub_settings(): + calls.append(1) + if len(calls) < 3: + raise ConnectionError("нет связи") + return dict(SETTINGS) + + self.runner.hub.settings = hub_settings + self.runner.loop() + self.assertEqual(len(calls), 3) + self.assertEqual(self.sleeps, [0, 0]) + + +if __name__ == "__main__": + unittest.main()