feat(relay): реле через общую папку — запросы агента из файлов в свой HTTP и обратно (#5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/jira_hub/relay.py

@@ -0,0 +1,223 @@
+"""Реле через общую папку (specs/hub-api, «Реле через общую папку»): файлы-запросы агента из <relay_dir>/req/
+пересылаются в HTTP этого же хаба, ответы кладутся в <relay_dir>/resp/. HTTP API, обработчики и токены не меняются;
+реле для них — ещё один клиент. Нужно, когда VPN на виртуалке закрывает сеть до хаба (design.md folder-transport)."""
+import json
+import os
+import threading
+import time
+import traceback
+import urllib.error
+import urllib.request
+from datetime import datetime, timezone
+from pathlib import Path
+
+from . import settings
+
+CLAIM_PATH = "/agent/claim"
+CLAIM_MIN_LEFT = 35       # с: claim пересылается, только если до deadline осталось не меньше (хаб держит claim ≤ 30 с)
+SETTINGS_EVERY = 5        # с: как часто перечитывать relay_dir
+CLEANUP_EVERY = 60        # с: как часто убирать мусор
+RESP_TTL = 600            # с: ответ, за которым агент не пришёл
+REQ_GRACE = 60            # с после deadline: запрос, который никто не взял
+TMP_TTL = 600             # с: брошенные временные файлы (имя на точку)
+BAD_JSON_WAIT = 2         # с: сколько перечитывать нечитаемый запрос
+FORWARD_TIMEOUT = 90      # с: ожидание ответа своего HTTP (claim держит до 30 с)
+METHODS = ("GET", "POST", "PUT")
+
+
+def say(msg: str) -> None:
+    print(f"[{time.strftime('%d/%b/%Y %H:%M:%S')}] {msg}", flush=True)
+
+
+def parse_deadline(value) -> float | None:
+    """ISO 8601 → секунды эпохи; без зоны считается UTC; мусор → None."""
+    if not isinstance(value, str):
+        return None
+    try:
+        dt = datetime.fromisoformat(value)
+    except ValueError:
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.timestamp()
+
+
+def write_atomic(directory: Path, name: str, payload: dict) -> None:
+    """Записать JSON под временным именем на точку и переименовать: читатель видит файл только целиком."""
+    tmp = Path(directory) / f".{name}.tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(payload, f, ensure_ascii=False)
+        f.flush()
+        os.fsync(f.fileno())
+    os.replace(tmp, Path(directory) / name)
+
+
+class Relay:
+    def __init__(self, hub, port: int, host: str = "127.0.0.1", poll: float = 0.1):
+        self.hub, self.port, self.host, self.poll = hub, port, host, poll
+        self.root: Path | None = None
+        self._settings_at: float | None = None
+        self._cleanup_at = 0.0
+        self._ready = False
+        self._warned = False
+        self.workers: list[threading.Thread] = []
+
+    # --- цикл ---------------------------------------------------------------------
+    def run(self, stop: threading.Event) -> None:
+        while not stop.is_set():
+            try:
+                self.step()
+            except Exception:  # noqa: BLE001 — сбой одного шага не останавливает реле
+                traceback.print_exc()
+            stop.wait(self.poll)
+
+    def step(self) -> None:
+        now = time.time()
+        if self._settings_at is None or now - self._settings_at >= SETTINGS_EVERY:
+            self._settings_at = now
+            self._reload()
+        if self.root is None:
+            return
+        if not self.root.is_dir():
+            if not self._warned:
+                say(f"relay: папки {self.root} нет — жду")
+                self._warned = True
+            self._ready = False
+            return
+        if not self._ready:
+            self._prepare()
+        for path in sorted((self.root / "req").iterdir()):
+            if path.name.startswith(".") or not path.name.endswith(".json"):
+                continue
+            self._take(path)
+        if now - self._cleanup_at >= CLEANUP_EVERY:
+            self._cleanup_at = now
+            self.cleanup(now)
+
+    def join(self, timeout: float = 5) -> None:
+        for w in list(self.workers):
+            w.join(timeout)
+
+    # --- папка ----------------------------------------------------------------------
+    def _reload(self) -> None:
+        conn = self.hub.conn()
+        try:
+            value = settings.get(conn, "relay_dir")
+        finally:
+            conn.close()
+        root = Path(value) if value else None
+        if root != self.root:
+            self.root, self._ready, self._warned = root, False, False
+            say(f"relay: папка {root}" if root else "relay: выключено")
+
+    def _prepare(self) -> None:
+        for sub in ("req", "resp", "work"):
+            (self.root / sub).mkdir(exist_ok=True)
+        for stale in (self.root / "work").iterdir():
+            stale.unlink(missing_ok=True)
+            say(f"relay: удалён незавершённый запрос {stale.name} (перезапуск реле)")
+        self._ready, self._warned = True, False
+
+    def _take(self, path: Path) -> None:
+        target = self.root / "work" / path.name
+        try:
+            os.rename(path, target)
+        except FileNotFoundError:          # агент успел удалить свой просроченный запрос
+            return
+        worker = threading.Thread(target=self.process, args=(self.root, target), daemon=True)
+        worker.start()
+        self.workers = [w for w in self.workers if w.is_alive()] + [worker]
+
+    # --- обработка запроса -------------------------------------------------------------
+    def process(self, root: Path, work: Path) -> None:
+        started = time.monotonic()
+        try:
+            req = self._read(work)
+            if req is None:
+                say(f"relay: нечитаемый запрос {work.name} — выброшен")
+                return
+            method, path = req.get("method"), req.get("path")
+            deadline = parse_deadline(req.get("deadline"))
+            left = deadline - time.time() if deadline is not None else -1.0
+            if left <= 0 or (method == "POST" and path == CLAIM_PATH and left < CLAIM_MIN_LEFT):
+                say(f"relay {method} {path} просрочен (осталось {left:.0f} с) — выброшен")
+                return
+            status, body = self.forward(method, path, req.get("body"), req.get("authorization"))
+            write_atomic(root / "resp", work.name, {"id": req.get("id"), "status": status, "body": body})
+            say(f"relay {method} {path} {status} {time.monotonic() - started:.1f} с")
+        except OSError as e:               # папка пропала на ходу — агент дождётся своего срока
+            say(f"relay: сбой файла {work.name}: {e}")
+        finally:
+            work.unlink(missing_ok=True)
+
+    def _read(self, work: Path) -> dict | None:
+        end = time.monotonic() + BAD_JSON_WAIT
+        while True:
+            try:
+                with open(work, encoding="utf-8") as f:
+                    data = json.load(f)
+                return data if isinstance(data, dict) else None
+            except (ValueError, OSError):
+                if time.monotonic() >= end:
+                    return None
+                time.sleep(0.1)
+
+    def forward(self, method, path, body, authorization) -> tuple[int, str]:
+        if method not in METHODS or not isinstance(path, str) or not path.startswith("/"):
+            return 400, _error("bad_request", f"неверный метод или путь: {method} {path}")
+        data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
+        try:
+            req = urllib.request.Request(f"http://{self.host}:{self.port}{path}", data=data, method=method)
+            if authorization:
+                req.add_header("Authorization", str(authorization))
+            if data is not None:
+                req.add_header("Content-Type", "application/json; charset=utf-8")
+            with urllib.request.urlopen(req, timeout=FORWARD_TIMEOUT) as resp:
+                return resp.status, resp.read().decode("utf-8", "replace")
+        except urllib.error.HTTPError as e:
+            try:
+                return e.code, e.read().decode("utf-8", "replace")
+            finally:
+                e.close()
+        except ValueError as e:            # недопустимый заголовок (перевод строки) или URL
+            return 400, _error("bad_request", str(e))
+        except (urllib.error.URLError, OSError) as e:
+            return 502, _error("relay", f"хаб недоступен изнутри: {e}")
+
+    # --- уборка ------------------------------------------------------------------------
+    def cleanup(self, now: float) -> None:
+        if self.root is None or not self.root.is_dir():
+            return
+        for p in _entries(self.root / "resp"):
+            ttl = TMP_TTL if p.name.startswith(".") else RESP_TTL
+            if _age(p, now) > ttl:
+                p.unlink(missing_ok=True)
+        for p in _entries(self.root / "req"):
+            if p.name.startswith("."):
+                if _age(p, now) > TMP_TTL:
+                    p.unlink(missing_ok=True)
+                continue
+            try:
+                deadline = parse_deadline(json.loads(p.read_text(encoding="utf-8")).get("deadline"))
+            except (OSError, ValueError, AttributeError):
+                deadline = None
+            if deadline is None or now - deadline > REQ_GRACE:
+                p.unlink(missing_ok=True)
+
+
+def _error(error: str, detail: str) -> str:
+    return json.dumps({"error": error, "detail": detail}, ensure_ascii=False)
+
+
+def _entries(directory: Path) -> list[Path]:
+    try:
+        return list(directory.iterdir())
+    except FileNotFoundError:
+        return []
+
+
+def _age(p: Path, now: float) -> float:
+    try:
+        return now - p.stat().st_mtime
+    except FileNotFoundError:
+        return 0.0

tests/test_relay.py

@@ -0,0 +1,243 @@
+import json
+import os
+import threading
+import time
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from jira_hub import api, relay, settings
+from jira_hub.client import HubClient
+
+
+@pytest.fixture
+def env(tmp_path):
+    h = api.Hub(tmp_path / "hub.sqlite", backups_dir=tmp_path / "backups")
+    h.claim_wait = 0
+    conn = h.conn()
+    root = tmp_path / "bridge"
+    root.mkdir()
+    settings.set_(conn, "relay_dir", str(root))
+    tok = settings.tokens(conn)
+    conn.close()
+    srv = api.make_server(h, "127.0.0.1", 0)
+    th = threading.Thread(target=srv.serve_forever, daemon=True)
+    th.start()
+    r = relay.Relay(h, srv.server_port)
+    client = HubClient(f"http://127.0.0.1:{srv.server_port}", tok["client"])
+    yield {"hub": h, "root": root, "relay": r, "agent": "Bearer " + tok["agent"], "client": client,
+           "port": srv.server_port}
+    srv.shutdown()
+    srv.server_close()
+    th.join(5)
+
+
+def deadline(seconds):
+    return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat()
+
+
+def put_request(root, rid, method, path, auth, body=None, seconds=60, **extra):
+    req = {"id": rid, "method": method, "path": path, "body": body, "authorization": auth,
+           "deadline": deadline(seconds), **extra}
+    relay.write_atomic(root / "req", f"{rid}.json", req)
+
+
+def response(root, rid):
+    p = root / "resp" / f"{rid}.json"
+    return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None
+
+
+def run_once(r):
+    r.step()
+    r.join()
+
+
+def test_request_roundtrip(env):
+    r, root = env["relay"], env["root"]
+    r.step()                                            # подготовка: req/, resp/, work/
+    put_request(root, "a1", "GET", "/agent/settings", env["agent"])
+    run_once(r)
+    resp = response(root, "a1")
+    assert resp["status"] == 200 and resp["id"] == "a1"
+    assert json.loads(resp["body"])["project_key"] == "AURORA"
+    assert not list((root / "work").iterdir()) and not list((root / "req").iterdir())
+
+
+def test_status_passes_through(env):
+    r, root = env["relay"], env["root"]
+    r.step()
+    put_request(root, "u1", "GET", "/issues", env["agent"])             # агентский токен на клиентский маршрут
+    put_request(root, "u2", "GET", "/agent/commands/999", env["agent"])
+    run_once(r)
+    assert response(root, "u1")["status"] == 401
+    assert response(root, "u2")["status"] == 404
+
+
+def test_expired_and_short_claim_dropped(env):
+    r, root, c = env["relay"], env["root"], env["client"]
+    b = c.post("/batches", {"title": "t", "author": "test",
+                            "commands": [{"type": "ensure_fields", "params": {"key": "A-1", "duedate": "2026-10-23"}}]})
+    r.step()
+    put_request(root, "c1", "POST", "/agent/claim", env["agent"], body={}, seconds=20)
+    put_request(root, "s1", "GET", "/agent/settings", env["agent"], seconds=-1)
+    run_once(r)
+    assert response(root, "c1") is None and response(root, "s1") is None
+    statuses = [x["status"] for x in c.get(f"/batches/{b['id']}")["commands"]]
+    assert statuses[0] == "queued"
+    assert not list((root / "work").iterdir())
+
+
+def test_claim_with_enough_time_is_forwarded(env):
+    r, root = env["relay"], env["root"]
+    r.step()
+    put_request(root, "c2", "POST", "/agent/claim", env["agent"], body={"agent_version": 2}, seconds=50)
+    run_once(r)
+    assert response(root, "c2")["status"] == 200
+
+
+def test_long_claim_does_not_block_other_requests(env):
+    env["hub"].claim_wait = 3
+    r, root = env["relay"], env["root"]
+    r.step()
+    put_request(root, "c3", "POST", "/agent/claim", env["agent"], body={}, seconds=60)
+    r.step()
+    put_request(root, "s3", "GET", "/agent/settings", env["agent"])
+    t0 = time.monotonic()
+    while response(root, "s3") is None and time.monotonic() - t0 < 2.5:
+        r.step()
+        time.sleep(0.05)
+    assert response(root, "s3")["status"] == 200
+    assert response(root, "c3") is None                 # claim ещё ждёт
+    r.join()
+    assert response(root, "c3")["status"] == 200
+
+
+def test_unreachable_hub_gives_502(env):
+    r = relay.Relay(env["hub"], 1)                      # на порту 1 никто не слушает
+    root = env["root"]
+    r.step()
+    put_request(root, "p1", "GET", "/agent/settings", env["agent"])
+    run_once(r)
+    resp = response(root, "p1")
+    assert resp["status"] == 502 and "хаб недоступен" in resp["body"]
+
+
+def test_dotfiles_ignored_and_bad_json_dropped(env, monkeypatch, capsys):
+    monkeypatch.setattr(relay, "BAD_JSON_WAIT", 0.2)
+    r, root = env["relay"], env["root"]
+    r.step()
+    (root / "req" / ".x1.json.tmp").write_text("{}", encoding="utf-8")
+    (root / "req" / "bad.json").write_text("{не json", encoding="utf-8")
+    run_once(r)
+    assert (root / "req" / ".x1.json.tmp").exists()
+    assert response(root, "bad") is None and not list((root / "work").iterdir())
+    assert "нечитаемый запрос bad.json" in capsys.readouterr().out
+
+
+def test_bad_request_fields_get_400(env):
+    r, root = env["relay"], env["root"]
+    r.step()
+    put_request(root, "b1", "GET", "agent/settings", env["agent"])                 # путь без /
+    put_request(root, "b2", "DELETE", "/agent/settings", env["agent"])             # метод не из списка
+    put_request(root, "b3", "GET", "/agent/settings", "Bearer x\r\nX-Evil: 1")     # перевод строки в заголовке
+    run_once(r)
+    for rid in ("b1", "b2", "b3"):
+        assert response(root, rid)["status"] == 400, rid
+
+
+def test_missing_deadline_is_dropped(env):
+    r, root = env["relay"], env["root"]
+    r.step()
+    relay.write_atomic(root / "req", "d1.json", {"id": "d1", "method": "GET", "path": "/agent/settings",
+                                                 "authorization": env["agent"]})
+    relay.write_atomic(root / "req", "d2.json", {"id": "d2", "method": "GET", "path": "/agent/settings",
+                                                 "authorization": env["agent"], "deadline": "завтра"})
+    run_once(r)
+    assert response(root, "d1") is None and response(root, "d2") is None
+
+
+def test_vanished_request_is_skipped(env, capsys):
+    r, root = env["relay"], env["root"]
+    r.step()
+    put_request(root, "v1", "GET", "/agent/settings", env["agent"])
+    orig = os.rename
+
+    def vanish(src, dst):
+        os.remove(src)                                  # агент удалил свой запрос раньше реле
+        return orig(src, dst)
+
+    os.rename = vanish
+    try:
+        run_once(r)
+    finally:
+        os.rename = orig
+    assert response(root, "v1") is None
+    assert "Traceback" not in capsys.readouterr().err
+
+
+def test_work_cleared_on_prepare(env, capsys):
+    r, root = env["relay"], env["root"]
+    (root / "work").mkdir()
+    (root / "work" / "old.json").write_text("{}", encoding="utf-8")
+    r.step()
+    assert not (root / "work" / "old.json").exists()
+    assert "old.json" in capsys.readouterr().out
+
+
+def test_missing_dir_warns_once_then_works(env, capsys):
+    r, root = env["relay"], env["root"]
+    gone = root.parent / "bridge-later"
+    conn = env["hub"].conn()
+    settings.set_(conn, "relay_dir", str(gone))
+    conn.close()
+    for _ in range(5):
+        r.step()
+    out = capsys.readouterr().out
+    assert out.count("нет — жду") == 1
+    gone.mkdir()
+    r.step()
+    assert (gone / "req").is_dir() and (gone / "resp").is_dir() and (gone / "work").is_dir()
+    put_request(gone, "m1", "GET", "/agent/settings", env["agent"])
+    run_once(r)
+    assert response(gone, "m1")["status"] == 200
+
+
+def test_disabled_and_reenabled_without_restart(env, monkeypatch):
+    r, root = env["relay"], env["root"]
+    conn = env["hub"].conn()
+    settings.set_(conn, "relay_dir", None)
+    conn.close()
+    r.step()
+    assert r.root is None and not (root / "req").exists()
+    monkeypatch.setattr(relay, "SETTINGS_EVERY", 0)
+    conn = env["hub"].conn()
+    settings.set_(conn, "relay_dir", str(root))
+    conn.close()
+    r.step()
+    assert r.root == root and (root / "req").is_dir()
+
+
+def test_cleanup_by_age(env):
+    r, root = env["relay"], env["root"]
+    r.step()
+    old, fresh = root / "resp" / "old.json", root / "resp" / "fresh.json"
+    tmp_old = root / "resp" / ".t.json.tmp"
+    for p in (old, fresh, tmp_old):
+        p.write_text("{}", encoding="utf-8")
+    now = time.time()
+    os.utime(old, (now - 700, now - 700))
+    os.utime(tmp_old, (now - 700, now - 700))
+    stale_req = root / "req" / ".q.json.tmp"
+    stale_req.write_text("{}", encoding="utf-8")
+    os.utime(stale_req, (now - 700, now - 700))
+    r.cleanup(now)
+    assert not old.exists() and not tmp_old.exists() and not stale_req.exists()
+    assert fresh.exists()
+
+
+def test_log_line(env, capsys):
+    r, root = env["relay"], env["root"]
+    r.step()
+    put_request(root, "l1", "GET", "/agent/settings", env["agent"])
+    run_once(r)
+    assert "relay GET /agent/settings 200" in capsys.readouterr().out