aep/jira-agent
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(agent): запись claimed до паузы (R-9), повтор при обрыве связи (R-10), agent_version, устойчивость цикла к сбоям хаба

Browse Source 
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 2026-10-12 00:31:52 +04:00
parent 4018f994dc
commit 6bcc6f611f
4 changed files with 323 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

231
src/agent.py
View File

@ -11,6 +11,7 @@ Python 3.9. Спека: jira-hub/openspec/changes/bridge-core/specs/agent/spec.m
""" """
import argparse import argparse
import base64 import base64
import http.client
import json import json
import os import os
import signal import signal
@ -70,7 +71,7 @@ class Http:
def __repr__(self): def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, self.base) return "<%s %s>" % (self.__class__.__name__, self.base)
def call(self, method: str, path: str, body=None, params=None) -> Any: def call(self, method: str, path: str, body=None, params=None, timeout=None) -> Any:
url = self.base + path url = self.base + path
if params: if params:
url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None}) url += "?" + urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
@ -82,7 +83,7 @@ class Http:
if data is not None: if data is not None:
req.add_header("Content-Type", "application/json; charset=utf-8") req.add_header("Content-Type", "application/json; charset=utf-8")
try: try:
with urllib.request.urlopen(req, timeout=self.timeout, context=self.ctx) as resp: with urllib.request.urlopen(req, timeout=timeout or self.timeout, context=self.ctx) as resp:
raw = resp.read() raw = resp.read()
status = resp.status status = resp.status
except urllib.error.HTTPError as e: except urllib.error.HTTPError as e:
@ -90,12 +91,12 @@ class Http:
try: try:
headers = {k.lower(): v for k, v in e.headers.items()} headers = {k.lower(): v for k, v in e.headers.items()}
text = e.read().decode("utf-8", "replace") text = e.read().decode("utf-8", "replace")
except (OSError, ConnectionError, TimeoutError) as err: except (OSError, ConnectionError, TimeoutError, http.client.HTTPException) as err:
raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None raise ConnectionError("%s: обрыв при чтении ответа: %s" % (self.base, err)) from None
finally: finally:
e.close() e.close()
raise HttpError(e.code, text, headers) from None raise HttpError(e.code, text, headers) from None
except (urllib.error.URLError, ConnectionError, TimeoutError, OSError) as e: except (urllib.error.URLError, ConnectionError, TimeoutError, OSError, http.client.HTTPException) as e:
raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None raise ConnectionError("%s недоступен: %s" % (self.base, e)) from None
self._touch() self._touch()
try: try:
@ -181,7 +182,7 @@ class HubClient(Http):
def claim(self, agent_version, settings_version, last_jira_call_at) -> dict: def claim(self, agent_version, settings_version, last_jira_call_at) -> dict:
return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version, return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version,
"last_jira_call_at": last_jira_call_at}) "last_jira_call_at": last_jira_call_at}, timeout=max(self.timeout, 45))
def settings(self) -> dict: def settings(self) -> dict:
return self.call("GET", "/agent/settings") return self.call("GET", "/agent/settings")
@ -255,6 +256,10 @@ class Journal:
log("журнал: строка %d не читается, пропускаю" % n) log("журнал: строка %d не читается, пропускаю" % n)
return good return good
def claimed(self, command_id) -> None:
"""Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту."""
self.append({"kind": "claimed", "command_id": int(command_id)})
def started(self, command_id) -> None: def started(self, command_id) -> None:
self.append({"kind": "started", "command_id": int(command_id)}) self.append({"kind": "started", "command_id": int(command_id)})
@ -276,11 +281,11 @@ class Journal:
return [r for cid, r in last.items() if cid not in acked] return [r for cid, r in last.items() if cid not in acked]
def unfinished_started(self) -> List[int]: def unfinished_started(self) -> List[int]:
"""Порядковый учёт: started открывает id, finished/acked закрывает; порядок — первое открытие.""" """Порядковый учёт: claimed/started открывают id, finished/acked закрывает; порядок — первое открытие."""
opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия
for r in self.records(): for r in self.records():
kind, cid = r.get("kind"), r["command_id"] kind, cid = r.get("kind"), r["command_id"]
if kind == "started": if kind in ("claimed", "started"):
opened[cid] = True opened[cid] = True
elif kind in ("finished", "acked") and cid in opened: elif kind in ("finished", "acked") and cid in opened:
opened[cid] = False opened[cid] = False
@ -299,7 +304,7 @@ class Journal:
cid = r["command_id"] cid = r["command_id"]
if r.get("kind") == "acked": if r.get("kind") == "acked":
last_ack[cid] = i last_ack[cid] = i
elif r.get("kind") in ("started", "finished"): elif r.get("kind") in ("claimed", "started", "finished"):
last_open[cid] = i last_open[cid] = i
drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется
for cid, i in last_ack.items(): for cid, i in last_ack.items():
@ -344,6 +349,10 @@ def _date(value) -> Optional[str]:
return (value or "")[:10] or None return (value or "")[:10] or None
class ConfigGap(Exception):
"""В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле."""
class Executor: class Executor:
def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None): def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None):
self.jira, self.s, self.hub = jira, settings, hub self.jira, self.s, self.hub = jira, settings, hub
@ -359,10 +368,16 @@ class Executor:
return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type")) return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type"))
try: try:
return handler(cmd.get("params") or {}, cmd) return handler(cmd.get("params") or {}, cmd)
except HttpError as e: except ConfigGap as e:
return self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500]))
except ConnectionError as e:
return self._out("failed", error=str(e)) return self._out("failed", error=str(e))
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)
return out
except ConnectionError as e:
out = self._out("failed", error=str(e))
out["retryable"] = True
return out
def _out(self, status, result=None, error=None) -> dict: def _out(self, status, result=None, error=None) -> dict:
return {"status": status, "result": result, "error": error, return {"status": status, "result": result, "error": error,
@ -389,19 +404,36 @@ class Executor:
it = (self.s.get("issue_types") or {}).get(kind) or {} it = (self.s.get("issue_types") or {}).get(kind) or {}
return {"id": str(it["id"])} if it.get("id") else {"name": it.get("name")} return {"id": str(it["id"])} if it.get("id") else {"name": it.get("name")}
def epic_fields(self, epic_key) -> dict: def epic_field_id(self) -> str:
ep = self.s.get("epic_link") or {} ep = self.s.get("epic_link") or {}
if epic_key and ep.get("mode", "field") == "field" and ep.get("field_id"): if ep.get("mode", "field") != "field":
return {ep["field_id"]: epic_key} raise ConfigGap("epic_key передан, но epic_link.mode = %r (поддержан только «field»)" % ep.get("mode"))
return {} if not ep.get("field_id"):
raise ConfigGap("epic_key передан, но не настроено epic_link.field_id")
return ep["field_id"]
def epic_fields(self, epic_key) -> dict:
return {self.epic_field_id(): epic_key} if epic_key else {}
def need_field(self, logical: str) -> str:
if not self.has_field(logical):
raise ConfigGap("поле передано, но не настроено field_ids.%s" % logical)
return self.fid(logical)
# --- ensure_issue ---------------------------------------------------------- # --- ensure_issue ----------------------------------------------------------
def do_ensure_issue(self, p: dict, cmd: dict) -> dict: def do_ensure_issue(self, p: dict, cmd: dict) -> dict:
jql = "project = %s AND reporter = currentUser() AND created >= -7d" % self.s["project_key"] summary = (p["summary"] or "").strip()
found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200) jql = "project = %s AND reporter = currentUser() AND created >= -7d ORDER BY created DESC" % self.s["project_key"]
for it in found.get("issues", []): start = 0
if (it.get("fields", {}).get("summary") or "") == p["summary"]: while start < 1000:
return self._out("done", {"key": it["key"], "existed": True}) found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200, startAt=start)
issues = found.get("issues", [])
for it in issues:
if (it.get("fields", {}).get("summary") or "").strip() == summary:
return self._out("done", {"key": it["key"], "existed": True})
start += len(issues)
if not issues or start >= int(found.get("total") or 0):
break
fields = {"project": {"key": self.s["project_key"]}, "issuetype": self.issue_type_ref(p["issue_type"]), fields = {"project": {"key": self.s["project_key"]}, "issuetype": self.issue_type_ref(p["issue_type"]),
"summary": p["summary"]} "summary": p["summary"]}
if p.get("description") is not None: if p.get("description") is not None:
@ -411,8 +443,8 @@ class Executor:
if p.get("duedate"): if p.get("duedate"):
fields[self.fid("duedate")] = p["duedate"] fields[self.fid("duedate")] = p["duedate"]
for logical in ("planned_start", "planned_end"): for logical in ("planned_start", "planned_end"):
if p.get(logical) and self.has_field(logical): if p.get(logical):
fields[self.fid(logical)] = p[logical] fields[self.need_field(logical)] = p[logical]
if p.get("labels") is not None: if p.get("labels") is not None:
fields[self.fid("labels")] = list(p["labels"]) fields[self.fid("labels")] = list(p["labels"])
if p.get("assignee"): if p.get("assignee"):
@ -437,17 +469,22 @@ class Executor:
if "priority" in p: if "priority" in p:
want = self.priority_value(p["priority"]) want = self.priority_value(p["priority"])
vkey = list(want)[0] vkey = list(want)[0]
if ((cur.get(self.fid("priority")) or {}).get(vkey)) != want[vkey]: have = (cur.get(self.fid("priority")) or {}).get(vkey)
if (None if have is None else str(have)) != str(want[vkey]):
fields[self.fid("priority")] = want fields[self.fid("priority")] = want
changed.append("priority") changed.append("priority")
for logical in ("duedate", "planned_start", "planned_end"): for logical in ("duedate", "planned_start", "planned_end"):
if logical in p and self.has_field(logical) and _date(cur.get(self.fid(logical))) != _date(p[logical]): if logical not in p:
continue
if logical != "duedate" and p[logical] and not self.has_field(logical):
self.need_field(logical)
if self.has_field(logical) and _date(cur.get(self.fid(logical))) != _date(p[logical]):
fields[self.fid(logical)] = p[logical] fields[self.fid(logical)] = p[logical]
changed.append(logical) changed.append(logical)
if "epic_key" in p: if "epic_key" in p and (p["epic_key"] or (self.s.get("epic_link") or {}).get("field_id")):
ep = self.s.get("epic_link") or {} fid = self.epic_field_id()
if ep.get("field_id") and (cur.get(ep["field_id"]) or None) != (p["epic_key"] or None): if (cur.get(fid) or None) != (p["epic_key"] or None):
fields[ep["field_id"]] = p["epic_key"] fields[fid] = p["epic_key"]
changed.append("epic_key") changed.append("epic_key")
if "assignee" in p and ((cur.get("assignee") or {}).get("name") or None) != (p["assignee"] or None): if "assignee" in p and ((cur.get("assignee") or {}).get("name") or None) != (p["assignee"] or None):
fields["assignee"] = {"name": p["assignee"]} if p["assignee"] else None fields["assignee"] = {"name": p["assignee"]} if p["assignee"] else None
@ -467,12 +504,19 @@ class Executor:
# --- ensure_comment -------------------------------------------------------- # --- ensure_comment --------------------------------------------------------
def do_ensure_comment(self, p: dict, cmd: dict) -> dict: def do_ensure_comment(self, p: dict, cmd: dict) -> dict:
key, marker, body = p["key"], p["marker"], p["body"] key, marker, body = p["key"], p["marker"], p["body"]
if marker not in body[:80]: heads = (marker, "*%s.*" % marker) # маркер — только в начале тела (формат агента: «*маркер.* текст»)
if not body.startswith(heads):
body = "*%s.* %s" % (marker, body) body = "*%s.* %s" % (marker, body)
existing = None existing, start = None, 0
for c in self.jira.get(API + "/issue/%s/comment" % key).get("comments", []): while existing is None:
if marker in (c.get("body") or "")[:80]: page = self.jira.get(API + "/issue/%s/comment" % key, startAt=start, maxResults=100)
existing = c comments = page.get("comments", [])
for c in comments:
if (c.get("body") or "").startswith(heads):
existing = c
break
start += len(comments)
if not comments or page.get("total") is None or start >= int(page["total"]):
break break
if existing and (existing.get("body") or "") == body: if existing and (existing.get("body") or "") == body:
return self._out("done", {"comment_id": existing["id"], "changed": False}) return self._out("done", {"comment_id": existing["id"], "changed": False})
@ -740,7 +784,7 @@ class Executor:
chunk = issues[i:i + 100] chunk = issues[i:i + 100]
chunk_keys = {x["key"] for x in chunk} chunk_keys = {x["key"] for x in chunk}
pages += 1 pages += 1
self.hub.state({"command_id": cmd.get("id"), "issues": chunk, 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], "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], "neighbor_keys": [k for k in neighbor_keys if k in chunk_keys],
"page": pages, "last_page": i + 100 >= len(issues)}) "page": pages, "last_page": i + 100 >= len(issues)})
@ -754,6 +798,9 @@ class Executor:
# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте # Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
class Runner: class Runner:
RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Jira / 5xx (R-10)
RETRY_MIN_WAIT = 30 # секунд до повтора
def __init__(self, cfg: dict, hub: HubClient, jira: JiraClient, journal: Journal): def __init__(self, cfg: dict, hub: HubClient, jira: JiraClient, journal: Journal):
self.cfg, self.hub, self.jira, self.journal = cfg, hub, jira, journal self.cfg, self.hub, self.jira, self.journal = cfg, hub, jira, journal
self.settings = None # type: Optional[dict] self.settings = None # type: Optional[dict]
@ -766,6 +813,14 @@ class Runner:
def poll(self) -> float: def poll(self) -> float:
return max(float(self.cfg.get("poll_seconds", 5) or 0), 0) return max(float(self.cfg.get("poll_seconds", 5) or 0), 0)
def hub_wait(self, e: Exception, base: float) -> float:
"""Залогировать сбой хаба и вернуть, сколько ждать. 401/403 — токен неверен: повторять часто бессмысленно."""
if isinstance(e, HttpError) and e.code in (401, 403):
log("хаб: HTTP %d — неверный hub_token в agent.json, повтор через 60 с" % e.code)
return 60
log("хаб: %s, повтор через %d с" % (e, base))
return base
def load_settings(self) -> None: def load_settings(self) -> None:
self.settings = self.hub.settings() self.settings = self.hub.settings()
self.executor = Executor(self.jira, self.settings, self.hub) self.executor = Executor(self.jira, self.settings, self.hub)
@ -777,7 +832,8 @@ class Runner:
или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка.""" или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка."""
cid = rec["command_id"] cid = rec["command_id"]
payload = {"status": rec["status"], "result": rec.get("result"), "error": rec.get("error"), 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)} "jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0),
"agent_version": AGENT_VERSION}
while True: while True:
try: try:
self.hub.result(cid, payload) self.hub.result(cid, payload)
@ -788,43 +844,65 @@ class Runner:
log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e)) log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e))
self.journal.acked(cid) self.journal.acked(cid)
return True return True
log("хаб: %s, повтор через 10 с" % e) wait = self.hub_wait(e, 10)
except ConnectionError as e: except ConnectionError as e:
log("хаб недоступен: %s, повтор через 10 с" % e) wait = self.hub_wait(e, 10)
if self.stop: if self.stop:
log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid) log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid)
return False return False
self.sleep(10) self.sleep(wait)
def wait_pause(self, pause_seconds) -> bool: def wait_pause(self, pause_seconds, force=False) -> bool:
"""Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы.""" """Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы.
force: если вызовов Jira ещё не было, считать от текущего момента."""
pause = float(pause_seconds or 0) pause = float(pause_seconds or 0)
if pause > 0 and self.jira.last_call_at is not None: if pause > 0 and (self.jira.last_call_at is not None or force):
remaining = pause - (time.monotonic() - self.jira.last_call_at) since = time.monotonic() - self.jira.last_call_at if self.jira.last_call_at is not None else 0
remaining = pause - since
if remaining > 0: if remaining > 0:
log(" пауза %.0f с до команды" % remaining) log(" пауза %.0f с" % remaining)
while remaining > 0 and not self.stop: while remaining > 0 and not self.stop:
step = min(remaining, 1.0) step = min(remaining, 1.0)
self.sleep(step) self.sleep(step)
remaining -= step remaining -= step
return not self.stop return not self.stop
def execute(self, cmd: dict) -> None: def run_once(self, cmd: dict) -> dict:
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: try:
out = self.executor.run(cmd) return self.executor.run(cmd)
except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление
log(" ! необработанная ошибка: %r" % e) log(" ! необработанная ошибка: %r" % e)
out = {"status": "failed", "result": None, return {"status": "failed", "result": None,
"error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]), "error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]),
"jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited} "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"])
def execute(self, cmd: dict) -> None:
cid = cmd["id"]
log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200]))
self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду
if not self.wait_pause(cmd.get("pause_seconds")):
return
self.journal.started(cid) # started — только после паузы
calls = limited = attempt = 0
while True:
out = self.run_once(cmd)
calls += out["jira_calls"]
limited += out["rate_limited"]
if not out.pop("retryable", False):
break
if attempt >= self.RETRY_ATTEMPTS:
out = {"status": "failed", "result": None,
"error": "результат неизвестен (обрыв связи с Jira), повтор безопасен: %s" % out.get("error")}
break
attempt += 1
log(" ! обрыв связи с Jira, повтор %d/%d через проверку: %s" % (attempt, self.RETRY_ATTEMPTS, out.get("error")))
if not self.wait_pause(max(float(cmd.get("pause_seconds") or 0), self.RETRY_MIN_WAIT), force=True):
return # остановка: finished не пишем, команду подхватит recover
out["jira_calls"], out["rate_limited"] = calls, limited
self.journal.finished(cid, out["status"], out.get("result"), out.get("error"), calls, limited)
log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False))) log(" → %s %s" % (out["status"], out.get("error") or json.dumps(out.get("result"), ensure_ascii=False)))
rec = dict(out) rec = dict(out)
rec["command_id"] = cmd["id"] rec["command_id"] = cid
self.send_result(rec) self.send_result(rec)
def fetch_command(self, cid): def fetch_command(self, cid):
@ -836,12 +914,12 @@ class Runner:
if e.code in (404, 409, 422): if e.code in (404, 409, 422):
log("#%s: хаб ответил %s" % (cid, e)) log("#%s: хаб ответил %s" % (cid, e))
return None return None
log("хаб: %s, повтор через 10 с" % e) wait = self.hub_wait(e, 10)
except ConnectionError as e: except ConnectionError as e:
log("хаб недоступен: %s, повтор через 10 с" % e) wait = self.hub_wait(e, 10)
if self.stop: if self.stop:
raise InterruptedError(cid) raise InterruptedError(cid)
self.sleep(10) self.sleep(wait)
def recover(self) -> None: def recover(self) -> None:
for rec in self.journal.unacked_finished(): for rec in self.journal.unacked_finished():
@ -861,10 +939,28 @@ class Runner:
log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status"))) log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status")))
self.journal.acked(cid) self.journal.acked(cid)
continue continue
log("#%s начата до перезапуска — выполняю заново через проверку" % cid) log("#%s выдана до перезапуска — выполняю через проверку" % cid)
cmd["pause_seconds"] = 0 cmd["pause_seconds"] = 0
self.execute(cmd) self.execute(cmd)
def reload_settings_before(self, cmd) -> bool:
"""Настройки сменились: перечитать ДО выполнения. False — остановка, команда не должна выполняться
(если она уже выдана хабом, фиксируем claimed — подхватит recover)."""
while True:
try:
self.load_settings()
return True
except (ConnectionError, HttpError) as e:
wait = self.hub_wait(e, max(self.poll, 1))
if self.stop:
if cmd:
self.journal.claimed(cmd["id"])
return False
if not cmd:
self.sleep(wait)
return False
self.sleep(wait)
def loop(self) -> None: def loop(self) -> None:
idle, recovered = 0, False idle, recovered = 0, False
while not self.stop: while not self.stop:
@ -872,8 +968,7 @@ class Runner:
try: try:
self.load_settings() self.load_settings()
except (ConnectionError, HttpError) as e: except (ConnectionError, HttpError) as e:
log("хаб: %s" % e) self.sleep(self.hub_wait(e, max(self.poll, 1)))
self.sleep(self.poll)
continue continue
if not recovered: if not recovered:
self.recover() self.recover()
@ -882,15 +977,12 @@ class Runner:
try: try:
resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso) resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso)
except (ConnectionError, HttpError) as e: except (ConnectionError, HttpError) as e:
log("хаб: %s" % e) self.sleep(self.hub_wait(e, max(self.poll, 1)))
self.sleep(self.poll)
continue 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") cmd = resp.get("command")
if resp.get("settings_version") != self.settings.get("settings_version"):
if not self.reload_settings_before(cmd):
continue
if cmd: if cmd:
idle = 0 idle = 0
self.execute(cmd) self.execute(cmd)
@ -905,11 +997,16 @@ def cmd_run(cfg: dict, journal_path: str) -> int:
runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path)) runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path))
def on_signal(signum, frame): def on_signal(signum, frame):
try:
if runner.stop:
log("повторный сигнал — выхожу немедленно")
else:
log("сигнал %s: доделываю текущую команду и выхожу" % signum)
except RuntimeError: # print внутри обработчика, прервавшего другой print (reentrant call)
pass
if runner.stop: if runner.stop:
log("повторный сигнал — выхожу немедленно")
sys.exit(130) sys.exit(130)
runner.stop = True runner.stop = True
log("сигнал %s: доделываю текущую команду и выхожу" % signum)
signal.signal(signal.SIGINT, on_signal) signal.signal(signal.SIGINT, on_signal)
signal.signal(signal.SIGTERM, on_signal) signal.signal(signal.SIGTERM, on_signal)

