diff --git a/openspec/changes/eva-agent-mvp/design.md b/openspec/changes/eva-agent-mvp/design.md index c17844f..5bb26d3 100644 --- a/openspec/changes/eva-agent-mvp/design.md +++ b/openspec/changes/eva-agent-mvp/design.md @@ -162,3 +162,18 @@ Issue агента — aep/eva-agent#1. Версия агента — 1. Аге - В `tests/test_py39.py` в белый список stdlib добавлен `html` (`html.parser`). Обоснование: форматы из Context design хаба. Цена ошибки: потеря/дубль вложения, ловится на живой синхронизации. + +26. **Обход дерева `wiki_sync` ведётся по коду страницы, а не по `id` (уточнение решения 5).** + - `children`, `seen` и `parent_code` ключуются кодом (`ref.code`): код уникален и есть в каждой ссылке `parent`/`tree_parent`; `id` берётся только для `eva_id`. + - Соседи сортируются по `(orderno is None, orderno, name без регистра, code)` — как `collect_subtree` экспорта; решение 5 говорит «name», хвост по коду добавлен для детерминизма. + - Дерево с проектом — один список документов проекта; без проекта — обход в ширину через `tree_parent.code` (защита от циклов тоже по коду). + + Обоснование: ссылка Eva в фейке и в экспорте гарантированно несёт `code`; поле `id` в ссылке может не совпасть с `id` документа. Цена ошибки: если коды в Eva неуникальны (не бывает), страницы склеятся. +27. **Страница без названия получает `title = code`; корень без названия — `root_title = code`.** + Обоснование: `stage._check` хаба отклоняет пустой `title` (422 `<код>: title`), и одна такая страница завалила бы всю синхронизацию. Цена ошибки: в хабе страница названа кодом вместо пустого имени. +28. **Страница, исчезнувшая между списком и чтением (`get_doc` → `None`), — `failed` «страница <код> не найдена или нет прав»; частичная пропуск-синхронизация не делается.** + Обоснование: отчёт `codes` обязан совпадать со стадией хаба; тихий пропуск скрыл бы потерю. Цена ошибки: из-за одной удалённой на лету страницы нужна повторная синхронизация. +29. **Размер пачки считается тем же кодированием, что у `Http.call` (`ensure_ascii=False`, UTF-8) + 2 байта на разделитель; страница больше 4 МБ уходит одна пачкой (предел хаба 16 МБ).** + Обоснование: оценка совпадает с реальным телом запроса. Цена ошибки: пачка чуть больше 4 МБ; хаб примет до 16 МБ. +30. **`fake_hub.POST /agent/wiki/pages` проверяет пачку как `stage.accept`/`_check` хаба (≤ 50 страниц, `command_id` целое, команда `wiki_sync` в статусе `claimed`, типы и пределы полей, форма `attachments`/`links`) и отвечает `{error, detail}` с 422/404/409.** Тесты `test_folder.py` приведены к валидным страницам и выданной команде. + Обоснование: поручение владельца — фейк не мягче хаба. Цена ошибки: фейк строже или мягче хаба в мелочах; ловится e2e. diff --git a/src/agent.py b/src/agent.py index e77c3dd..2ad72f4 100644 --- a/src/agent.py +++ b/src/agent.py @@ -663,6 +663,71 @@ def _parse_at(value) -> float: return dt.timestamp() +# --------------------------------------------------------------------------- +# Дерево страниц Eva +# --------------------------------------------------------------------------- +MAX_PAGES = 50 # страниц в пачке (предел хаба, stage.MAX_PAGES) +MAX_BATCH_BYTES = 4 * 1024 * 1024 # JSON пачки; предел тела хаба — 16 МБ +TREE_FIELDS = ["id", "code", "name", "parent", "tree_parent", "project", "orderno"] + + +def _ref_code(ref) -> Optional[str]: + """Код из ссылки Eva {"id", "code"}; страница без родителя — None.""" + code = ref.get("code") if isinstance(ref, dict) else None + return code if isinstance(code, str) and code else None + + +def _orderno(value) -> Optional[int]: + if isinstance(value, bool): + return None + if isinstance(value, float) and value.is_integer(): + return int(value) + return value if isinstance(value, int) else None + + +def build_children(eva: "EvaClient", root: dict) -> Dict[str, List[dict]]: + """{код родителя: [дети]} для поддерева корня. С проектом — один список документов проекта и связи parent/tree_parent + (tree_parent в Eva бывает не заполнен); без проекта — обход в ширину через фильтр tree_parent.code.""" + children = {} # type: Dict[str, List[dict]] + project = _ref_code(root.get("project")) + if project: + for d in eva.list_docs([["project.code", "==", project]], TREE_FIELDS): + for pc in {_ref_code(d.get("parent")), _ref_code(d.get("tree_parent"))} - {None}: + children.setdefault(pc, []).append(d) + else: + queue, seen = [root], {root.get("code")} + while queue: + node = queue.pop(0) + kids = eva.list_docs([["tree_parent.code", "==", node["code"]]], TREE_FIELDS) + children[node["code"]] = kids + for k in kids: + if k.get("code") not in seen: + seen.add(k.get("code")) + queue.append(k) + for pc, kids in children.items(): + uniq = {k["code"]: k for k in kids if k.get("code")} + children[pc] = sorted(uniq.values(), key=lambda k: (_orderno(k.get("orderno")) is None, + _orderno(k.get("orderno")) or 0, + (k.get("name") or "").lower(), k["code"])) + return children + + +def walk(root: dict, children: Dict[str, List[dict]]): + """Обход в глубину, корень первым: (документ, код родителя по обходу). Каждый код — один раз (циклы, два родителя).""" + seen = set() # type: set + stack = [(root, None)] + while stack: + doc, parent = stack.pop() + code = doc["code"] + if code in seen: + continue + seen.add(code) + yield doc, parent + for kid in reversed(children.get(code, [])): + if kid["code"] not in seen: + stack.append((kid, code)) + + # --------------------------------------------------------------------------- # Исполнитель: команда wiki_sync / wiki_fetch (читает Eva, отдаёт хабу) # --------------------------------------------------------------------------- @@ -713,7 +778,44 @@ class Executor: raise HubError(e) from None def do_wiki_sync(self, p: dict, cmd: dict) -> dict: - return self._out("failed", error="wiki_sync: не реализовано") + """Дерево страниц Eva от корня → хаб пачками; отчёт — коды в порядке обхода (корень первым). Eva только читается, + поэтому после перезапуска команда выполняется заново целиком (design 9).""" + root_code = str(p.get("root") or "").strip() + if not root_code: + return self._out("failed", error="wiki_sync: не задан root") + root = self.eva.get_doc(root_code, TREE_FIELDS + ["text"]) + if not root: + return self._out("failed", error="страница %s не найдена или нет прав" % root_code) + children = build_children(self.eva, root) + root_code = root.get("code") or root_code + base = self.eva.base + codes = [] # type: List[str] + batch = [] # type: list + size = 0 + for doc, parent_code in walk(root, children): + code = doc["code"] + full = root if code == root_code else self.eva.get_doc(code, ["code", "name", "text"]) + if not full: + return self._out("failed", error="страница %s не найдена или нет прав" % code) + html = full.get("text") if isinstance(full.get("text"), str) else "" + attachments, links = classify_links(html, base) + page = {"code": code, "eva_id": str(doc["id"]) if doc.get("id") is not None else None, + # хаб отклоняет пустой title (stage._check): без названия страница называется своим кодом + "title": (full.get("name") or doc.get("name") or "").strip() or code, + "parent_code": parent_code, "orderno": _orderno(doc.get("orderno")), + "url": urllib.parse.urljoin(base + "/", "project/Document/" + code), + "html": html, "attachments": attachments, "links": links} + psize = len(json.dumps(page, ensure_ascii=False).encode("utf-8")) + 2 + if batch and (len(batch) >= MAX_PAGES or size + psize > MAX_BATCH_BYTES): + self._hub(self.hub.wiki_pages, cmd["id"], batch) + batch, size = [], 0 + batch.append(page) + size += psize + codes.append(code) + if batch: + self._hub(self.hub.wiki_pages, cmd["id"], batch) + title = (root.get("name") or "").strip() or root_code + return self._out("done", {"root_title": title, "codes": codes, "eva_calls": self.eva.calls}) def do_wiki_fetch(self, p: dict, cmd: dict) -> dict: return self._out("failed", error="wiki_fetch: не реализовано") @@ -1001,7 +1103,7 @@ class Runner: 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))[:300])) rec = dict(out) rec["command_id"] = cid self.send_result(rec) diff --git a/tests/fake_hub.py b/tests/fake_hub.py index eabe8f8..d8337c1 100644 --- a/tests/fake_hub.py +++ b/tests/fake_hub.py @@ -9,6 +9,55 @@ import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +MAX_PAGES = 50 + + +def check_pages(command, command_id, pages): + """Форма пачки как в stage.accept/_check хаба: (код, тело ошибки) или None. 422 — {"error", "detail"}.""" + def bad(detail, status=422, error="bad_pages"): + return status, {"error": error, "detail": detail} + + if isinstance(command_id, bool) or not isinstance(command_id, int): + return bad("command_id — целое", error="bad_request") + if not isinstance(pages, list) or not pages or len(pages) > MAX_PAGES: + return bad("pages — список от 1 до %d страниц" % MAX_PAGES) + if command is None: + return bad("command %s" % command_id, 404, "not_found") + if command["type"] != "wiki_sync" or command["status"] != "claimed": + return bad("команда %s не выдана агенту как wiki_sync" % command_id, 409, "not_claimed") + limits = {"code": 200, "title": 1000, "url": 4000, "parent_code": 200, "eva_id": 200} + for i, page in enumerate(pages, 1): + if not isinstance(page, dict): + return bad("страница %d: не объект" % i) + code = page.get("code") + if not isinstance(code, str) or not code.strip() or len(code) > limits["code"]: + return bad("страница %d: code" % i) + for key, required in (("title", True), ("parent_code", False), ("url", False)): + v = page.get(key) + if v is None and not required: + continue + if not isinstance(v, str) or (required and not v.strip()) or len(v) > limits[key]: + return bad("%s: %s" % (code, key)) + if not isinstance(page.get("html"), str): + return bad("%s: html" % code) + o = page.get("orderno") + if o is not None and (not isinstance(o, int) or isinstance(o, bool)): + return bad("%s: orderno" % code) + e = page.get("eva_id") + if e is not None and (not isinstance(e, str) or len(e) > limits["eva_id"]): + return bad("%s: eva_id" % code) + att, links = page.get("attachments") or [], page.get("links") or [] + if not isinstance(att, list) or not all( + isinstance(a, dict) and isinstance(a.get("ref"), str) and isinstance(a.get("url"), str) + and isinstance(a.get("name") or "", str) for a in att): + return bad("%s: attachments" % code) + if not isinstance(links, list) or not all( + isinstance(l, dict) and isinstance(l.get("ref"), str) and isinstance(l.get("code"), str) + for l in links): + return bad("%s: links" % code) + return None + + class FakeHub: token = "agent-token" @@ -126,6 +175,10 @@ class FakeHub: if path == "/agent/wiki/pages" and method == "POST": if self.reject_pages: return self.reject_pages + cid = body.get("command_id") + bad = check_pages(self.commands.get(cid) if isinstance(cid, int) else None, cid, body.get("pages")) + if bad: + return bad self.pages.append(body) return 200, {"accepted": len(body.get("pages") or [])} if path == "/agent/files" and method == "POST": diff --git a/tests/test_folder.py b/tests/test_folder.py index 0e3ff9e..42f46ee 100644 --- a/tests/test_folder.py +++ b/tests/test_folder.py @@ -138,18 +138,26 @@ class FolderTransportTest(unittest.TestCase): self.assertIsNone(hub.folder) self.assertEqual(hub.settings()["settings_version"], 1) + def claimed_sync(self): + cid = self.hub.enqueue("wiki_sync", {"label": "x", "root": "DOC-1"}) + self.hub.queue.remove(cid) + self.hub.commands[cid]["status"] = "claimed" + return cid + def test_wiki_pages_through_folder(self): self.run_relay() - pages = [{"code": "DOC-1", "name": "Корень", "html": "

