feat(agent): запись claimed до паузы (R-9), повтор при обрыве связи (R-10), agent_version, устойчивость цикла к сбоям хаба
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
4018f994dc
commit
6bcc6f611f
231
src/agent.py
231
src/agent.py
@ -11,6 +11,7 @@ Python 3.9. Спека: jira-hub/openspec/changes/bridge-core/specs/agent/spec.m
|
||||
"""
|
||||
import argparse
|
||||
import base64
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import signal
|
||||
@ -70,7 +71,7 @@ class Http:
|
||||
def __repr__(self):
|
||||
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
|
||||
if params:
|
||||
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:
|
||||
req.add_header("Content-Type", "application/json; charset=utf-8")
|
||||
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()
|
||||
status = resp.status
|
||||
except urllib.error.HTTPError as e:
|
||||
@ -90,12 +91,12 @@ class Http:
|
||||
try:
|
||||
headers = {k.lower(): v for k, v in e.headers.items()}
|
||||
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
|
||||
finally:
|
||||
e.close()
|
||||
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
|
||||
self._touch()
|
||||
try:
|
||||
@ -181,7 +182,7 @@ class HubClient(Http):
|
||||
|
||||
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,
|
||||
"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:
|
||||
return self.call("GET", "/agent/settings")
|
||||
@ -255,6 +256,10 @@ class Journal:
|
||||
log("журнал: строка %d не читается, пропускаю" % n)
|
||||
return good
|
||||
|
||||
def claimed(self, command_id) -> None:
|
||||
"""Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту."""
|
||||
self.append({"kind": "claimed", "command_id": int(command_id)})
|
||||
|
||||
def started(self, command_id) -> None:
|
||||
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]
|
||||
|
||||
def unfinished_started(self) -> List[int]:
|
||||
"""Порядковый учёт: started открывает id, finished/acked закрывает; порядок — первое открытие."""
|
||||
"""Порядковый учёт: claimed/started открывают id, finished/acked закрывает; порядок — первое открытие."""
|
||||
opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия
|
||||
for r in self.records():
|
||||
kind, cid = r.get("kind"), r["command_id"]
|
||||
if kind == "started":
|
||||
if kind in ("claimed", "started"):
|
||||
opened[cid] = True
|
||||
elif kind in ("finished", "acked") and cid in opened:
|
||||
opened[cid] = False
|
||||
@ -299,7 +304,7 @@ class Journal:
|
||||
cid = r["command_id"]
|
||||
if r.get("kind") == "acked":
|
||||
last_ack[cid] = i
|
||||
elif r.get("kind") in ("started", "finished"):
|
||||
elif r.get("kind") in ("claimed", "started", "finished"):
|
||||
last_open[cid] = i
|
||||
drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется
|
||||
for cid, i in last_ack.items():
|
||||
@ -344,6 +349,10 @@ def _date(value) -> Optional[str]:
|
||||
return (value or "")[:10] or None
|
||||
|
||||
|
||||
class ConfigGap(Exception):
|
||||
"""В настройках хаба нет того, что нужно команде (R-11): команда падает с понятным текстом, а не молча теряет поле."""
|
||||
|
||||
|
||||
class Executor:
|
||||
def __init__(self, jira: JiraClient, settings: dict, hub: Optional[HubClient] = None):
|
||||
self.jira, self.s, self.hub = jira, settings, hub
|
||||
@ -359,10 +368,16 @@ class Executor:
|
||||
return self._out("failed", error="неизвестный тип команды %s" % cmd.get("type"))
|
||||
try:
|
||||
return handler(cmd.get("params") or {}, cmd)
|
||||
except HttpError as e:
|
||||
return self._out("failed", error="HTTP %d %s" % (e.code, e.text[:1500]))
|
||||
except ConnectionError as e:
|
||||
except ConfigGap as 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:
|
||||
return {"status": status, "result": result, "error": error,
|
||||
@ -389,19 +404,36 @@ class Executor:
|
||||
it = (self.s.get("issue_types") or {}).get(kind) or {}
|
||||
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 {}
|
||||
if epic_key and ep.get("mode", "field") == "field" and ep.get("field_id"):
|
||||
return {ep["field_id"]: epic_key}
|
||||
return {}
|
||||
if ep.get("mode", "field") != "field":
|
||||
raise ConfigGap("epic_key передан, но epic_link.mode = %r (поддержан только «field»)" % ep.get("mode"))
|
||||
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 ----------------------------------------------------------
|
||||
def do_ensure_issue(self, p: dict, cmd: dict) -> dict:
|
||||
jql = "project = %s AND reporter = currentUser() AND created >= -7d" % self.s["project_key"]
|
||||
found = self.jira.get(API + "/search", jql=jql, fields="summary", maxResults=200)
|
||||
for it in found.get("issues", []):
|
||||
if (it.get("fields", {}).get("summary") or "") == p["summary"]:
|
||||
return self._out("done", {"key": it["key"], "existed": True})
|
||||
summary = (p["summary"] or "").strip()
|
||||
jql = "project = %s AND reporter = currentUser() AND created >= -7d ORDER BY created DESC" % self.s["project_key"]
|
||||
start = 0
|
||||
while start < 1000:
|
||||
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"]),
|
||||
"summary": p["summary"]}
|
||||
if p.get("description") is not None:
|
||||
@ -411,8 +443,8 @@ class Executor:
|
||||
if p.get("duedate"):
|
||||
fields[self.fid("duedate")] = p["duedate"]
|
||||
for logical in ("planned_start", "planned_end"):
|
||||
if p.get(logical) and self.has_field(logical):
|
||||
fields[self.fid(logical)] = p[logical]
|
||||
if p.get(logical):
|
||||
fields[self.need_field(logical)] = p[logical]
|
||||
if p.get("labels") is not None:
|
||||
fields[self.fid("labels")] = list(p["labels"])
|
||||
if p.get("assignee"):
|
||||
@ -437,17 +469,22 @@ class Executor:
|
||||
if "priority" in p:
|
||||
want = self.priority_value(p["priority"])
|
||||
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
|
||||
changed.append("priority")
|
||||
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]
|
||||
changed.append(logical)
|
||||
if "epic_key" in p:
|
||||
ep = self.s.get("epic_link") or {}
|
||||
if ep.get("field_id") and (cur.get(ep["field_id"]) or None) != (p["epic_key"] or None):
|
||||
fields[ep["field_id"]] = p["epic_key"]
|
||||
if "epic_key" in p and (p["epic_key"] or (self.s.get("epic_link") or {}).get("field_id")):
|
||||
fid = self.epic_field_id()
|
||||
if (cur.get(fid) or None) != (p["epic_key"] or None):
|
||||
fields[fid] = p["epic_key"]
|
||||
changed.append("epic_key")
|
||||
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
|
||||
@ -467,12 +504,19 @@ class Executor:
|
||||
# --- ensure_comment --------------------------------------------------------
|
||||
def do_ensure_comment(self, p: dict, cmd: dict) -> dict:
|
||||
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)
|
||||
existing = None
|
||||
for c in self.jira.get(API + "/issue/%s/comment" % key).get("comments", []):
|
||||
if marker in (c.get("body") or "")[:80]:
|
||||
existing = c
|
||||
existing, start = None, 0
|
||||
while existing is None:
|
||||
page = self.jira.get(API + "/issue/%s/comment" % key, startAt=start, maxResults=100)
|
||||
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
|
||||
if existing and (existing.get("body") or "") == body:
|
||||
return self._out("done", {"comment_id": existing["id"], "changed": False})
|
||||
@ -740,7 +784,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"), "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],
|
||||
"neighbor_keys": [k for k in neighbor_keys if k in chunk_keys],
|
||||
"page": pages, "last_page": i + 100 >= len(issues)})
|
||||
@ -754,6 +798,9 @@ class Executor:
|
||||
# Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте
|
||||
# ---------------------------------------------------------------------------
|
||||
class Runner:
|
||||
RETRY_ATTEMPTS = 3 # повторов команды после обрыва связи с Jira / 5xx (R-10)
|
||||
RETRY_MIN_WAIT = 30 # секунд до повтора
|
||||
|
||||
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]
|
||||
@ -766,6 +813,14 @@ class Runner:
|
||||
def poll(self) -> float:
|
||||
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:
|
||||
self.settings = self.hub.settings()
|
||||
self.executor = Executor(self.jira, self.settings, self.hub)
|
||||
@ -777,7 +832,8 @@ class Runner:
|
||||
или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка."""
|
||||
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)}
|
||||
"jira_calls": rec.get("jira_calls", 0), "rate_limited": rec.get("rate_limited", 0),
|
||||
"agent_version": AGENT_VERSION}
|
||||
while True:
|
||||
try:
|
||||
self.hub.result(cid, payload)
|
||||
@ -788,43 +844,65 @@ class Runner:
|
||||
log("хаб отверг отчёт по #%s: %s — считаю закрытым" % (cid, e))
|
||||
self.journal.acked(cid)
|
||||
return True
|
||||
log("хаб: %s, повтор через 10 с" % e)
|
||||
wait = self.hub_wait(e, 10)
|
||||
except ConnectionError as e:
|
||||
log("хаб недоступен: %s, повтор через 10 с" % e)
|
||||
wait = self.hub_wait(e, 10)
|
||||
if self.stop:
|
||||
log("остановка: отчёт по #%s не подтверждён, будет дослан при следующем старте" % cid)
|
||||
return False
|
||||
self.sleep(10)
|
||||
self.sleep(wait)
|
||||
|
||||
def wait_pause(self, pause_seconds) -> bool:
|
||||
"""Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы."""
|
||||
def wait_pause(self, pause_seconds, force=False) -> bool:
|
||||
"""Досыпать до зазора от последнего вызова Jira; False — остановлены во время паузы.
|
||||
force: если вызовов Jira ещё не было, считать от текущего момента."""
|
||||
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 pause > 0 and (self.jira.last_call_at is not None or force):
|
||||
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:
|
||||
log(" пауза %.0f с до команды" % remaining)
|
||||
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 — только после паузы
|
||||
def run_once(self, cmd: dict) -> dict:
|
||||
try:
|
||||
out = self.executor.run(cmd)
|
||||
return self.executor.run(cmd)
|
||||
except Exception as e: # баг в обработчике не должен ронять цикл и зацикливать восстановление
|
||||
log(" ! необработанная ошибка: %r" % e)
|
||||
out = {"status": "failed", "result": None,
|
||||
return {"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"])
|
||||
|
||||
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)))
|
||||
rec = dict(out)
|
||||
rec["command_id"] = cmd["id"]
|
||||
rec["command_id"] = cid
|
||||
self.send_result(rec)
|
||||
|
||||
def fetch_command(self, cid):
|
||||
@ -836,12 +914,12 @@ class Runner:
|
||||
if e.code in (404, 409, 422):
|
||||
log("#%s: хаб ответил %s" % (cid, e))
|
||||
return None
|
||||
log("хаб: %s, повтор через 10 с" % e)
|
||||
wait = self.hub_wait(e, 10)
|
||||
except ConnectionError as e:
|
||||
log("хаб недоступен: %s, повтор через 10 с" % e)
|
||||
wait = self.hub_wait(e, 10)
|
||||
if self.stop:
|
||||
raise InterruptedError(cid)
|
||||
self.sleep(10)
|
||||
self.sleep(wait)
|
||||
|
||||
def recover(self) -> None:
|
||||
for rec in self.journal.unacked_finished():
|
||||
@ -861,10 +939,28 @@ class Runner:
|
||||
log("#%s уже %s в хабе — пропускаю" % (cid, cmd.get("status")))
|
||||
self.journal.acked(cid)
|
||||
continue
|
||||
log("#%s начата до перезапуска — выполняю заново через проверку" % cid)
|
||||
log("#%s выдана до перезапуска — выполняю через проверку" % cid)
|
||||
cmd["pause_seconds"] = 0
|
||||
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:
|
||||
idle, recovered = 0, False
|
||||
while not self.stop:
|
||||
@ -872,8 +968,7 @@ class Runner:
|
||||
try:
|
||||
self.load_settings()
|
||||
except (ConnectionError, HttpError) as e:
|
||||
log("хаб: %s" % e)
|
||||
self.sleep(self.poll)
|
||||
self.sleep(self.hub_wait(e, max(self.poll, 1)))
|
||||
continue
|
||||
if not recovered:
|
||||
self.recover()
|
||||
@ -882,15 +977,12 @@ class Runner:
|
||||
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)
|
||||
self.sleep(self.hub_wait(e, max(self.poll, 1)))
|
||||
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 resp.get("settings_version") != self.settings.get("settings_version"):
|
||||
if not self.reload_settings_before(cmd):
|
||||
continue
|
||||
if cmd:
|
||||
idle = 0
|
||||
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))
|
||||
|
||||
def on_signal(signum, frame):
|
||||
try:
|
||||
if runner.stop:
|
||||
log("повторный сигнал — выхожу немедленно")
|
||||
else:
|
||||
log("сигнал %s: доделываю текущую команду и выхожу" % signum)
|
||||
except RuntimeError: # print внутри обработчика, прервавшего другой print (reentrant call)
|
||||
pass
|
||||
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)
|
||||
|
||||
@ -2,6 +2,7 @@
|
||||
только маршруты, которые использует агент. Семантика связей — как у настоящего Jira Server."""
|
||||
import json
|
||||
import re
|
||||
import socket
|
||||
import threading
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
@ -31,6 +32,8 @@ class FakeJira:
|
||||
self.id_seq = 10000
|
||||
self.fail_next = [] # [(код, {заголовки})] — следующие ответы с ошибкой
|
||||
self.requests = [] # [(метод, путь)]
|
||||
self.queries = [] # параметры запросов (q) — для проверки JQL
|
||||
self.drop_next_post_issue = False # выполнить создание задачи и оборвать соединение без ответа
|
||||
self.lock = threading.Lock()
|
||||
fake = self
|
||||
|
||||
@ -104,7 +107,8 @@ class FakeJira:
|
||||
keys = [k.strip() for k in m.group(1).split(",")]
|
||||
return [k for k in keys if k in self.issues]
|
||||
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"]]
|
||||
|
||||
# --- маршрутизация ----------------------------------------------------------
|
||||
@ -115,6 +119,7 @@ class FakeJira:
|
||||
body = json.loads(h.rfile.read(length).decode("utf-8")) if length else {}
|
||||
with self.lock:
|
||||
self.requests.append((method, url.path))
|
||||
self.queries.append(q)
|
||||
if self.fail_next:
|
||||
code, headers = self.fail_next.pop(0)
|
||||
return self._send(h, code, {"errorMessages": ["имитация ошибки"]}, headers)
|
||||
@ -124,6 +129,11 @@ class FakeJira:
|
||||
code, payload = 404, {"errorMessages": ["Issue Does Not Exist: %s" % e]}
|
||||
except Exception as 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)
|
||||
|
||||
def _send(self, h, code, payload, headers=None):
|
||||
@ -193,7 +203,8 @@ class FakeJira:
|
||||
self.issues[key]
|
||||
lst = self.comments.setdefault(key, [])
|
||||
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
|
||||
lst.append({"id": str(self.comment_seq), "body": body["body"]})
|
||||
return 201, lst[-1]
|
||||
|
||||
@ -17,7 +17,7 @@ class Py39Test(unittest.TestCase):
|
||||
def test_no_third_party_imports(self):
|
||||
with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f:
|
||||
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):
|
||||
if isinstance(node, ast.Import):
|
||||
for a in node.names:
|
||||
|
||||
@ -47,7 +47,7 @@ class RunnerTest(unittest.TestCase):
|
||||
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.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
|
||||
|
||||
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.runner.sleep = lambda s: setattr(self.runner, "stop", True)
|
||||
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.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):
|
||||
self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
|
||||
self.hub.enqueue("refresh", {"keys": ["AURORA-1"]})
|
||||
@ -159,7 +190,7 @@ class RunnerTest(unittest.TestCase):
|
||||
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)])
|
||||
self.assertEqual(self.kinds(), [("claimed", 1), ("started", 1), ("finished", 1), ("acked", 1)])
|
||||
|
||||
def test_stop_during_execution_hub_down_single_attempt(self):
|
||||
self.hub.fail_results = 5
|
||||
@ -231,7 +262,117 @@ class RunnerTest(unittest.TestCase):
|
||||
self.runner.hub.settings = hub_settings
|
||||
self.runner.loop()
|
||||
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__":
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user