15
tests/fake_jira.py
View File

@ -2,6 +2,7 @@
только маршруты, которые использует агент. Семантика связей — как у настоящего Jira Server.""" только маршруты, которые использует агент. Семантика связей — как у настоящего Jira Server."""
import json import json
import re import re
import socket
import threading import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse from urllib.parse import parse_qs, urlparse
@ -31,6 +32,8 @@ class FakeJira:
self.id_seq = 10000 self.id_seq = 10000
self.fail_next = [] # [(код, {заголовки})] — следующие ответы с ошибкой self.fail_next = [] # [(код, {заголовки})] — следующие ответы с ошибкой
self.requests = [] # [(метод, путь)] self.requests = [] # [(метод, путь)]
self.queries = [] # параметры запросов (q) — для проверки JQL
self.drop_next_post_issue = False # выполнить создание задачи и оборвать соединение без ответа
self.lock = threading.Lock() self.lock = threading.Lock()
fake = self fake = self
@ -104,7 +107,8 @@ class FakeJira:
keys = [k.strip() for k in m.group(1).split(",")] keys = [k.strip() for k in m.group(1).split(",")]
return [k for k in keys if k in self.issues] return [k for k in keys if k in self.issues]
if "reporter = currentUser()" in (jql or ""): if "reporter = currentUser()" in (jql or ""):
return [k for k, f in self.issues.items() if f["reporter"] == "me"] keys = [k for k, f in self.issues.items() if f["reporter"] == "me"]
return keys[::-1] if "ORDER BY created DESC" in jql else keys
return [k for k, f in self.issues.items() if "AWPCA" in f["labels"]] return [k for k, f in self.issues.items() if "AWPCA" in f["labels"]]
# --- маршрутизация ---------------------------------------------------------- # --- маршрутизация ----------------------------------------------------------
@ -115,6 +119,7 @@ class FakeJira:
body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {} body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {}
with self.lock: with self.lock:
self.requests.append((method, url.path)) self.requests.append((method, url.path))
self.queries.append(q)
if self.fail_next: if self.fail_next:
code, headers = self.fail_next.pop(0) code, headers = self.fail_next.pop(0)
return self._send(h, code, {"errorMessages": ["имитация ошибки"]}, headers) return self._send(h, code, {"errorMessages": ["имитация ошибки"]}, headers)
@ -124,6 +129,11 @@ class FakeJira:
code, payload = 404, {"errorMessages": ["Issue Does Not Exist: %s" % e]} code, payload = 404, {"errorMessages": ["Issue Does Not Exist: %s" % e]}
except Exception as e: # некорректный запрос к фейку except Exception as e: # некорректный запрос к фейку
code, payload = 400, {"errorMessages": ["фейк: %s: %s" % (type(e).__name__, e)]} code, payload = 400, {"errorMessages": ["фейк: %s: %s" % (type(e).__name__, e)]}
if self.drop_next_post_issue and method == "POST" and url.path == "/rest/api/2/issue":
self.drop_next_post_issue = False
h.close_connection = True
h.connection.shutdown(socket.SHUT_RDWR) # задача создана, ответа клиент не получит
return
self._send(h, code, payload) self._send(h, code, payload)
def _send(self, h, code, payload, headers=None): def _send(self, h, code, payload, headers=None):
@ -193,7 +203,8 @@ class FakeJira:
self.issues[key] self.issues[key]
lst = self.comments.setdefault(key, []) lst = self.comments.setdefault(key, [])
if method == "GET": if method == "GET":
return 200, {"comments": lst} start, maxr = int(q.get("startAt", 0)), int(q.get("maxResults", 1000))
return 200, {"startAt": start, "maxResults": maxr, "total": len(lst), "comments": lst[start:start + maxr]}
self.comment_seq += 1 self.comment_seq += 1
lst.append({"id": str(self.comment_seq), "body": body["body"]}) lst.append({"id": str(self.comment_seq), "body": body["body"]})
return 201, lst[-1] return 201, lst[-1]