" + "я" * 200000 + "

"}] - got = self.client().wiki_pages(5, pages) + cid = self.claimed_sync() + pages = [{"code": "DOC-1", "title": "Корень", "html": "

" + "я" * 200000 + "

"}] + got = self.client().wiki_pages(cid, pages) self.assertEqual(got, {"accepted": 1}) - self.assertEqual(self.hub.pages, [{"command_id": 5, "pages": pages}]) + self.assertEqual(self.hub.pages, [{"command_id": cid, "pages": pages}]) self.assertEqual(self.relay.requests[0]["path"], "/agent/wiki/pages") def test_wiki_pages_over_http(self): hub = agent.HubClient({"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5}) - self.assertEqual(hub.wiki_pages(5, [{"code": "DOC-1"}]), {"accepted": 1}) - self.assertEqual(self.hub.pages[0]["command_id"], 5) + cid = self.claimed_sync() + self.assertEqual(hub.wiki_pages(cid, [{"code": "DOC-1", "title": "T", "html": ""}]), {"accepted": 1}) + self.assertEqual(self.hub.pages[0]["command_id"], cid) def test_wiki_pages_rejected_is_http_error(self): self.hub.reject_pages = (422, {"error": "bad_page"}) diff --git a/tests/test_wiki_sync.py b/tests/test_wiki_sync.py new file mode 100644 index 0000000..3a0a086 --- /dev/null +++ b/tests/test_wiki_sync.py @@ -0,0 +1,199 @@ +import json +import os +import tempfile +import unittest + +from _common import SETTINGS, agent, eva_cfg, no_sleep +from fake_eva import FakeEva +from fake_hub import FakeHub + + +class SyncBase(unittest.TestCase): + def setUp(self): + self.eva = FakeEva() + self.hub = FakeHub(SETTINGS) + self.addCleanup(self.eva.stop) + self.addCleanup(self.hub.stop) + self.cfg = eva_cfg(self.eva.base_url, hub_url=self.hub.base_url, hub_token=self.hub.token) + self.client = agent.EvaClient(self.cfg) + self.client.sleep = no_sleep + self.ex = agent.Executor(self.client, agent.HubClient(self.cfg), dict(SETTINGS)) + + def sync(self, root="DOC-1", label="tz"): + cid = self.hub.enqueue("wiki_sync", {"label": label, "root": root}) + self.hub.queue.remove(cid) + self.hub.commands[cid]["status"] = "claimed" + self.cid = cid + return self.ex.run({"id": cid, "type": "wiki_sync", "params": {"label": label, "root": root}}) + + def sent(self): + return [p for body in self.hub.pages for p in body["pages"]] + + +class WikiSyncTest(SyncBase): + def test_sync_tree_order(self): + e = self.eva + e.add_doc(1, "DOC-1", "Корень", project="P", text="

корень

") + e.add_doc(2, "DOC-2", "Б", parent="DOC-1", project="P", orderno=2, text="

б

") + e.add_doc(3, "DOC-3", "А", parent="DOC-1", project="P", orderno=1, text="

а

") + e.add_doc(4, "DOC-4", "Внук", parent="DOC-3", project="P", text="

внук

") + e.add_doc(5, "DOC-5", "Чужой", parent="DOC-1", project="OTHER") + out = self.sync() + self.assertEqual(out["status"], "done", out) + self.assertEqual(out["result"]["codes"], ["DOC-1", "DOC-3", "DOC-4", "DOC-2"]) + self.assertEqual(out["result"]["root_title"], "Корень") + self.assertEqual(out["result"]["eva_calls"], self.client.calls) + pages = {p["code"]: p for p in self.sent()} + self.assertEqual([p["code"] for p in self.sent()], out["result"]["codes"]) + self.assertIsNone(pages["DOC-1"]["parent_code"]) + self.assertEqual(pages["DOC-3"]["parent_code"], "DOC-1") + self.assertEqual(pages["DOC-4"]["parent_code"], "DOC-3") + self.assertEqual(pages["DOC-3"]["orderno"], 1) + self.assertIsNone(pages["DOC-4"]["orderno"]) + self.assertEqual(pages["DOC-3"]["url"], self.eva.base_url + "/project/Document/DOC-3") + self.assertEqual(pages["DOC-3"]["eva_id"], "3") + self.assertEqual(pages["DOC-3"]["title"], "А") + self.assertEqual(pages["DOC-3"]["html"], "

а

") + self.assertEqual(self.hub.pages[0]["command_id"], self.cid) + + def test_sync_passes_links_and_attachments(self): + self.eva.add_doc(1, "DOC-1", "К", project="P", text=( + '

xСхема

')) + out = self.sync() + self.assertEqual(out["status"], "done", out) + page = self.sent()[0] + self.assertEqual(page["links"], [{"ref": "/project/Document/DOC-7", "code": "DOC-7"}]) + self.assertEqual(page["attachments"], [{"ref": "/files/a/b.png", "url": self.eva.base_url + "/files/a/b.png", + "name": "Схема"}]) + + def test_sync_batches_50(self): + self.eva.add_doc(1, "DOC-1", "Корень", project="P") + for i in range(2, 121): + self.eva.add_doc(i, "DOC-%d" % i, "С%03d" % i, parent="DOC-1", project="P", orderno=i) + out = self.sync() + self.assertEqual(out["status"], "done", out) + self.assertEqual([len(b["pages"]) for b in self.hub.pages], [50, 50, 20]) + self.assertEqual(len(out["result"]["codes"]), 120) + self.assertEqual(out["result"]["codes"][0], "DOC-1") + + def test_sync_batch_bytes(self): + big = "

" + "x" * (1500 * 1024) + "

" + self.eva.add_doc(1, "DOC-1", "Корень", project="P", text=big) + for i in (2, 3): + self.eva.add_doc(i, "DOC-%d" % i, "С", parent="DOC-1", project="P", orderno=i, text=big) + out = self.sync() + self.assertEqual(out["status"], "done", out) + sizes = [len(json.dumps(b, ensure_ascii=False).encode("utf-8")) for b in self.hub.pages] + self.assertTrue(all(s <= 4 * 1024 * 1024 for s in sizes), sizes) + self.assertEqual([len(b["pages"]) for b in self.hub.pages], [2, 1]) + + def test_tree_cycle_and_two_parents(self): + e = self.eva + e.add_doc(1, "DOC-1", "Корень", project="P") + e.add_doc(3, "DOC-3", "Три", parent="DOC-1", project="P", orderno=1) + e.add_doc(10, "DOC-A", "А", parent="DOC-B", tree_parent="DOC-1", project="P", orderno=2) + e.add_doc(11, "DOC-B", "Б", tree_parent="DOC-A", project="P") + e.add_doc(12, "DOC-C", "Ц", parent="DOC-1", tree_parent="DOC-3", project="P", orderno=3) + out = self.sync() + self.assertEqual(out["status"], "done", out) + codes = out["result"]["codes"] + self.assertEqual(len(codes), len(set(codes))) + self.assertEqual(sorted(codes), ["DOC-1", "DOC-3", "DOC-A", "DOC-B", "DOC-C"]) + self.assertEqual([p["code"] for p in self.sent()], codes) + + def test_root_without_project(self): + e = self.eva + e.add_doc(1, "DOC-1", "Корень", text="

к

") + e.add_doc(2, "DOC-2", "Б", tree_parent="DOC-1", orderno=2) + e.add_doc(3, "DOC-3", "А", tree_parent="DOC-1", orderno=1) + e.add_doc(4, "DOC-4", "Внук", tree_parent="DOC-3") + e.add_doc(5, "DOC-5", "Посторонний", tree_parent="DOC-9") + out = self.sync() + self.assertEqual(out["status"], "done", out) + self.assertEqual(out["result"]["codes"], ["DOC-1", "DOC-3", "DOC-4", "DOC-2"]) + self.assertEqual({p["code"]: p["parent_code"] for p in self.sent()}, + {"DOC-1": None, "DOC-3": "DOC-1", "DOC-4": "DOC-3", "DOC-2": "DOC-1"}) + + def test_root_missing(self): + out = self.sync("DOC-9") + self.assertEqual(out["status"], "failed") + self.assertEqual(out["error"], "страница DOC-9 не найдена или нет прав") + self.assertEqual(self.hub.pages, []) + + def test_eva_failure_reported(self): + self.eva.add_doc(1, "DOC-1", "Корень", project="P") + self.eva.fail_next(503, 4) + out = self.sync() + self.assertEqual(out["status"], "failed") + self.assertIn("503", out["error"]) + self.assertNotIn(FakeEva.token, out["error"]) + self.assertEqual(self.hub.pages, []) + + def test_hub_rejects_batch(self): + self.eva.add_doc(1, "DOC-1", "Корень", project="P") + self.hub.reject_pages = (422, {"error": "bad_pages", "detail": "DOC-1: html"}) + out = self.sync() + self.assertEqual(out["status"], "failed") + self.assertIn("422", out["error"]) + self.assertIn("DOC-1: html", out["error"]) + + def test_empty_name_falls_back_to_code(self): + self.eva.add_doc(1, "DOC-1", "", project="P") # хаб отклоняет пустой title (stage._check) + out = self.sync() + self.assertEqual(out["status"], "done", out) + self.assertEqual(self.sent()[0]["title"], "DOC-1") + self.assertEqual(out["result"]["root_title"], "DOC-1") + + def test_page_vanished(self): + e = self.eva + e.add_doc(1, "DOC-1", "Корень", project="P") + e.add_doc(2, "DOC-2", "Б", parent="DOC-1", project="P") + real = self.client.get_doc + self.client.get_doc = lambda code, fields: None if code == "DOC-2" else real(code, fields) + out = self.sync() + self.assertEqual(out["status"], "failed") + self.assertIn("DOC-2", out["error"]) + + def test_counts(self): + e = self.eva + e.add_doc(1, "DOC-1", "Корень", project="P") + e.add_doc(2, "DOC-2", "Б", parent="DOC-1", project="P") + e.fail_next(429, 2, retry_after=0) + out = self.sync() + self.assertEqual(out["status"], "done", out) + self.assertEqual(out["jira_calls"], len(e.requests)) + self.assertEqual(out["jira_calls"], self.client.calls) + self.assertEqual(out["rate_limited"], 2) + + +class WikiSyncRecoverTest(unittest.TestCase): + def test_recover_resyncs(self): + eva, hub = FakeEva(), FakeHub(SETTINGS) + tmp = tempfile.TemporaryDirectory() + for x in (eva.stop, hub.stop, tmp.cleanup): + self.addCleanup(x) + eva.add_doc(1, "DOC-1", "Корень", project="P") + for i in range(2, 61): + eva.add_doc(i, "DOC-%d" % i, "С", parent="DOC-1", project="P", orderno=i) + cid = hub.enqueue("wiki_sync", {"label": "tz", "root": "DOC-1"}) + hub.queue.remove(cid) + hub.commands[cid]["status"] = "claimed" + hub.in_flight = cid + journal = agent.Journal(os.path.join(tmp.name, "journal.jsonl")) + journal.claimed(cid) + journal.started(cid) # перезапуск посреди выполнения: finished нет + cfg = eva_cfg(eva.base_url, hub_url=hub.base_url, hub_token=hub.token) + runner = agent.Runner(cfg, agent.HubClient(cfg), agent.EvaClient(cfg), journal) + runner.sleep = lambda s: None + runner.eva.sleep = no_sleep + runner.idle_limit = 1 + runner.loop() + self.assertEqual(hub.results[cid]["status"], "done", hub.results[cid]) + self.assertEqual(len(hub.results[cid]["result"]["codes"]), 60) + self.assertEqual([len(b["pages"]) for b in hub.pages], [50, 10]) + self.assertEqual(len([r for r in journal.records() if r["kind"] == "finished"]), 1) + self.assertEqual(journal.unfinished_started(), []) + + +if __name__ == "__main__": + unittest.main()