fix(mirror): флаг neighbor не сбрасывается точечным refresh и ставится полным прогоном (I-2); комментарий про соединение

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o

src/jira_hub/api.py

@@ -111,7 +111,8 @@ def h_agent_result(hub, conn, m, q, body):
 
 def h_agent_state(hub, conn, m, q, body):
     with db.transaction(conn):
-        out = mirror.apply_state(conn, body.get("issues") or [], body.get("links") or [], body.get("neighbor_keys") or [])
+        out = mirror.apply_state(conn, body.get("issues") or [], body.get("links") or [], body.get("neighbor_keys") or [],
+                                 authoritative=queue.is_full_refresh(conn, body.get("command_id")))
         if body.get("command_id"):
             queue.record_state(conn, body["command_id"], out["issues"], out["links"])
     return 200, out

src/jira_hub/mirror.py

@@ -12,9 +12,12 @@ _FILTERS = {"sprint": "i.sprint = ?", "assignee": "i.assignee = ?", "status": "i
             "work_kind": "n.work_kind = ?", "fact_status": "n.fact_status = ?"}
 
 
-def apply_state(conn, issues, links, neighbor_keys=(), fetched_at=None) -> dict:
+def apply_state(conn, issues, links, neighbor_keys=(), fetched_at=None, authoritative=True) -> dict:
     """Upsert задач; связи с source_key среди присланных задач заменяются присланными.
-    Связи, у которых источник не прислан, не трогаются до загрузки их источника."""
+    Связи, у которых источник не прислан, не трогаются до загрузки их источника.
+    authoritative=True — загрузка по базовой JQL: флаг neighbor пересчитывается по neighbor_keys (задача, вошедшая
+    в базовую выборку, перестаёт быть соседней). authoritative=False — точечное обновление (по ключам, по
+    произвольной JQL): флаг только поднимается, никогда не снижается."""
     fetched_at = fetched_at or db.now()
     neighbors = set(neighbor_keys)
     keys = []
@@ -25,9 +28,11 @@ def apply_state(conn, issues, links, neighbor_keys=(), fetched_at=None) -> dict:
             row["fetched_at"] = fetched_at
             row["neighbor"] = 1 if it["key"] in neighbors else 0
             cols = list(row)
+            sets = [f"{c} = excluded.{c}" for c in cols if c not in ("key", "neighbor")]
+            sets.append("neighbor = excluded.neighbor" if authoritative else "neighbor = MAX(issue.neighbor, excluded.neighbor)")
             conn.execute(
                 f"INSERT INTO issue({', '.join(cols)}) VALUES ({', '.join('?' * len(cols))}) "
-                "ON CONFLICT(key) DO UPDATE SET " + ", ".join(f"{c} = excluded.{c}" for c in cols if c != "key"),
+                "ON CONFLICT(key) DO UPDATE SET " + ", ".join(sets),
                 [row[c] for c in cols])
             keys.append(it["key"])
         conn.executemany("DELETE FROM issue_link WHERE source_key = ?", [(k,) for k in keys])
@@ -95,7 +100,7 @@ def set_note(conn, key, fields: dict, updated_by: str) -> dict:
     unknown = set(fields) - set(NOTE_FIELDS)
     if unknown:
         raise ValueError(f"неизвестные поля заметки: {', '.join(sorted(unknown))}")
-    with db.transaction(conn):   # чтение, слияние и запись — атомарно (соединение общее для потоков)
+    with db.transaction(conn):   # чтение, слияние и запись — атомарно (каждый запрос и поток открывает своё соединение, R-2)
         cur = get_note(conn, key) or {"key": key}
         cur.update(fields)
         cur["updated_at"], cur["updated_by"] = db.now(), updated_by

src/jira_hub/queue.py

@@ -174,6 +174,15 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non
 
 # --- отчёты -----------------------------------------------------------------
 
+def is_full_refresh(conn, command_id) -> bool:
+    """Команда — refresh по базовой JQL (полный прогон): только он вправе сбрасывать флаг neighbor."""
+    if not command_id:
+        return False
+    cmd = get_command(conn, command_id)
+    return bool(cmd and cmd["type"] == "refresh" and cmd["params"].get("jql")
+                and cmd["params"]["jql"] == settings.get(conn, "base_jql"))
+
+
 def record_state(conn, command_id, issues_count, links_count) -> None:
     conn.execute("UPDATE sync_run SET issues_count = issues_count + ?, links_count = links_count + ? WHERE command_id = ?",
                  (int(issues_count), int(links_count), command_id))
@@ -198,6 +207,11 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls=
         if cmd["type"] == "refresh":
             conn.execute("UPDATE sync_run SET finished_at = ?, jira_calls = ?, rate_limited = ? WHERE command_id = ?",
                          (now, int(jira_calls or 0), int(rate_limited or 0), command_id))
+            if status == "done" and is_full_refresh(conn, command_id):
+                # полный прогон не обновил задачу — в базовую JQL она больше не входит: сосед
+                run = conn.execute("SELECT started_at FROM sync_run WHERE command_id = ?", (command_id,)).fetchone()
+                if run:
+                    conn.execute("UPDATE issue SET neighbor = 1 WHERE fetched_at < ?", (run["started_at"],))
     if cmd["type"] == "refresh" and status == "done":
         reconcile(conn)
     return get_command(conn, command_id)

tests/test_api.py

@@ -241,3 +241,36 @@ def test_settings_put_invalid_is_4xx_and_atomic(hub):
         c.put("/settings", {"pacing": {"refresh": 3}, "tokens": {}})
     assert 400 <= e.value.status < 500
     assert c.get("/settings")["settings_version"] == v0
+
+
+def _run_refresh(hub, body, issues, neighbor_keys=()):
+    """Поставить refresh, взять его агентом, загрузить состояние, отчитаться done."""
+    hub["client"].post("/refresh", body)
+    cmd = hub["agent"].post("/agent/claim", {})["command"]
+    assert cmd["type"] == "refresh"
+    hub["agent"].post("/agent/state", {"command_id": cmd["id"], "issues": issues, "links": [],
+                                       "neighbor_keys": list(neighbor_keys)})
+    hub["agent"].post(f"/agent/commands/{cmd['id']}/result", {"status": "done"})
+
+
+def test_refresh_by_keys_keeps_neighbor_flag(hub):
+    """I-2(b): точечное обновление соседней задачи не опускает флаг."""
+    _run_refresh(hub, {"full": True}, [{"key": "A-1", "summary": "т", "labels": []},
+                                       {"key": "B-1", "summary": "сосед", "labels": []}], neighbor_keys=["B-1"])
+    assert hub["client"].get("/issues/B-1")["neighbor"] is True
+    _run_refresh(hub, {"keys": ["B-1"]}, [{"key": "B-1", "summary": "сосед 2", "labels": []}])
+    issue = hub["client"].get("/issues/B-1")
+    assert issue["summary"] == "сосед 2" and issue["neighbor"] is True
+
+
+def test_full_refresh_marks_unseen_issue_as_neighbor(hub):
+    """I-2(a): полный прогон по базовой JQL: не обновлённая им задача получает neighbor=1."""
+    conn = hub["hub"].conn()
+    conn.execute("INSERT INTO issue(key, summary, labels, fetched_at) VALUES ('X-1', 'вышла из выборки', '[]', '2020-01-01T00:00:00+00:00')")
+    conn.close()
+    _run_refresh(hub, {"full": True}, [{"key": "A-1", "summary": "т", "labels": []}])
+    assert hub["client"].get("/issues/X-1")["neighbor"] is True
+    assert hub["client"].get("/issues/A-1")["neighbor"] is False
+    # частичный refresh (по ключам) других задач ничего не помечает
+    _run_refresh(hub, {"keys": ["A-1"]}, [{"key": "A-1", "summary": "т2", "labels": []}])
+    assert hub["client"].get("/issues/A-1")["neighbor"] is False

tests/test_mirror.py

@@ -44,6 +44,15 @@ def test_apply_state_marks_neighbors():
     assert mirror.known_keys(conn) == ["A-1", "B-9"]
 
 
+def test_partial_load_never_lowers_neighbor():
+    conn = db.connect(":memory:")
+    mirror.apply_state(conn, [issue("B-9")], [], neighbor_keys={"B-9"})
+    mirror.apply_state(conn, [issue("B-9", summary="новая")], [], authoritative=False)   # refresh {keys}
+    assert mirror.get_issue(conn, "B-9")["neighbor"] is True
+    mirror.apply_state(conn, [issue("B-9")], [], authoritative=True)                    # полный прогон: вошла в базовую
+    assert mirror.get_issue(conn, "B-9")["neighbor"] is False
+
+
 def test_note_survives_refresh_and_renames():
     conn = db.connect(":memory:")
     mirror.set_note(conn, "new:7", {"fact_status": "В работе", "work_kind": "разработка"}, "import")