2
tests/test_py39.py
View File

@ -17,7 +17,7 @@ class Py39Test(unittest.TestCase):
def test_no_third_party_imports(self): def test_no_third_party_imports(self):
with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f: with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f:
tree = ast.parse(f.read()) tree = ast.parse(f.read())
allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "sys", "time", "urllib", "datetime", "typing"} allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "http", "sys", "time", "urllib", "datetime", "typing"}
for node in ast.walk(tree): for node in ast.walk(tree):
if isinstance(node, ast.Import): if isinstance(node, ast.Import):
for a in node.names: for a in node.names:

149
tests/test_runner.py
View File

@ -47,7 +47,7 @@ class RunnerTest(unittest.TestCase):
self.assertEqual(self.hub.results[2]["status"], "done") self.assertEqual(self.hub.results[2]["status"], "done")
self.assertEqual(len(self.hub.states), 1) self.assertEqual(len(self.hub.states), 1)
self.assertEqual(self.hub.states[0]["issues"][0]["duedate"], "2026-10-23") 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.kinds(), [("claimed", 1), ("started", 1), ("finished", 1), ("acked", 1), ("claimed", 2), ("started", 2), ("finished", 2), ("acked", 2)])
self.assertEqual(self.sleeps, []) # пауз не было: last_call_at пуст, потом pause 0 self.assertEqual(self.sleeps, []) # пауз не было: last_call_at пуст, потом pause 0
def test_pause_waits_from_last_jira_call(self): def test_pause_waits_from_last_jira_call(self):
@ -141,10 +141,41 @@ class RunnerTest(unittest.TestCase):
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=5) 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.sleep = lambda s: setattr(self.runner, "stop", True)
self.runner.loop() self.runner.loop()
self.assertEqual(self.journal.records(), []) self.assertEqual([r["kind"] for r in self.journal.records()], ["claimed"]) # started нет
self.assertEqual(self.hub.results, {}) self.assertEqual(self.hub.results, {})
self.assertEqual(self.hub.commands[1]["status"], "claimed") self.assertEqual(self.hub.commands[1]["status"], "claimed")
def test_stop_during_pause_then_restart_executes_once(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.unfinished_started(), [1])
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
second = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), agent.Journal(self.journal.path))
second.sleep = self.sleeps.append
second.idle_limit = 1
second.loop()
self.assertEqual(self.hub.results[1]["status"], "done")
self.assertEqual(len(self.fake.writes()), 1)
self.assertEqual(self.journal.unfinished_started(), [])
self.assertEqual(self.sleeps, []) # при восстановлении пауза не применяется
def test_stop_between_claim_and_execute_is_recorded(self):
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
real = self.runner.hub.claim
def claim_then_stop(*a):
resp = real(*a)
self.runner.stop = True
return resp
self.runner.hub.claim = claim_then_stop
self.runner.loop()
self.assertEqual([r["kind"] for r in self.journal.records()], ["claimed"])
self.assertEqual(self.fake.writes(), [])
self.assertEqual(self.journal.unfinished_started(), [1])
def test_stop_during_execution_finishes_and_reports(self): def test_stop_during_execution_finishes_and_reports(self):
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]}) self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
@ -159,7 +190,7 @@ class RunnerTest(unittest.TestCase):
self.runner.loop() self.runner.loop()
self.assertEqual(self.hub.results[1]["status"], "done") self.assertEqual(self.hub.results[1]["status"], "done")
self.assertNotIn(2, self.hub.results) # вторая команда не начата self.assertNotIn(2, self.hub.results) # вторая команда не начата
self.assertEqual(self.kinds(), [("started", 1), ("finished", 1), ("acked", 1)]) self.assertEqual(self.kinds(), [("claimed", 1), ("started", 1), ("finished", 1), ("acked", 1)])
def test_stop_during_execution_hub_down_single_attempt(self): def test_stop_during_execution_hub_down_single_attempt(self):
self.hub.fail_results = 5 self.hub.fail_results = 5
@ -231,7 +262,117 @@ class RunnerTest(unittest.TestCase):
self.runner.hub.settings = hub_settings self.runner.hub.settings = hub_settings
self.runner.loop() self.runner.loop()
self.assertEqual(len(calls), 3) self.assertEqual(len(calls), 3)
self.assertEqual(self.sleeps, [0, 0]) self.assertEqual(self.sleeps, [1, 1]) # пол паузы на ошибках хаба — 1 с
def test_retry_after_connection_drop_creates_one_issue(self):
self.fake.drop_next_post_issue = True
self.hub.enqueue("ensure_issue", {"temp_key": "new:1", "issue_type": "task", "summary": "Новая задача"})
self.runner.loop()
r = self.hub.results[1]
self.assertEqual(r["status"], "done")
self.assertEqual(r["result"]["existed"], True)
self.assertEqual(len([k for k in self.fake.issues if k != "AURORA-1"]), 1)
self.assertAlmostEqual(sum(self.sleeps), 30, delta=1.5) # ожидание 30 с перед повтором
self.assertEqual(r["jira_calls"], 2) # поиск + повторный поиск (оборванный ответ не считается)
self.assertEqual([k for k, _ in self.kinds()].count("started"), 1)
def test_retry_exhausted_gives_failed_unknown(self):
self.runner.load_settings()
runs = []
def down(cmd):
runs.append(1)
return {"status": "failed", "result": None, "error": "нет связи", "jira_calls": 1, "rate_limited": 0, "retryable": True}
self.runner.executor.run = down
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.runner.loop()
self.assertEqual(len(runs), 4) # первый запуск + 3 повтора
r = self.hub.results[1]
self.assertEqual(r["status"], "failed")
self.assertIn("результат неизвестен", r["error"])
self.assertEqual(r["jira_calls"], 4)
def test_stop_during_retry_wait_leaves_command_for_recovery(self):
self.runner.load_settings()
self.runner.executor.run = lambda cmd: {"status": "failed", "result": None, "error": "x", "jira_calls": 1,
"rate_limited": 0, "retryable": True}
self.runner.sleep = lambda s: setattr(self.runner, "stop", True)
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.runner.loop()
self.assertEqual(self.journal.unfinished_started(), [1])
self.assertEqual(self.hub.results, {})
def test_agent_version_in_result(self):
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
self.runner.loop()
self.assertEqual(self.hub.results[1]["agent_version"], agent.AGENT_VERSION)
def test_hub_auth_error_waits_60(self):
def denied(*a):
raise agent.HttpError(401, "unauthorized")
self.runner.hub.claim = denied
self.runner.sleep = lambda s: (self.sleeps.append(s), setattr(self.runner, "stop", True))
self.runner.loop()
self.assertEqual(self.sleeps, [60])
def test_wrong_hub_token_401_real_hub(self):
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token="bad")
r = agent.Runner(cfg, agent.HubClient(cfg), agent.JiraClient(cfg), self.journal)
r.sleep = lambda s: (self.sleeps.append(s), setattr(r, "stop", True))
r.loop()
self.assertEqual(self.sleeps, [60])
def test_settings_reload_failure_retried_before_command(self):
self.runner.load_settings()
self.hub.settings["settings_version"] = 2
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
real = self.runner.hub.settings
calls = []
def flaky():
calls.append(1)
if len(calls) < 3:
raise ConnectionError("нет связи")
return real()
self.runner.hub.settings = flaky
self.runner.loop()
self.assertEqual(len(calls), 3)
self.assertEqual(self.sleeps, [1, 1])
self.assertEqual(self.runner.settings["settings_version"], 2)
self.assertEqual(self.hub.results[1]["status"], "done")
def test_claim_timeout_at_least_45(self):
seen = []
real = agent.Http.call
agent.Http.call = lambda self_, m, p, body=None, params=None, timeout=None: (seen.append(timeout), {"command": None, "settings_version": 1})[1]
try:
agent.HubClient(jira_cfg(self.fake, hub_url="http://x", hub_token="t", timeout=5)).claim(1, 1, None)
finally:
agent.Http.call = real
self.assertEqual(seen, [45])
def test_signal_handler(self):
import signal
from unittest import mock
handlers = {}
seen = {}
def fake_loop(runner):
handlers[signal.SIGINT](signal.SIGINT, None)
seen["stop_after_first"] = runner.stop
with self.assertRaises(SystemExit) as cm:
handlers[signal.SIGINT](signal.SIGINT, None)
seen["code"] = cm.exception.code
cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token)
with mock.patch.object(agent.signal, "signal", lambda n, h: handlers.__setitem__(n, h)), \
mock.patch.object(agent.Runner, "loop", fake_loop):
agent.cmd_run(cfg, self.journal.path)
self.assertTrue(seen["stop_after_first"])
self.assertEqual(seen["code"], 130)
if __name__ == "__main__": if __name__ == "__main__":