feat(agent): цикл claim → пауза → журнал → отчёт, восстановление после перезапуска, run
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
This commit is contained in:
Andrey Epifancev
parent
8ebf5250b7
commit
ea1ab864da
172
src/agent.py
172
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__":
|
||||
|
||||
238
tests/test_runner.py
Normal file
238
tests/test_runner.py
Normal file
@ -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()
|
||||
Loading…
x
Reference in New Issue
Block a user