From b90c639eaa7ee4438b51efe1ea877ef587d3af6e Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Thu, 15 Oct 2026 00:14:58 +0400 Subject: [PATCH] =?UTF-8?q?fix(agent):=20=D0=BA=D0=BE=D0=B4=2073=20=D0=BF?= =?UTF-8?q?=D1=80=D0=B8=20=D0=B7=D0=B0=D0=BD=D1=8F=D1=82=D0=BE=D0=BC=20?= =?UTF-8?q?=D0=B6=D1=83=D1=80=D0=BD=D0=B0=D0=BB=D0=B5,=20=D0=BE=D0=B4?= =?UTF-8?q?=D0=BD=D0=B0=20=D1=81=D1=82=D1=80=D0=BE=D0=BA=D0=B0=20=D1=82?= =?UTF-8?q?=D0=B5=D0=BC=D0=BF=D0=B0,=20=D0=B8=D1=81=D1=82=D0=BE=D1=87?= =?UTF-8?q?=D0=BD=D0=B8=D0=BA=20=D0=BE=D1=88=D0=B8=D0=B1=D0=BA=D0=B8=20?= =?UTF-8?q?=D1=85=D0=B0=D0=B1=D0=B0=20=D0=B2=20refresh=20(M-7=20#22;=20M-2?= =?UTF-8?q?,=20M-3,=20M-5,=20M-6=20#18)?= 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 | 38 ++++++++++++++++++++++++++++------- tests/test_refresh.py | 23 ++++++++++++++++++++++ tests/test_runner.py | 46 ++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 99 insertions(+), 8 deletions(-) diff --git a/src/agent.py b/src/agent.py index 1f06544..a021a0c 100644 --- a/src/agent.py +++ b/src/agent.py @@ -34,6 +34,7 @@ from datetime import datetime, timezone from typing import Any, Dict, List, Optional AGENT_VERSION = 9 +EXIT_LOCKED = 73 # «журнал занят другим процессом»: для run.py это не сбой версии (design 8) EXIT_UPDATE = 75 # «обновление готово»: run.py подменит agent.py на agent.py.new HERE = os.path.dirname(os.path.abspath(__file__)) API = "/rest/api/2" @@ -283,7 +284,7 @@ class FolderTransport: os.remove(path) except OSError: pass - status, text = int(resp.get("status") or 0), resp.get("body") or "" + text = body or "" # status и body проверены выше (M-5 #18) if not 200 <= status < 300: raise HttpError(status, text) try: @@ -502,6 +503,15 @@ class ConfigGap(Exception): """В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле.""" +class HubError(Exception): + """Ошибка хаба внутри команды (refresh: known_keys, state): источник — хаб, не Jira (M-6 #18, design 12). + Повтор — только при недоступности хаба и 5xx; refresh в Jira не пишет, повтор безопасен.""" + + def __init__(self, cause: Exception): + super().__init__("хаб: %s" % cause) + self.retryable = isinstance(cause, ConnectionError) or (isinstance(cause, HttpError) and cause.code >= 500) + + class Executor: def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None): self.jira, self.s, self.hub = jira, settings, hub @@ -519,6 +529,11 @@ class Executor: return handler(cmd.get("params") or {}, cmd) except ConfigGap as e: return self._out("failed", error=str(e)) + except HubError as e: + out = self._out("failed", error=str(e)) + out["retryable"] = e.retryable + out["peer"] = "хабом" + return out 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) @@ -532,6 +547,13 @@ class Executor: return {"status": status, "result": result, "error": error, "jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited} + def _hub(self, fn, *args): + """Вызов хаба из команды: его сбой — HubError, чтобы не выдать его за обрыв связи с Jira.""" + try: + return fn(*args) + except (ConnectionError, HttpError) as e: + raise HubError(e) from None + def gap(self, cmd_type: str) -> None: """Зазор между записями внутри одной команды — по политике хаба для этого типа.""" seconds = int((self.s.get("pacing") or {}).get(cmd_type, 0) or 0) @@ -907,7 +929,7 @@ class Executor: 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() + known = set(self._hub(self.hub.known_keys)) if self.hub else set() linked = set() for r in raws: for l in self.links_from(r): @@ -935,7 +957,7 @@ class Executor: 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, + self._hub(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)}) @@ -1136,7 +1158,8 @@ class Runner: ago = (datetime.now(timezone.utc) - at).total_seconds() if ago < 0: log("журнал: время последней записи в будущем (%s) — пауза перед первой командой полная" % at.isoformat()) - ago = 0 + self.jira.last_call_at = time.monotonic() + return # вторая строка «не позже <будущее>» противоречила бы первой (M-2 #18) self.jira.last_call_at = time.monotonic() - ago log("темп: последний вызов Jira не позже %s (по журналу)" % at.astimezone(timezone.utc).replace(microsecond=0).isoformat()) @@ -1207,14 +1230,15 @@ class Runner: out = self.run_once(cmd) calls += out["jira_calls"] limited += out["rate_limited"] + peer = out.pop("peer", "Jira") # с кем оборвалась связь: Jira или хаб (refresh) if not out.pop("retryable", False): break if attempt >= self.RETRY_ATTEMPTS: out = {"status": "failed", "result": None, - "error": "результат неизвестен (обрыв связи с Jira), повтор безопасен: %s" % out.get("error")} + "error": "результат неизвестен (обрыв связи с %s), повтор безопасен: %s" % (peer, out.get("error"))} break attempt += 1 - log(" ! обрыв связи с Jira, повтор %d/%d через проверку: %s" % (attempt, self.RETRY_ATTEMPTS, out.get("error"))) + log(" ! обрыв связи с %s, повтор %d/%d через проверку: %s" % (peer, 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 @@ -1337,7 +1361,7 @@ def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> lock = acquire_lock(journal_path) if lock is None: log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path) - return 1 + return EXIT_LOCKED try: return _run(cfg, journal_path, config_path) finally: diff --git a/tests/test_refresh.py b/tests/test_refresh.py index b6c16b6..d5235fc 100644 --- a/tests/test_refresh.py +++ b/tests/test_refresh.py @@ -184,6 +184,29 @@ class RefreshTest(unittest.TestCase): self.assertEqual(self.hub.commands[cid]["status"], "done") self.assertEqual(self.hub.results[cid]["status"], "done") + def test_hub_unreachable_in_refresh_names_hub(self): + self.fake.add("AURORA-1") + self.ex.hub = agent.HubClient({"hub_url": "http://127.0.0.1:1", "hub_token": "t", "timeout": 2}) + out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) + self.assertEqual(out["status"], "failed") + self.assertTrue(out["error"].startswith("хаб: "), out["error"]) + self.assertTrue(out["retryable"]) + self.assertEqual(out["peer"], "хабом") + + def test_hub_4xx_in_refresh_not_retryable(self): + self.fake.add("AURORA-1") + self.ex.hub = agent.HubClient({"hub_url": self.hub.base_url, "hub_token": "wrong-token", "timeout": 5}) + out = self.ex.run({"id": 1, "type": "refresh", "params": {"keys": ["AURORA-1"]}}) + self.assertEqual(out["status"], "failed") + self.assertTrue(out["error"].startswith("хаб: HTTP 401"), out["error"]) + self.assertFalse(out["retryable"]) + + def test_jira_error_has_no_peer(self): + self.fake.fail_next.append((500, {})) + out = self.ex.run({"id": 1, "type": "refresh", "params": {"jql": "project = AURORA"}}) + self.assertTrue(out["retryable"]) + self.assertNotIn("peer", out) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_runner.py b/tests/test_runner.py index 9839fc8..92fd89e 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -400,6 +400,7 @@ class RunnerTest(unittest.TestCase): r = self.hub.results[1] self.assertEqual(r["status"], "failed") self.assertIn("результат неизвестен", r["error"]) + self.assertIn("(обрыв связи с Jira)", r["error"]) self.assertEqual(r["jira_calls"], 4) def test_stop_during_retry_wait_leaves_command_for_recovery(self): @@ -507,12 +508,55 @@ class RunnerTest(unittest.TestCase): self.assertIsNone(agent.acquire_lock(self.journal.path)) cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token) with mock.patch.object(agent.Runner, "loop", lambda r: self.fail("цикл не должен стартовать")): - self.assertEqual(agent.cmd_run(cfg, self.journal.path), 1) + self.assertEqual(agent.cmd_run(cfg, self.journal.path), agent.EXIT_LOCKED) + self.assertEqual(agent.EXIT_LOCKED, 73) finally: held.close() again = agent.acquire_lock(self.journal.path) self.assertIsNotNone(again) # после выхода первого — снова можно again.close() + def test_future_journal_time_single_log_line(self): + at = (datetime.now(timezone.utc) + timedelta(hours=2)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + logs = [] + orig = agent.log + agent.log = logs.append + try: + self.runner.restore_pace() + finally: + agent.log = orig + self.assertEqual(len(logs), 1, logs) # M-2 #18: без «не позже <будущее>» + self.assertIn("в будущем", logs[0]) + self.assertIsNotNone(self.runner.jira.last_call_at) + + def test_future_journal_time_then_stop_in_pause(self): + at = (datetime.now(timezone.utc) + timedelta(hours=2)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=30) + self.runner.sleep = lambda s: setattr(self.runner, "stop", True) # Ctrl-C в первую секунду паузы + self.runner.loop() + self.assertEqual([r["kind"] for r in self.journal.records() if r["command_id"] == 1], ["claimed"]) + self.assertEqual(self.fake.requests, []) # в Jira ни одного вызова (M-3 #18) + self.assertEqual(self.hub.commands[1]["status"], "claimed") + + def test_retry_exhausted_on_hub_names_hub(self): + self.runner.load_settings() + logs = [] + orig = agent.log + agent.log = logs.append + self.runner.executor.run = lambda cmd: {"status": "failed", "result": None, "error": "хаб: не ответил", + "jira_calls": 0, "rate_limited": 0, "retryable": True, "peer": "хабом"} + self.hub.enqueue("refresh", {"keys": ["AURORA-1"]}) + try: + self.runner.loop() + finally: + agent.log = orig + r = self.hub.results[1] + self.assertEqual(r["error"], "результат неизвестен (обрыв связи с хабом), повтор безопасен: хаб: не ответил") + self.assertTrue(any("обрыв связи с хабом, повтор 1/3" in l for l in logs), logs) + self.assertNotIn("peer", r) + + if __name__ == "__main__": unittest.main()