From 8e82b31d480397cef2ef567d9ffaecd20f9921b7 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 14 Oct 2026 13:11:03 +0400 Subject: [PATCH] =?UTF-8?q?fix:=20=D0=BF=D1=80=D0=B0=D0=B2=D0=BA=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=BE=20=D1=84=D0=B8=D0=BD=D0=B0=D0=BB=D1=8C=D0=BD=D0=BE?= =?UTF-8?q?=D0=BC=D1=83=20=D1=80=D0=B5=D0=B2=D1=8C=D1=8E=20bridge-hygiene?= =?UTF-8?q?=20=E2=80=94=20=D0=BF=D1=80=D0=B0=D0=B2=D0=B0=20backups/,=20cwd?= =?UTF-8?q?,=20=D1=87=D0=B8=D1=81=D1=82=D0=BA=D0=B0=20work/=20=D0=BF=D0=BE?= =?UTF-8?q?=20=D0=BF=D0=B0=D0=BF=D0=BA=D0=B0=D0=BC,=20=D0=BE=D1=82=D0=BC?= =?UTF-8?q?=D0=B5=D0=BD=D0=B0=20=D0=B7=D0=B0=D0=B2=D0=B5=D1=80=D1=88=D1=91?= =?UTF-8?q?=D0=BD=D0=BD=D0=BE=D0=B3=D0=BE=20=D0=BF=D0=B0=D0=BA=D0=B5=D1=82?= =?UTF-8?q?=D0=B0=20(#13)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- ARCHITECTURE.md | 2 +- .../bridge-hygiene/specs/hub-api/spec.md | 2 +- src/jira_hub/api.py | 2 + src/jira_hub/db.py | 5 ++- src/jira_hub/queue.py | 6 ++- src/jira_hub/relay.py | 6 +-- tests/test_api.py | 10 +++++ tests/test_db.py | 16 ++++++++ tests/test_mirror.py | 12 ++++++ tests/test_queue.py | 28 ++++++++++++++ tests/test_relay.py | 37 ++++++++++++++++++- 11 files changed, 116 insertions(+), 10 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 0979031..dbd6379 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -20,7 +20,7 @@ | `importer` | разовый импорт CSV-пакета в заметки, сверка с зеркалом | `mirror`, `settings` | | `cli` | `init`, `serve`, `import`, `mcp`, `tokens` | `api`, `mcp_server`, `importer`, `settings`, `db`, `paths` | -Цепочки: `api → queue → commands/mirror/settings → db`; `api → relay → settings`; `mcp_server → client → HTTP API`; `importer → mirror/settings`. +Цепочки: `api → queue → commands/mirror/settings → db`; `db → backup → paths`, `db → paths`; `api → relay → settings`; `mcp_server → client → HTTP API`; `importer → mirror/settings`. ## Потоки данных - Запись в Jira: клиент (`mcp_server`/интерфейс) → `POST /batches` → `queue` → агент `claim` → Jira → `POST /agent/commands/{id}/result`. diff --git a/openspec/changes/bridge-hygiene/specs/hub-api/spec.md b/openspec/changes/bridge-hygiene/specs/hub-api/spec.md index 8ffb82c..0e420f4 100644 --- a/openspec/changes/bridge-hygiene/specs/hub-api/spec.md +++ b/openspec/changes/bridge-hygiene/specs/hub-api/spec.md @@ -167,4 +167,4 @@ #### Scenario: Повторная подготовка не трогает взятые запросы - **WHEN** реле работает, в `work/` лежит взятый запрос, и пропал подкаталог `resp/` -- **THEN** реле создаёт `resp/` заново; файл в `work/` не удалён, в логе нет «перезапуск реле» +- **THEN** реле создаёт `resp/` заново; файл в `work/` не удалён, в логе нет «удалён незавершённый запрос» diff --git a/src/jira_hub/api.py b/src/jira_hub/api.py index 163aad5..aa087d8 100644 --- a/src/jira_hub/api.py +++ b/src/jira_hub/api.py @@ -14,6 +14,7 @@ from urllib.parse import parse_qs, urlparse from . import backup, db, mirror, queue, relay, settings from .commands import CommandError from .paths import backups_dir as default_backups_dir +from .paths import secure_dir from .queue import ClaimedError log = logging.getLogger(__name__) @@ -33,6 +34,7 @@ class Hub: self.db_path = Path(db_path) self.backups_dir = Path(backups_dir) if backups_dir else default_backups_dir() self.claim_wait = 30 + secure_dir(self.backups_dir) # каталог копий закрыт при каждом старте, а не только после первой копии conn = self.conn() settings.ensure_defaults(conn) conn.close() diff --git a/src/jira_hub/db.py b/src/jira_hub/db.py index f345824..58f3b71 100644 --- a/src/jira_hub/db.py +++ b/src/jira_hub/db.py @@ -117,7 +117,7 @@ def connect(path) -> sqlite3.Connection: """Открыть базу (создать при отсутствии), применить схему. Соединение в автокоммите; для атомарных операций — transaction().""" is_memory = str(path) == ":memory:" - if not is_memory: + if not is_memory and Path(path).parent != Path("."): # путь без каталога — рабочий каталог процесса не трогаем paths.secure_dir(Path(path).parent) conn = sqlite3.connect(str(path), timeout=10, isolation_level=None, check_same_thread=False) conn.row_factory = sqlite3.Row @@ -137,7 +137,8 @@ def migrate(conn: sqlite3.Connection, path=None) -> None: """Создать недостающие таблицы и поднять схему существующей базы до SCHEMA_VERSION. Копия базы (<каталог базы>/backups, без ротации) и миграция — под одной блокировкой записи (BEGIN IMMEDIATE): параллельный connect дождётся её, перечитает версию и не станет ни копировать, ни мигрировать повторно (M-2 #12). Копию делает - отдельное соединение: VACUUM INTO внутри транзакции SQLite запрещает, а читатель при WAL видит базу до миграции.""" + отдельное соединение: VACUUM INTO внутри транзакции SQLite запрещает, а читатель при WAL видит базу до миграции. + Вызывается только вне транзакции.""" conn.executescript(SCHEMA) version = _schema_version(conn) if version is None: diff --git a/src/jira_hub/queue.py b/src/jira_hub/queue.py index 135ac59..2d46c71 100644 --- a/src/jira_hub/queue.py +++ b/src/jira_hub/queue.py @@ -237,7 +237,7 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls= log.warning("команда %s уже %s, но агент создал %s — ключ записан в результат", command_id, cmd["status"], result["key"]) return get_command(conn, command_id) - return cmd # идемпотентно; cancelled не затирается + return cmd # идемпотентно; терминальный статус (cancelled/failed) не затирается now = db.now() conn.execute("UPDATE command SET status = ?, finished_at = ?, result = ?, error = ?, jira_calls = ?, rate_limited = ? WHERE id = ?", (status, now, _dump(result) if result is not None else None, error, int(jira_calls or 0), int(rate_limited or 0), command_id)) @@ -285,7 +285,9 @@ def cancel_batch(conn, batch_id) -> dict: if conn.execute("SELECT 1 FROM batch WHERE id = ?", (batch_id,)).fetchone() is None: raise KeyError(batch_id) now = db.now() - conn.execute("UPDATE batch SET cancelled_at = ? WHERE id = ? AND cancelled_at IS NULL", (now, batch_id)) + # завершённый пакет (нет queued/claimed) отменённым не становится: отменять нечего + conn.execute("UPDATE batch SET cancelled_at = ? WHERE id = ? AND cancelled_at IS NULL AND EXISTS " + "(SELECT 1 FROM command WHERE batch_id = ? AND status IN ('queued', 'claimed'))", (now, batch_id, batch_id)) conn.execute("UPDATE command SET status = 'cancelled', finished_at = ? WHERE batch_id = ? AND status = 'queued'", (now, batch_id)) view = batch_view(conn, batch_id) # взятые агентом команды отмена пакета не останавливает: они будут выполнены (M-11 #12) diff --git a/src/jira_hub/relay.py b/src/jira_hub/relay.py index 0c36f58..68ffb61 100644 --- a/src/jira_hub/relay.py +++ b/src/jira_hub/relay.py @@ -61,7 +61,7 @@ class Relay: self._cleanup_at = 0.0 self._ready = False self._warned: str | None = None - self._cleaned: Path | None = None # папка, чей work/ вычищен после старта реле + self._cleaned: set[Path] = set() # папки, чей work/ вычищен после старта реле self.workers: list[threading.Thread] = [] # --- цикл --------------------------------------------------------------------- @@ -124,12 +124,12 @@ class Relay: def _prepare(self) -> None: for sub in ("req", "resp", "work"): (self.root / sub).mkdir(exist_ok=True) - if self._cleaned != self.root: + if self.root not in self._cleaned: # первая подготовка этой папки после старта: взятые прошлым процессом запросы уже никто не обработает for stale in (self.root / "work").iterdir(): stale.unlink(missing_ok=True) say(f"relay: удалён незавершённый запрос {stale.name} (старт реле)") - self._cleaned = self.root + self._cleaned.add(self.root) else: # пропал подкаталог или не было доступа: взятые запросы ещё обрабатываются своими потоками — work/ не трогаем say(f"relay: папка {self.root} подготовлена заново") diff --git a/tests/test_api.py b/tests/test_api.py index d48517a..6710d5a 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -75,6 +75,16 @@ def test_agent_flow(hub): assert c.get("/agent")["online"] is True +def test_cancel_batch_api_reports_still_claimed(hub): + c, a = hub["client"], hub["agent"] + b = c.post("/batches", {"title": "x", "author": "claude", "commands": [ + {"type": "ensure_fields", "params": {"key": "A-1", "duedate": "2026-10-23"}}, + {"type": "ensure_fields", "params": {"key": "A-2", "duedate": "2026-10-23"}}]}) + cmd = a.post("/agent/claim", {})["command"] + out = c.post("/batches/%d/cancel" % b["id"], {}) + assert out["still_claimed"] == [{"id": cmd["id"], "type": "ensure_fields"}] + + def test_interface_routes(hub): c = hub["client"] with pytest.raises(HubError) as e: diff --git a/tests/test_db.py b/tests/test_db.py index fb5c3d6..8fd7025 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -128,6 +128,22 @@ def test_data_dir_is_private(tmp_path): assert _mode(home) == 0o700 +def test_existing_backups_dir_made_private_on_start(tmp_path): + from jira_hub import api + backups = tmp_path / "backups" + backups.mkdir() + backups.chmod(0o755) + api.Hub(tmp_path / "hub.sqlite", backups_dir=backups) + assert _mode(backups) == 0o700 + + +def test_connect_without_directory_does_not_chmod_cwd(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + tmp_path.chmod(0o755) + db.connect("hub.sqlite").close() + assert _mode(tmp_path) == 0o755 + + def test_backups_dir_is_private(tmp_path): conn = db.connect(tmp_path / "hub.sqlite") backup.make_backup(conn, tmp_path / "backups", "batch-1") diff --git a/tests/test_mirror.py b/tests/test_mirror.py index 894f1fe..ac73780 100644 --- a/tests/test_mirror.py +++ b/tests/test_mirror.py @@ -116,6 +116,18 @@ def test_rename_note_merges_with_existing(): assert mirror.get_note(conn, "new:5") is None +def test_rename_note_keeps_newer_real_key_stamp(): + conn = db.connect(":memory:") + mirror.set_note(conn, "new:6", {"fact_status": "В работе"}, "import") + mirror.set_note(conn, "A-6", {"owner_note": "позже"}, "human") + conn.execute("UPDATE issue_note SET updated_at = '2020-01-01T00:00:00+00:00' WHERE key = 'new:6'") + conn.execute("UPDATE issue_note SET updated_at = '2030-01-01T00:00:00+00:00' WHERE key = 'A-6'") + mirror.rename_note(conn, "new:6", "A-6") + note = mirror.get_note(conn, "A-6") + assert note["updated_by"] == "human" and note["updated_at"] == "2030-01-01T00:00:00+00:00" + assert note["fact_status"] == "В работе" and note["owner_note"] == "позже" + + def test_rename_note_empty_string_does_not_overwrite(): conn = db.connect(":memory:") mirror.set_note(conn, "A-6", {"rest_text": "доделать отчёт"}, "human") diff --git a/tests/test_queue.py b/tests/test_queue.py index bd91ca6..066f7ca 100644 --- a/tests/test_queue.py +++ b/tests/test_queue.py @@ -435,6 +435,34 @@ def test_empty_full_refresh_keeps_neighbors(): assert conn.execute("SELECT neighbor FROM issue WHERE key = 'A-1'").fetchone()["neighbor"] == 0 +def test_cancel_finished_batch_stays_done(): + conn = fresh() + b = queue.create_batch(conn, "x", "claude", [fields("A-1")]) + while (c := queue.claim(conn)): + queue.report_result(conn, c["id"], "done", result={"changed": []}) + view = queue.cancel_batch(conn, b["id"]) + assert view["status"] == "done" + assert conn.execute("SELECT cancelled_at FROM batch WHERE id = ?", (b["id"],)).fetchone()["cancelled_at"] is None + + +def test_redelivered_full_refresh_marks_neighbors_once(): + conn = fresh() + jql = settings.get(conn, "base_jql") + mirror.apply_state(conn, [{"key": "OLD", "summary": "x", "labels": []}], [], fetched_at="2020-01-01T00:00:00+00:00") + queue.create_batch(conn, "полный", "claude", [{"type": "refresh", "params": {"jql": jql}}]) + r = queue.claim(conn) + queue.record_state(conn, r["id"], 1, 0) + mirror.apply_state(conn, [{"key": "NEW", "summary": "y", "labels": []}], []) + assert queue.claim(conn)["id"] == r["id"] # отчёт потерян — довыдача + assert conn.execute("SELECT deliveries FROM command WHERE id = ?", (r["id"],)).fetchone()["deliveries"] == 2 + queue.record_state(conn, r["id"], 1, 0) + mirror.apply_state(conn, [{"key": "NEW", "summary": "y", "labels": []}], []) + queue.report_result(conn, r["id"], "done") + nb = {row["key"]: row["neighbor"] for row in conn.execute("SELECT key, neighbor FROM issue")} + assert nb == {"OLD": 1, "NEW": 0} + assert conn.execute("SELECT issues_count FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["issues_count"] == 1 + + def test_redelivered_refresh_resets_counts(): conn = fresh() queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}]) diff --git a/tests/test_relay.py b/tests/test_relay.py index f35506a..088e806 100644 --- a/tests/test_relay.py +++ b/tests/test_relay.py @@ -1,6 +1,7 @@ import http.client import json import os +import shutil import threading import time from datetime import datetime, timedelta, timezone @@ -349,7 +350,41 @@ def test_reprepare_keeps_work(env, capsys): r.step() assert (root / "req").is_dir() and (root / "work" / "busy.json").exists() out = capsys.readouterr().out - assert "перезапуск реле" not in out and "подготовлена заново" in out + assert "удалён незавершённый запрос" not in out and "подготовлена заново" in out + + +def test_vanished_tree_keeps_work(env, capsys): + r, root = env["relay"], env["root"] + r.step() + (root / "work" / "busy.json").write_text("{}", encoding="utf-8") + capsys.readouterr() + shutil.rmtree(root) # папка пропала целиком и появилась снова + root.mkdir() + (root / "work").mkdir() + (root / "work" / "busy.json").write_text("{}", encoding="utf-8") # взятый запрос, который ещё обрабатывается + for _ in range(3): + r.step() + assert (root / "work" / "busy.json").exists() and (root / "req").is_dir() + assert "удалён незавершённый запрос" not in capsys.readouterr().out + + +def test_each_dir_cleaned_once(env, tmp_path, capsys): + r, root, h = env["relay"], env["root"], env["hub"] + other = tmp_path / "other" + (other / "work").mkdir(parents=True) + (other / "work" / "old.json").write_text("{}", encoding="utf-8") + r.step() # root вычищен + conn = h.conn() + settings.set_(conn, "relay_dir", str(other)) + r._settings_at = None + r.step() + assert not (other / "work" / "old.json").exists() # новая папка чистится + settings.set_(conn, "relay_dir", str(root)) + r._settings_at = None + (root / "work" / "busy.json").write_text("{}", encoding="utf-8") + r.step() + conn.close() + assert (root / "work" / "busy.json").exists() # возврат A->B->A: A повторно не чистится def test_relay_targets_serve_host(tmp_path, monkeypatch):