diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 0bc527b..8b09ddf 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -11,7 +11,7 @@ Jira и хаба. Спека: `../jira-hub/openspec/changes/bridge-core/`, пл | HTTP | `Http`, `HttpError`, `JiraClient`, `HubClient`, `multipart_file`, `wire_filename` | запросы; `Http.send` — запрос с готовым телом (JSON-вызов `call` идёт через него); `JiraClient` считает вызовы, повторяет 429/503 по `Retry-After` (до 5 запросов, не более 4 ожиданий), хранит время последнего вызова; `JiraClient.upload` — вложение `multipart/form-data` (таймаут не меньше 120 с); `HubClient.file` — файл вложения у хаба (base64 в JSON) | `urllib` | | Папка | `FolderTransport` | транспорт `folder:` к хабу через общую папку: `req/.json` → реле хаба → `resp/.json`; ошибки как у `Http`; `HubClient` делегирует ему `call` (спека `jira-hub/openspec/changes/folder-transport/`) | `os`, `uuid` | | Журнал | `Journal`, `_parse_at` | дозапись `journal.jsonl`: `started` → `finished` → `acked`; восстановление; `prune` (из CLI — под блокировкой журнала) | файловая система | -| Исполнитель | `Executor` и `do_<тип>` | по одному методу на тип команды: читает Jira, пишет только разницу; `refresh` читает и отправляет состояние в хаб; `ensure_attachment` берёт файл вложения у хаба (`hub_file`: sha256 и размер, повреждение — `HubError` с повтором); доски — из параметров команд (`ensure_sprint.board_id`, `refresh.boards`, тип доски — `board_type`), `board_id` настроек — запасной | `JiraClient`, `HubClient` (`refresh` и файл в `ensure_attachment`), настройки хаба | +| Исполнитель | `Executor` и `do_<тип>` | по одному методу на тип команды: читает Jira, пишет только разницу; `refresh` читает и отправляет состояние в хаб, затем (v13, параметр `history`) журнал изменений Jira — `refresh_history`: изменившиеся с `history.since` задачи этого прогона и стартовая выборка `history.jql`, `search` с `expand=changelog`, `POST /agent/history` пачками по 500; ответ не 2xx Jira или хаба — `history.error`/`backfill_error`, `refresh` остаётся `done`; `ensure_attachment` берёт файл вложения у хаба (`hub_file`: sha256 и размер, повреждение — `HubError` с повтором); доски — из параметров команд (`ensure_sprint.board_id`, `refresh.boards`, тип доски — `board_type`), `board_id` настроек — запасной | `JiraClient`, `HubClient` (`refresh` и файл в `ensure_attachment`), настройки хаба | | Самообновление | `read_update_state`, `write_update_state`, `Updater`, `launcher_version` | `update.json`; после `claim` без команды — проверка `agent_release`, скачивание, sha256, компиляция, `selftest`, `staged`; подтверждение `ok` новой версией; без запускателя — только подсказка в лог | `HubClient`, `subprocess` | | Цикл | `Runner` | claim, пауза до команды, журнал, выполнение, отчёт до подтверждения, восстановление при старте; после `claim` без команды спрашивает `Updater`, готова версия — выход с `EXIT_UPDATE` (75) | `HubClient`, `JiraClient`, `Journal`, `Executor` | | CLI | `cmd_run`, `cmd_check`, `cmd_selftest`, `cmd_prune`, `main` | команды `run`, `check`, `selftest`, `prune` (`run` и `prune` берут `acquire_lock` на `<журнал>.lock`); обработка сигналов | `Runner`, `Journal` | diff --git a/README.md b/README.md index 59577e2..d1332eb 100644 --- a/README.md +++ b/README.md @@ -174,7 +174,7 @@ python3 agent.py prune | `ensure_link` | создаёт связь; конфликт без `replace` — `failed`; с `replace` сначала удаляет лишнее | | `ensure_sprint` | переносит в спринт доски `board_id` команды (без него — доски из настроек хаба; активные и будущие спринты) или в бэклог; доска без спринтов (канбан) — `failed` с текстом Jira | | `ensure_attachment` | ищет у задачи вложение с тем же именем и размером — есть: ничего не делает; нет: берёт файл у хаба (`/agent/files/{sha256}`), сверяет sha256 и размер и загружает в Jira (`multipart/form-data`); файл удалён с хаба — `failed` «загрузите заново»; повреждённый файл — повтор, как обрыв связи с хабом. Содержимое файла в журнал и лог не пишется | -| `refresh` | только читает Jira и отправляет состояние задач в хаб; ничего не пишет; доски с неизвестным типом (`boards`, `type = null`) — запрос `/rest/agile/1.0/board/{id}`, тип в результате `boards`; спринты — со всех scrum-досок | +| `refresh` | только читает Jira и отправляет состояние задач в хаб (с v13 — и журнал изменений по `history`, `POST /agent/history`); ничего не пишет; доски с неизвестным типом (`boards`, `type = null`) — запрос `/rest/agile/1.0/board/{id}`, тип в результате `boards`; спринты — со всех scrum-досок | Правила работы: - **Проверь, потом сделай.** Каждый тип сначала читает текущее состояние и записывает только разницу. @@ -203,6 +203,8 @@ python3 agent.py prune Версия 12 — доски проектов из параметров команд (`ensure_sprint.board_id`, `refresh.boards`), тип доски в результате `refresh`, спринты со всех scrum-досок; без этих параметров — `board_id` из настроек, как v11. Хаб с проектами выкатывается только после v12 (design хаба `projects`, решение 14). + Версия 13 — журнал изменений Jira в `refresh` (параметр `history`: `since` — изменившиеся задачи, `jql` — стартовая + загрузка), отправка `POST /agent/history`, итог в `result.history` (design хаба `change-history`). Если хаб назначил `min_agent_version` выше, он пишет в `agent_status.warning`: «обновите агента»; команды при этом выдаются по-прежнему. @@ -252,6 +254,10 @@ python3 agent.py prune `/issue/{key}/comment`. С v11 комментарий без `id` пропускается (строка в логе с ключом задачи), а если Jira не дала дочитать комментарии задачи (403 или 404), задача уходит без `comments` — хаб оставляет прежние; `refresh` продолжается. + С v13 после состояния `refresh` читает журнал изменений (если хаб передал `history`): по задачам этого прогона с + `updated` не раньше `history.since` и по стартовой выборке `history.jql`. Неполный журнал дочитывается + `GET /issue/{key}?expand=changelog`, всё ещё неполный — в логе «получено N из M» и в `result.history.incomplete`. + Ошибка Jira или хаба (не 2xx) на этом шаге не роняет `refresh`: текст в `result.history.error` или `backfill_error`. 5. **Сравнение текстов без различия переводов строк.** Описание и тело комментария сравниваются после замены `\r\n` на `\n` (Jira хранит правленное в UI с `\r\n`); отправляется текст команды как есть. Прочие различия (пробелы, завершающий перевод строки) — изменение. diff --git a/src/agent.py b/src/agent.py index 90e8d26..ca5f1db 100644 --- a/src/agent.py +++ b/src/agent.py @@ -33,7 +33,7 @@ import uuid from datetime import datetime, timezone from typing import Any, Callable, Dict, List, Optional, Tuple -AGENT_VERSION = 12 +AGENT_VERSION = 13 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__)) @@ -363,6 +363,9 @@ class HubClient(Http): def known_keys(self) -> List[str]: return self.call("GET", "/agent/known-keys")["keys"] + def history(self, payload: dict) -> dict: + return self.call("POST", "/agent/history", payload) + def file(self, sha256: str) -> dict: """Файл вложения из хранилища хаба: {file, size, content (base64)}; нет файла — HttpError 404 no_file.""" return self.call("GET", "/agent/files/%s" % sha256) @@ -1199,6 +1202,9 @@ class Executor: "neighbor_keys": [k for k in neighbor_keys if k in chunk_keys], "page": pages, "last_page": i + 100 >= len(issues)}) result = {"issues": len(raws), "neighbors": len(neighbors), "pages": pages} + history = self.refresh_history(p, cmd, everything) + if history is not None: + result["history"] = history if missing: result["missing"] = list(dict.fromkeys(missing))[:50] if types: @@ -1206,6 +1212,102 @@ class Executor: return self._out("done", result) + # --- журнал изменений (v13, spec agent «Журнал изменений») ----------------------------------------------- + HISTORY_BATCH = 500 + + def search_changelog(self, jql: str) -> List[dict]: + out, start = [], 0 + while True: + page = self.jira.get(API + "/search", jql=jql, fields="key", expand="changelog", startAt=start, maxResults=100) + got = page.get("issues") or [] + out += got + start += len(got) + if not got or start >= int(page.get("total") or 0): + return out + + def full_changelog(self, raw: dict, incomplete: List[str]) -> List[dict]: + """Журнал задачи из ответа search; неполный — дочитать GET /issue, всё ещё неполный — что есть, ключ в incomplete.""" + key, cl = raw["key"], raw.get("changelog") or {} + got = list(cl.get("histories") or []) + total = cl.get("total") + if isinstance(total, int) and total > len(got): + more = (self.jira.get(API + "/issue/%s" % key, fields="key", expand="changelog").get("changelog") or {}) + more_got = list(more.get("histories") or []) + if len(more_got) > len(got): + got = more_got + if total > len(got): + log("журнал %s: получено %d из %d" % (key, len(got), total)) + incomplete.append(key) + return got + + def refresh_history(self, p: dict, cmd: dict, fetched: List[dict]) -> Optional[dict]: + """Журнал после состояния: since — задачи этого refresh с updated не раньше since, jql — стартовая загрузка. + Ответ не 2xx Jira или хаба — текст в error (since, отправка) или backfill_error (jql), refresh остаётся done; + обрыв связи — как во всём refresh.""" + h = p.get("history") + if not isinstance(h, dict): + return None + out: Dict[str, Any] = {"issues": 0, "changes": 0, "incomplete": []} + names: Dict[str, Any] = {} + + def field_id_of(name: str) -> Optional[str]: + if "map" not in names: # один GET /field за refresh, только по нужде + names["map"] = {f.get("name"): f.get("id") for f in self.jira.get(API + "/field") or [] + if isinstance(f, dict)} + return names["map"].get(name) + + items: List[dict] = [] + read = set() + + def collect(raws: List[dict], since: Optional[datetime]) -> None: + for raw in raws: + if raw["key"] in read: + continue + read.add(raw["key"]) + items.extend(self.changelog_items(raw["key"], self.full_changelog(raw, out["incomplete"]), since, + field_id_of)) + + if h.get("since"): + since = _jira_time(h["since"]) + try: + if since is None: + raise ValueError("history.since — неразбираемое время: %r" % h["since"]) + changed = [] + for r in fetched: + upd = _jira_time(((r.get("fields") or {}).get("updated"))) + if upd is not None and upd >= since and r["key"] not in changed: + changed.append(r["key"]) + for i in range(0, len(changed), 100): + collect(self.search_changelog("key in (%s)" % ",".join(changed[i:i + 100])), since) + except HttpError as e: + out["error"] = "HTTP %d %s" % (e.code, e.text[:500]) + log("refresh: журнал не прочитан: %s" % out["error"]) + except ValueError as e: + out["error"] = str(e) + log("refresh: %s" % e) + if h.get("jql"): + try: + collect(self.search_changelog(h["jql"]), None) + except HttpError as e: + out["backfill_error"] = "HTTP %d %s" % (e.code, e.text[:500]) + log("refresh: стартовая загрузка журнала не прочитана: %s" % out["backfill_error"]) + sent_keys = set() + for i in range(0, len(items), self.HISTORY_BATCH): + chunk = items[i:i + self.HISTORY_BATCH] + try: + self.hub.history({"command_id": cmd.get("id"), "changes": chunk}) + except HttpError as e: + out["error"] = "хаб: HTTP %d %s" % (e.code, e.text[:500]) + log("refresh: журнал не принят: %s" % out["error"]) + break + except ConnectionError as e: + raise HubError(e) from None + out["changes"] += len(chunk) + sent_keys.update(c["key"] for c in chunk) + out["issues"] = len(sent_keys) + return out + + # --------------------------------------------------------------------------- # Цикл: claim → пауза → журнал → выполнение → отчёт до подтверждения; восстановление при старте # --------------------------------------------------------------------------- diff --git a/tests/fake_hub.py b/tests/fake_hub.py index 68d795c..ad9fb4d 100644 --- a/tests/fake_hub.py +++ b/tests/fake_hub.py @@ -3,6 +3,7 @@ import base64 import hashlib import json import re +import socket import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -28,6 +29,9 @@ class FakeHub: self.files = {} # sha256 → содержимое (хранилище файлов вложений хаба) self.file_requests = 0 self.claim_wait = 0 # с: при пустой очереди держать claim, как хаб + self.histories = [] # тела POST /agent/history + self.reject_history = None # (код, тело) — ответ на /agent/history ошибкой; "drop" — оборвать без ответа + self.posts = [] # порядок POST /agent/state и /agent/history self.lock = threading.Lock() hub = self @@ -81,6 +85,10 @@ class FakeHub: if h.headers.get("Authorization") != "Bearer " + self.token: return self._send(h, 401, {"error": "unauthorized"}) path = h.path.split("?")[0] + if path == "/agent/history" and self.reject_history == "drop": + h.close_connection = True + h.connection.shutdown(socket.SHUT_RDWR) # хаб недоступен: ответа клиент не получит + return if path == "/agent/claim" and self.claim_wait: end = time.monotonic() + self.claim_wait while time.monotonic() < end: # ждать без блокировки: другие запросы (enqueue) проходят @@ -123,7 +131,14 @@ class FakeHub: return 200, self.settings if path == "/agent/known-keys": return 200, {"keys": list(self.known)} + if path == "/agent/history": + self.posts.append(path) + if self.reject_history: + return self.reject_history + self.histories.append(body) + return 200, {"accepted": len(body.get("changes") or []), "duplicates": 0} if path == "/agent/state": + self.posts.append(path) self.states.append(body) return 200, {"issues": len(body.get("issues") or []), "links": len(body.get("links") or [])} m = re.match(r"^/agent/files/([^/]+)$", path) diff --git a/tests/test_history.py b/tests/test_history.py index 97c781b..f5b0920 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -107,5 +107,180 @@ class FakeChangelogTest(Base): self.assertTrue(any(f["name"] == "Плановое окончание" for f in self.jira.get(agent.API + "/field"))) +SINCE = "2026-10-22T08:45:00Z" +CHANGED = "2026-10-22T12:30:00.000+0300" # 09:30 UTC — позже SINCE +BASE = "project = AURORA AND labels = AWPCA" + + +class RefreshHistoryTest(Base): + def refresh(self, history=None, **params): + p = dict({"jql": BASE}, **params) + if history is not None: + p["history"] = history + return self.ex.run({"id": 5, "type": "refresh", "params": p}) + + def changelog_queries(self): + return [q for q in self.fake.queries if q.get("expand") == "changelog"] + + def sent(self): + return [c for b in self.hub.histories for c in b["changes"]] + + def test_nothing_changed_no_changelog_calls(self): # «Ничего не менялось» + for n in range(3): + self.fake.add("AURORA-%d" % n) + out = self.refresh({"since": SINCE}) + self.assertEqual(out["status"], "done") + self.assertEqual(out["result"]["history"], {"issues": 0, "changes": 0, "incomplete": []}) + self.assertEqual(self.changelog_queries(), []) + self.assertEqual(self.hub.histories, []) + + def test_two_changed_one_search(self): # «Две задачи изменились» + self.fake.add("X", updated=CHANGED) + self.fake.add("Y", updated=CHANGED) + self.fake.add("Z") + self.fake.add_history("X", "2026-10-21T12:00:00.000+0300", [_st("status", "a", "b")]) + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "b", "c"), _st("resolution", None, "Готово")]) + self.fake.add_history("Y", "2026-10-22T12:30:00.000+0300", [_st("description", "старое", "новое")]) + out = self.refresh({"since": SINCE}) + qs = self.changelog_queries() + self.assertEqual(len(qs), 1) + self.assertEqual(qs[0]["jql"], "key in (X,Y)") + self.assertEqual(qs[0]["fields"], "key") + self.assertEqual([(c["key"], c["seq"]) for c in self.sent()], [("X", 0), ("X", 1), ("Y", 0)]) + self.assertEqual(self.hub.histories[0]["command_id"], 5) + self.assertEqual(out["result"]["history"], {"issues": 2, "changes": 3, "incomplete": []}) + self.assertEqual(self.hub.posts, ["/agent/state", "/agent/history"]) # журнал — после состояния + + def test_backfill_by_jql(self): # «Стартовая загрузка» + jql = "(%s) AND sprint in openSprints() AND statusCategory != Done" % BASE + keys = ["AURORA-%d" % n for n in range(12)] + for k in keys: + self.fake.add(k) + self.fake.add_history(k, "2020-01-01T00:00:00.000+0300", [_st("status", "a", "b")]) + self.fake.jql_results[jql] = keys + out = self.refresh({"jql": jql}) + self.assertEqual([q["jql"] for q in self.changelog_queries()], [jql]) + self.assertEqual(len(self.sent()), 12) + self.assertEqual(out["result"]["history"], {"issues": 12, "changes": 12, "incomplete": []}) + + def test_since_and_jql_dedup(self): # Review Focus 4 + jql = "(%s) AND sprint in openSprints() AND statusCategory != Done" % BASE + self.fake.add("X", updated=CHANGED) + self.fake.add("Y") + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + self.fake.add_history("Y", "2020-01-01T00:00:00.000+0300", [_st("status", "a", "b")]) + self.fake.jql_results[jql] = ["X", "Y"] + out = self.refresh({"since": SINCE, "jql": jql}) + self.assertEqual(sorted(c["key"] for c in self.sent()), ["X", "Y"]) + self.assertEqual(out["result"]["history"]["changes"], 2) + + def test_neighbor_changed_included(self): # Review Focus 5 + self.fake.add("X") + self.fake.add("N", in_base=False, updated=CHANGED) + self.fake.links.append({"id": 1, "type": "ASUPR_Связанные", "source": "X", "target": "N"}) + self.fake.add_history("N", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + self.refresh({"since": SINCE}) + self.assertEqual([c["key"] for c in self.sent()], ["N"]) + + def test_jira_500_degrades(self): # «Ошибка Jira при чтении журнала» + self.fake.add("X", updated=CHANGED) + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + orig = self.fake.route + + def route(method, path, q, body): + if q.get("expand") == "changelog": + return 500, {"errorMessages": ["сбой"]} + return orig(method, path, q, body) + self.fake.route = route + out = self.refresh({"since": SINCE}) + self.assertEqual(out["status"], "done") + self.assertTrue(out["result"]["history"]["error"].startswith("HTTP 500")) + self.assertNotIn("retryable", out) + self.assertTrue(self.hub.states) + self.assertTrue(any("журнал" in l for l in self.logs)) + + def test_backfill_400(self): # «Ошибка стартовой загрузки» + jql = "bad jql" + self.fake.add("X", updated=CHANGED) + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + orig = self.fake.route + + def route(method, path, q, body): + if q.get("jql") == jql: + return 400, {"errorMessages": ["Значение не существует"]} + return orig(method, path, q, body) + self.fake.route = route + out = self.refresh({"since": SINCE, "jql": jql}) + h = out["result"]["history"] + self.assertEqual(out["status"], "done") + self.assertTrue(h["backfill_error"].startswith("HTTP 400")) + self.assertNotIn("error", h) + self.assertEqual([c["key"] for c in self.sent()], ["X"]) + + def test_incomplete(self): # «Неполный журнал» + self.fake.add("X", updated=CHANGED) + for n in range(250): + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + self.fake.changelog_cap = 100 + out = self.refresh({"since": SINCE}) + self.assertEqual(out["result"]["history"]["incomplete"], ["X"]) + self.assertEqual(len(self.sent()), 100) + self.assertTrue(any("X: получено 100 из 250" in l for l in self.logs)) + self.assertIn(("GET", "/rest/api/2/issue/X"), self.fake.requests) + + def test_truncated_search_completed_by_issue(self): + self.fake.add("X", updated=CHANGED) + for n in range(150): + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + self.fake.changelog_cap = 100 + self.fake.issue_changelog_cap = 150 + out = self.refresh({"since": SINCE}) + self.assertEqual(out["result"]["history"], {"issues": 1, "changes": 150, "incomplete": []}) + + def test_custom_field_lookup_once(self): # «Плановая дата без fieldId» + self.fake.add("X", updated=CHANGED) + self.fake.add("Y", updated=CHANGED) + for k in ("X", "Y"): + self.fake.add_history(k, "2026-10-22T12:10:00.000+0300", + [_st("Плановое окончание", "2026-10-25", "2026-10-30", fieldtype="custom")]) + self.refresh({"since": SINCE}) + self.assertEqual([c["field_id"] for c in self.sent()], ["customfield_10201", "customfield_10201"]) + self.assertEqual(self.fake.requests.count(("GET", "/rest/api/2/field")), 1) + + def test_hub_422_degrades(self): # Review Focus 3 + self.fake.add("X", updated=CHANGED) + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + self.hub.reject_history = (422, {"error": "bad_history", "detail": "changes[0].seq"}) + out = self.refresh({"since": SINCE}) + self.assertEqual(out["status"], "done") + self.assertTrue(out["result"]["history"]["error"].startswith("хаб: ")) + self.assertNotIn("retryable", out) + + def test_hub_down_fails(self): + self.fake.add("X", updated=CHANGED) + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + self.hub.reject_history = "drop" + out = self.refresh({"since": SINCE}) + self.assertEqual(out["status"], "failed") + self.assertTrue(out["retryable"]) + + def test_batches_of_500(self): + self.fake.add("X", updated=CHANGED) + for n in range(600): + self.fake.add_history("X", "2026-10-22T12:10:00.000+0300", [_st("status", "a", "b")]) + out = self.refresh({"since": SINCE}) + self.assertEqual([len(b["changes"]) for b in self.hub.histories], [500, 100]) + self.assertEqual(out["result"]["history"]["changes"], 600) + + def test_no_history_param(self): # «Хаб без журнала» + self.fake.add("X", updated=CHANGED) + out = self.refresh() + self.assertNotIn("history", out["result"]) + self.assertEqual(self.changelog_queries(), []) + + def test_version(self): + self.assertEqual(agent.AGENT_VERSION, 13) + + if __name__ == "__main__": unittest.main()