Отметка журнала: history в refresh, sync_run.history_at (#70)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
f0590ac9ea
commit
e3649a1a38
@ -1,6 +1,10 @@
|
||||
"""Журнал изменений задач (issue_change): приём от агента, логическое имя поля.
|
||||
Пишет таблицу только приём журнала (POST /agent/history); API и MCP её только читают (spec data-model)."""
|
||||
from . import db, settings
|
||||
"""Журнал изменений задач (issue_change): приём от агента, логическое имя поля, отметка «с какого момента» для
|
||||
refresh. Пишет таблицу только приём журнала (POST /agent/history); API и MCP её только читают (spec data-model)."""
|
||||
from datetime import timedelta
|
||||
|
||||
from . import db, projects, settings
|
||||
|
||||
SINCE_MARGIN = timedelta(minutes=15) # расхождение часов Mac mini и Jira и длительность прогона (design #5)
|
||||
|
||||
BUILTIN_FIELDS = ("status", "description", "summary", "assignee", "priority", "duedate", "labels", "resolution")
|
||||
_REQUIRED = ("history_id", "key", "created", "field_name")
|
||||
@ -46,7 +50,7 @@ def logical_field(item: dict, field_ids: dict | None, epic_link: dict | None) ->
|
||||
def record_changes(conn, changes: list[dict]) -> dict:
|
||||
"""Записать проверенную пачку (INSERT OR IGNORE по (history_id, seq)) в одной транзакции."""
|
||||
field_ids, epic_link = settings.get(conn, "field_ids"), settings.get(conn, "epic_link")
|
||||
now, accepted = db.now(), 0
|
||||
now, new = db.now(), 0
|
||||
with db.transaction(conn):
|
||||
for c in changes:
|
||||
cur = conn.execute(
|
||||
@ -55,5 +59,32 @@ def record_changes(conn, changes: list[dict]) -> dict:
|
||||
(c["history_id"], c["seq"], c["key"], db.jira_utc(c["created"]), c["created"], c.get("author"),
|
||||
logical_field(c, field_ids, epic_link), c.get("field_id"), c["field_name"], c.get("from"),
|
||||
c.get("from_string"), c.get("to"), c.get("to_string"), now))
|
||||
accepted += cur.rowcount
|
||||
return {"accepted": accepted, "duplicates": len(changes) - accepted}
|
||||
new += cur.rowcount
|
||||
return {"accepted": new, "duplicates": len(changes) - new}
|
||||
|
||||
|
||||
def backfill_jql(base_jql: str) -> str:
|
||||
"""JQL стартовой загрузки: незакрытые задачи открытых спринтов (design #6)."""
|
||||
return f"({base_jql}) AND sprint in openSprints() AND statusCategory != Done"
|
||||
|
||||
|
||||
def delivery_param(conn, params: dict) -> dict | None:
|
||||
"""Параметр history выдаваемого refresh (spec commands «Журнал изменений в refresh»); None — не дописывать."""
|
||||
base = projects.base_jql(conn)
|
||||
if base is None:
|
||||
return None
|
||||
mark = conn.execute("SELECT started_at, query FROM sync_run WHERE history_at IS NOT NULL "
|
||||
"ORDER BY started_at DESC, id DESC LIMIT 1").fetchone()
|
||||
out = {}
|
||||
if mark is not None:
|
||||
since = db.parse(mark["started_at"]) - SINCE_MARGIN
|
||||
out["since"] = since.strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
if params.get("jql") == base and (mark is None or mark["query"] != base):
|
||||
out["jql"] = backfill_jql(base)
|
||||
return out or None
|
||||
|
||||
|
||||
def accepted(result) -> bool:
|
||||
"""Итог журнала в результате refresh принят: history — словарь без непустого error (backfill_error и incomplete
|
||||
отметку не держат)."""
|
||||
return isinstance(result, dict) and isinstance(result.get("history"), dict) and not result["history"].get("error")
|
||||
|
||||
@ -7,7 +7,7 @@ from datetime import timedelta
|
||||
|
||||
from . import backup as _backup # noqa: F401 (имя модуля занято параметром backup в claim)
|
||||
from . import commands as C
|
||||
from . import db, files, mirror, projects, settings
|
||||
from . import db, files, history, mirror, projects, settings
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@ -264,10 +264,14 @@ def _redeliver(conn, now) -> dict | None:
|
||||
|
||||
def deliver_params(conn, cmd, files_dir) -> dict:
|
||||
"""Параметры команды для агента: ensure_attachment получает size — текущий размер файла в хранилище или None,
|
||||
если файла уже нет (решение 5); refresh — boards, доски проектов (design #7). В базе params остаются без них:
|
||||
повтор команды проходит проверку как новая."""
|
||||
если файла уже нет (решение 5); refresh — boards, доски проектов (design #7), и history — отметка журнала
|
||||
(change-history). В базе params остаются без них: повтор команды проходит проверку как новая."""
|
||||
if cmd["type"] == "refresh":
|
||||
return {**cmd["params"], "boards": projects.boards(conn)}
|
||||
out = {**cmd["params"], "boards": projects.boards(conn)}
|
||||
hist = history.delivery_param(conn, cmd["params"]) # отметка журнала (change-history, design #5)
|
||||
if hist is not None:
|
||||
out["history"] = hist
|
||||
return out
|
||||
if cmd["type"] != "ensure_attachment":
|
||||
return cmd["params"]
|
||||
size = files.size(files_dir, cmd["params"].get("file")) if files_dir is not None else None
|
||||
@ -370,6 +374,8 @@ def report_result(conn, command_id, status, result=None, error=None, jira_calls=
|
||||
(now, int(jira_calls or 0), int(rate_limited or 0), command_id))
|
||||
if isinstance(result, dict) and "boards" in result:
|
||||
projects.record_board_types(conn, result["boards"]) # в любом статусе; без ключа (агент v11) типы не трогаем
|
||||
if status == "done" and is_full_refresh(conn, command_id) and history.accepted(result):
|
||||
conn.execute("UPDATE sync_run SET history_at = ? WHERE command_id = ?", (now, command_id))
|
||||
if status == "done" and is_full_refresh(conn, command_id):
|
||||
# полный прогон не обновил задачу — в базовую JQL она больше не входит: сосед. Пустой прогон (0 задач) —
|
||||
# почти всегда сбой выборки, а не «все задачи ушли»: соседей не трогаем (M-12)
|
||||
|
||||
@ -963,3 +963,13 @@ def test_agent_history_route(hub):
|
||||
assert conn.execute("SELECT COUNT(*) AS n FROM issue_change").fetchone()["n"] == 3
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_sync_runs_history_at(hub):
|
||||
conn = hub["hub"].conn()
|
||||
try:
|
||||
conn.execute("INSERT INTO sync_run(started_at, kind, history_at) VALUES ('2026-10-22T09:00:00+00:00', 'jql', "
|
||||
"'2026-10-22T09:02:00+00:00')")
|
||||
finally:
|
||||
conn.close()
|
||||
assert hub["client"].get("/sync-runs")["runs"][0]["history_at"] == "2026-10-22T09:02:00+00:00"
|
||||
|
||||
@ -941,3 +941,90 @@ def test_refresh_failed_result_records_board_types(conn, add_project): # R-3
|
||||
c = queue.claim(conn)
|
||||
queue.report_result(conn, c["id"], "failed", result={"boards": {"5120": "kanban"}}, error="e")
|
||||
assert projects.get(conn, "STOCKS_DRRK")["board_type"] == "kanban"
|
||||
|
||||
|
||||
# --- журнал изменений: history в выдаче refresh, sync_run.history_at (#70, Task 3) ----------------------------------
|
||||
|
||||
def _full_run(conn, result, started_at=None, status="done"):
|
||||
"""Полный refresh: поставить, выдать, отчитаться; started_at — подменить время начала прогона."""
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": projects.base_jql(conn)}}])
|
||||
c = queue.claim(conn)
|
||||
if started_at:
|
||||
conn.execute("UPDATE sync_run SET started_at = ? WHERE command_id = ?", (started_at, c["id"]))
|
||||
queue.report_result(conn, c["id"], status, result=result)
|
||||
return c
|
||||
|
||||
|
||||
def test_first_full_refresh_gets_backfill(conn, add_project): # сценарий «Первый refresh после выкатки»
|
||||
add_project(conn)
|
||||
b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": projects.base_jql(conn)}}])
|
||||
got = queue.claim(conn)
|
||||
assert got["params"]["history"] == {
|
||||
"jql": '(project = AURORA AND labels in ("AWPCA")) AND sprint in openSprints() AND statusCategory != Done'}
|
||||
assert "history" not in queue.batch_view(conn, b["id"])["commands"][0]["params"]
|
||||
|
||||
|
||||
def test_regular_refresh_gets_since(conn, add_project): # «Регулярный refresh» и «Журнал не дочитан»
|
||||
add_project(conn)
|
||||
_full_run(conn, {"history": {"issues": 1, "changes": 2, "incomplete": []}}, "2026-10-22T09:00:00+00:00")
|
||||
_full_run(conn, {"history": {"error": "HTTP 500"}}, "2026-10-22T10:00:00+00:00")
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": projects.base_jql(conn)}}])
|
||||
assert queue.claim(conn)["params"]["history"] == {"since": "2026-10-22T08:45:00Z"}
|
||||
|
||||
|
||||
def test_new_project_gets_since_and_backfill(conn, add_project): # «Новый проект»
|
||||
add_project(conn)
|
||||
_full_run(conn, {"history": {"issues": 0, "changes": 0, "incomplete": []}}, "2026-10-22T09:00:00+00:00")
|
||||
add_project(conn, "STOCKS_DRRK", board_id=5120)
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": projects.base_jql(conn)}}])
|
||||
h = queue.claim(conn)["params"]["history"]
|
||||
assert h["since"] == "2026-10-22T08:45:00Z"
|
||||
assert h["jql"] == "(" + projects.base_jql(conn) + ") AND sprint in openSprints() AND statusCategory != Done"
|
||||
assert "STOCKS_DRRK" in h["jql"]
|
||||
|
||||
|
||||
def test_tail_refresh_since_and_no_mark(conn, add_project): # «Хвост пакета»
|
||||
add_project(conn)
|
||||
_full_run(conn, {"history": {"issues": 0, "changes": 0, "incomplete": []}}, "2026-10-22T09:00:00+00:00")
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
c = queue.claim(conn)
|
||||
assert c["params"]["history"] == {"since": "2026-10-22T08:45:00Z"}
|
||||
queue.report_result(conn, c["id"], "done", result={"history": {"issues": 0, "changes": 0, "incomplete": []}})
|
||||
assert conn.execute("SELECT history_at FROM sync_run WHERE command_id = ?", (c["id"],)).fetchone()["history_at"] is None
|
||||
|
||||
|
||||
def test_tail_refresh_without_mark_gets_nothing(conn, add_project): # «Хвост пакета до первой отметки»
|
||||
add_project(conn)
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
assert "history" not in queue.claim(conn)["params"]
|
||||
|
||||
|
||||
def test_no_projects_no_history(conn):
|
||||
queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
|
||||
assert "history" not in queue.claim(conn)["params"]
|
||||
|
||||
|
||||
def test_redelivery_recomputes_history(conn, add_project): # Review Focus 3
|
||||
add_project(conn)
|
||||
b = queue.create_batch(conn, "r", "api", [{"type": "refresh", "params": {"jql": projects.base_jql(conn)}}])
|
||||
first = queue.claim(conn)
|
||||
again = queue.claim(conn) # отчёта нет — довыдача
|
||||
assert again["id"] == first["id"] and again["params"]["history"] == first["params"]["history"]
|
||||
assert "history" not in queue.batch_view(conn, b["id"])["commands"][0]["params"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("result, status, marked", [
|
||||
({"history": {"issues": 3, "changes": 7, "incomplete": []}}, "done", True),
|
||||
({"history": {"issues": 1, "changes": 2, "incomplete": ["X"], "backfill_error": "HTTP 400"}}, "done", True),
|
||||
({"history": {"issues": 0, "changes": 0, "error": ""}}, "done", True),
|
||||
({"history": {"error": "HTTP 500"}}, "done", False),
|
||||
({"issues": 3}, "done", False), # агент v12
|
||||
({"history": "мусор"}, "done", False),
|
||||
(None, "done", False),
|
||||
({"history": {"issues": 3, "changes": 7, "incomplete": []}}, "failed", False),
|
||||
])
|
||||
def test_report_sets_history_at(conn, add_project, result, status, marked):
|
||||
add_project(conn)
|
||||
c = _full_run(conn, result, status=status)
|
||||
run = conn.execute("SELECT history_at FROM sync_run WHERE command_id = ?", (c["id"],)).fetchone()
|
||||
assert (run["history_at"] is not None) == marked
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user