feat(queue): довыдача команды без отчёта, предел 3 выдачи; R-13 и «свежая claimed» удалены (#9)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-13 22:44:16 +04:00
parent 4f375fd39c
commit 958dbe956f
9 changed files with 153 additions and 90 deletions

View File

@ -7,7 +7,7 @@
| Модуль | Ответственность | Зависит от |
|---|---|---|
| `paths` | каталог данных (`JIRA_HUB_HOME`), пути базы и `backups/` | stdlib |
| `db` | схема SQLite, миграции, `connect` | stdlib |
| `db` | схема SQLite, миграции по `schema_version` (копия базы перед миграцией), `connect` | `backup` |
| `settings` | настройки (`setting`), токены, темп, перенос из `config.example.json` | `db` |
| `mirror` | зеркало задач и связей, `apply_state`, заметки, фильтры | `db` |
| `commands` | контракт типов команд, валидация, временные ключи, оценка, подтверждение по состоянию | `mirror` |
@ -30,7 +30,7 @@
## Границы
- Зеркало пишет только `mirror.apply_state` по запросу агента (и заметки через `PUT /issues/{key}/note`); из очереди и API задач напрямую не правят.
- Хаб не знает пароль Jira и не обращается к Jira; `init` не переносит `auth`, `base_url`, `verify_ssl`, `ca_bundle`.
- `queue` не повторяет команды: нет автоматических ретраев, повтор — новый пакет.
- `queue` не создаёт команд сам: повтор после failed — новый пакет. Выданная без отчёта команда довыдаётся при следующем `claim` (до 3 выдач, затем failed).
- `mcp_server` не импортирует `queue`/`mirror`/`db` напрямую и ходит только через `client`.
- `commands`, `mirror`, `queue` не знают про HTTP; HTTP живёт только в `api`, `client` и `relay` (реле — клиент своего же HTTP).

View File

@ -1,20 +1,20 @@
# jira-hub
## Что это
Хаб моста Jira для АРМ КА: единственный источник истины по задачам проекта AURORA (метка AWPCA) на Mac mini (`192.168.0.106`). Хранит зеркало задач, служебные заметки и очередь команд в SQLite, отдаёт HTTP API в локальной сети (агенту на виртуалке `jira-agent` и интерфейсам) и MCP-сервер для Claude. Главный сценарий: из чата поставить пакет команд, агент на виртуалке выполняет их в Jira по правилу «проверь, потом сделай» и возвращает состояние задач в зеркало. Спеки: `openspec/specs/`; текущего change нет (последний — `openspec/changes/archive/2026-10-13-folder-transport/`).
Хаб моста Jira для АРМ КА: единственный источник истины по задачам проекта AURORA (метка AWPCA) на Mac mini (`192.168.0.106`). Хранит зеркало задач, служебные заметки и очередь команд в SQLite, отдаёт HTTP API в локальной сети (агенту на виртуалке `jira-agent` и интерфейсам) и MCP-сервер для Claude. Главный сценарий: из чата поставить пакет команд, агент на виртуалке выполняет их в Jira по правилу «проверь, потом сделай» и возвращает состояние задач в зеркало. Спеки: `openspec/specs/`; текущий change — `openspec/changes/bridge-reliability/`.
## Стек и ограничения
- Python 3.13, uv, hatchling, pytest; HTTP на stdlib (`http.server`), БД — SQLite (stdlib), MCP — пакет `mcp` (закреплён `<2`, рулинг R-10).
- Хаб не знает пароль Jira и не ходит в Jira: учётные данные и сеть до Jira только у агента.
- Зеркало задач пишет только `mirror.apply_state` по загрузке состояния от агента (`POST /agent/state`); правки идут командами через очередь.
- Хаб не повторяет команды сам: повтор — новая команда по запросу человека/интерфейса.
- Хаб не создаёт команды сам: повтор после failed — новая команда по запросу человека/интерфейса. Команду, выданную агенту и оставшуюся без отчёта, хаб довыдаёт при следующем claim (до 3 выдач, затем failed).
- Данные: `~/.local/share/jira-hub/` (`JIRA_HUB_HOME`): `hub.sqlite`, `backups/`. Порт по умолчанию 8790.
- Два статических токена (в таблице `setting`): агентский открывает только `/agent/*`, клиентский — остальное.
## Архитектура
- `src/jira_hub/` — весь код пакета, без разделения на ядро и адаптеры; чистая логика (`commands`, `mirror`, `queue`, `settings`) принимает соединение SQLite параметром.
- `db` (схема, миграции) ← `settings`, `mirror`, `commands` ← `queue` ← `api` (HTTP, маршруты в `ROUTES`, фоновый `tick`, реле общей папки `relay` (поток в `serve`, настройка `relay_dir`)) ← `cli`.
- `client` (HTTP-клиент хаба) ← `mcp_server` (stdio, инструменты MCP); `importer` → `mirror`/`settings`; `backup` вызывается из `api`.
- `client` (HTTP-клиент хаба) ← `mcp_server` (stdio, инструменты MCP); `importer` → `mirror`/`settings`; `backup` вызывается из `api` и из `db` (копия перед миграцией схемы).
- Подробная карта модулей и границ — `ARCHITECTURE.md`.
## Команды

View File

@ -165,9 +165,10 @@ curl -s -H "Authorization: Bearer $TOKEN" "http://127.0.0.1:8790/issues?sprint=
- Автообновление: если `auto_refresh_minutes > 0` (по умолчанию 60), агент на связи и очередь пуста дольше интервала,
хаб ставит пакет «Автообновление».
- Настройки: `PUT /settings` или инструмент `settings_set`; защищены `settings_version`, `schema_version`, `tokens`.
- Команда «нет ответа от агента»: команда в `claimed` без отчёта. Дождитесь агента (перезапущенный агент отчитается) или
отмените: `POST /commands/{id}/cancel` с `{"force": true}`. Хаб сам команды не повторяет; если факт уже виден после
`refresh`, команда становится `confirmed_by_state`. Повтор: `command_retry`.
- Довыдача: если агент пришёл за новой командой, а выданная ему осталась без отчёта (ответ потерян, агент
перезапущен), хаб выдаёт её снова; `agent_status` показывает такие команды с числом выдач. После 3 выдач без
отчёта команда становится `failed` («агент не отчитался после 3 выдач»), очередь идёт дальше; повтор —
`command_retry`. Отменить взятую команду: `POST /commands/{id}/cancel` с `{"force": true}`.
- Автозапуск `serve` через launchd: `~/Library/LaunchAgents/ru.aepif.jira-hub.plist`
```xml

View File

@ -101,8 +101,8 @@ def fmt_agent(a) -> str:
f"429/503 за сутки: {a.get('rate_limited_24h', 0)}"]
if a.get("warning"):
out.append("внимание: " + a["warning"])
for s in a.get("stuck", []):
out.append(f"нет ответа от агента: #{s['id']} {s['type']} с {s['claimed_at']}")
for s in a.get("redelivered", []):
out.append(f"выдана повторно: #{s['id']} {s['type']}, выдач {s['deliveries']}, последняя {s['claimed_at']}")
return "\n".join(out)

View File

@ -1,5 +1,6 @@
"""Очередь команд: пакеты, выдача агенту по одной, отчёты, отмена, повтор, сверка с зеркалом, автообновление.
Хаб никогда не повторяет команды сам (design.md, решение 4)."""
Новых команд хаб сам не создаёт: повтор после failed — новый пакет по запросу человека. Команду, выданную агенту и
оставшуюся без отчёта, хаб довыдаёт при следующем claim — это доставка той же выдачи (bridge-reliability, решение 1)."""
import json
import logging
from datetime import timedelta
@ -11,7 +12,7 @@ from . import db, mirror, settings
log = logging.getLogger(__name__)
TERMINAL = {"done", "failed", "skipped", "cancelled", "confirmed_by_state"}
STALE_MINUTES = 10 # взятая команда старше — не блокирует выдачу, показывается как «нет ответа»
MAX_DELIVERIES = 3 # выдач без отчёта, после которых команда — failed (bridge-reliability, решение 2)
ONLINE_MINUTES = 5
@ -108,31 +109,16 @@ def _definer_status(conn, batch_id, temp_key):
return r["status"] if r else "failed"
def _definer_stale(conn, batch_id, temp_key) -> bool:
"""Определитель взят агентом и давно не отвечает (старше STALE_MINUTES)."""
cutoff = (db.utcnow() - timedelta(minutes=STALE_MINUTES)).isoformat()
return conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type = 'ensure_issue' AND status = 'claimed' "
"AND claimed_at <= ? AND json_extract(params, '$.temp_key') = ?",
(batch_id, cutoff, temp_key)).fetchone() is not None
def _next(conn) -> dict | None:
rows = conn.execute("SELECT c.* FROM command c JOIN batch b ON b.id = c.batch_id "
"WHERE c.status = 'queued' AND b.cancelled_at IS NULL ORDER BY b.created_at, b.id, c.seq").fetchall()
blocked_batches = set() # пакеты, ждущие зависшего определителя (R-13)
for row in rows:
if row["batch_id"] in blocked_batches:
continue
cmd = _cmd(row)
needed = C.temp_keys_used(cmd)
if needed:
mapping = _mapping(conn, cmd["batch_id"])
unresolved = [k for k in needed if k not in mapping]
pending = [k for k in unresolved if _definer_status(conn, cmd["batch_id"], k) not in TERMINAL]
if pending:
if all(_definer_stale(conn, cmd["batch_id"], k) for k in pending):
blocked_batches.add(cmd["batch_id"]) # определитель завис: остаток пакета ждёт человека,
continue # очередь идёт к следующему пакету
if any(_definer_status(conn, cmd["batch_id"], k) not in TERMINAL for k in unresolved):
return None # создание ещё не завершено — ждём (одна в полёте)
cmd["params"] = C.substitute(cmd["type"], cmd["params"], mapping)
if cmd["type"] == "refresh":
@ -146,9 +132,20 @@ def _next(conn) -> dict | None:
return None
def _fresh_claimed_exists(conn) -> bool:
cutoff = (db.utcnow() - timedelta(minutes=STALE_MINUTES)).isoformat()
return conn.execute("SELECT 1 FROM command WHERE status = 'claimed' AND claimed_at > ? LIMIT 1", (cutoff,)).fetchone() is not None
def _redeliver(conn, now) -> dict | None:
"""Самая старая взятая команда без отчёта — выдать снова. Агент приходит за новой командой, только когда своей
незавершённой нет, значит выданную он не получил или потерял. После MAX_DELIVERIES выдач — failed."""
while True:
row = conn.execute("SELECT * FROM command WHERE status = 'claimed' ORDER BY claimed_at, id LIMIT 1").fetchone()
if row is None:
return None
cmd = _cmd(row)
if cmd["deliveries"] < MAX_DELIVERIES:
return cmd
conn.execute("UPDATE command SET status = 'failed', finished_at = ?, error = ? WHERE id = ?",
(now, f"агент не отчитался после {MAX_DELIVERIES} выдач", cmd["id"]))
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1 AND current_command_id = ?", (cmd["id"],))
log.warning("команда %s: агент не отчитался после %s выдач — failed", cmd["id"], MAX_DELIVERIES)
def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=None, backup=None) -> dict | None:
@ -158,26 +155,32 @@ def claim(conn, agent_version=None, settings_version=None, last_jira_call_at=Non
conn.execute("UPDATE agent_status SET last_seen_at = ?, agent_version = COALESCE(?, agent_version), "
"settings_version = COALESCE(?, settings_version), last_jira_call_at = COALESCE(?, last_jira_call_at) WHERE id = 1",
(now, None if agent_version is None else str(agent_version), settings_version, last_jira_call_at))
if _fresh_claimed_exists(conn):
return None
cmd = _next(conn)
if cmd is None:
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1")
return None
pause = int(settings.pacing(conn).get(cmd["type"], 0))
first = conn.execute("SELECT COUNT(*) AS n FROM command WHERE batch_id = ? AND status != 'queued'",
(cmd["batch_id"],)).fetchone()["n"] == 0
has_writes = conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type != 'refresh' LIMIT 1",
(cmd["batch_id"],)).fetchone() is not None
conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, pause_seconds = ?, params = ? WHERE id = ?",
(now, pause, _dump(cmd["params"]), cmd["id"]))
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],))
if cmd["type"] == "refresh":
p = cmd["params"]
conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, ?, ?, ?)",
(now, "jql" if p.get("jql") else "keys", p.get("jql") or ",".join(p.get("keys", [])), cmd["id"]))
if first and has_writes:
to_backup = f"batch-{cmd['batch_id']}"
cmd = _redeliver(conn, now)
if cmd is not None:
pause = int(settings.pacing(conn).get(cmd["type"], 0))
conn.execute("UPDATE command SET claimed_at = ?, pause_seconds = ?, deliveries = deliveries + 1 WHERE id = ?",
(now, pause, cmd["id"]))
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],))
log.info("команда %s выдана снова (выдача %s)", cmd["id"], cmd["deliveries"] + 1)
else:
cmd = _next(conn)
if cmd is None:
conn.execute("UPDATE agent_status SET current_command_id = NULL WHERE id = 1")
return None
pause = int(settings.pacing(conn).get(cmd["type"], 0))
first = conn.execute("SELECT COUNT(*) AS n FROM command WHERE batch_id = ? AND status != 'queued'",
(cmd["batch_id"],)).fetchone()["n"] == 0
has_writes = conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type != 'refresh' LIMIT 1",
(cmd["batch_id"],)).fetchone() is not None
conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, pause_seconds = ?, params = ?, deliveries = 1 "
"WHERE id = ?", (now, pause, _dump(cmd["params"]), cmd["id"]))
conn.execute("UPDATE agent_status SET current_command_id = ? WHERE id = 1", (cmd["id"],))
if cmd["type"] == "refresh":
p = cmd["params"]
conn.execute("INSERT INTO sync_run(started_at, kind, query, command_id) VALUES (?, ?, ?, ?)",
(now, "jql" if p.get("jql") else "keys", p.get("jql") or ",".join(p.get("keys", [])), cmd["id"]))
if first and has_writes:
to_backup = f"batch-{cmd['batch_id']}"
if to_backup and backup:
try:
backup(to_backup)
@ -296,9 +299,8 @@ def agent_view(conn) -> dict:
d["warning"] = "обновите агента" if d["agent_version"] and int(d["agent_version"]) < min_version else None
d["rate_limited_24h"] = conn.execute("SELECT COALESCE(SUM(rate_limited), 0) AS n FROM command WHERE finished_at >= ?",
((now - timedelta(days=1)).isoformat(),)).fetchone()["n"]
cutoff = (now - timedelta(minutes=STALE_MINUTES)).isoformat()
d["stuck"] = [dict(r) for r in conn.execute(
"SELECT id, type, claimed_at FROM command WHERE status = 'claimed' AND claimed_at <= ? ORDER BY id", (cutoff,))]
d["redelivered"] = [dict(r) for r in conn.execute(
"SELECT id, type, claimed_at, deliveries FROM command WHERE status = 'claimed' AND deliveries > 1 ORDER BY id")]
return d
@ -306,7 +308,7 @@ def maybe_auto_refresh(conn) -> dict | None:
minutes = int(settings.get(conn, "auto_refresh_minutes") or 0)
if minutes <= 0 or not agent_view(conn)["online"]:
return None
if conn.execute("SELECT 1 FROM command WHERE status = 'queued' LIMIT 1").fetchone() or _fresh_claimed_exists(conn):
if conn.execute("SELECT 1 FROM command WHERE status IN ('queued', 'claimed') LIMIT 1").fetchone():
return None
last = conn.execute("SELECT MAX(started_at) AS t FROM sync_run").fetchone()["t"]
if last and db.utcnow() - db.parse(last) < timedelta(minutes=minutes):

View File

@ -51,7 +51,7 @@ def test_agent_flow(hub):
assert cmd["type"] == "ensure_fields" and cmd["pause_seconds"] == 120 and claim["settings_version"] >= 1
assert a.get(f"/agent/commands/{cmd['id']}")["id"] == cmd["id"]
assert a.get("/agent/settings")["link_types"]["связанные"] == "ASUPR_Связанные"
assert a.post("/agent/claim", {})["command"] is None # одна в полёте, долгий запрос выключен
assert a.post("/agent/claim", {})["command"]["id"] == cmd["id"] # ответ потерян — довыдача той же команды
r = a.post(f"/agent/commands/{cmd['id']}/result", {"status": "done", "result": {"changed": ["duedate"]}, "jira_calls": 2})
assert r["status"] == "done"
assert a.post(f"/agent/commands/{cmd['id']}/result", {"status": "failed", "error": "x"})["status"] == "done"

View File

@ -44,8 +44,8 @@ def test_create_link_sprint_flow(live):
view = client.get(f"/batches/{bid}")
assert view["status"] == "running" and view["remaining_seconds"] > 0
if cmd["type"] == "ensure_issue":
# одна команда в полёте: зависимая ensure_link не выдаётся до done ensure_issue
assert agent.post("/agent/claim", {"agent_version": 1, "settings_version": 1})["command"] is None
# одна команда в полёте: повторный claim довыдаёт ensure_issue, зависимая ensure_link не выдаётся
assert agent.post("/agent/claim", {"agent_version": 1, "settings_version": 1})["command"]["id"] == cmd["id"]
agent.post(f"/agent/commands/{cmd['id']}/result", {"status": "done", "result": {"key": "A-90"}, "jira_calls": 2})
# done ensure_issue не пишет в зеркало; заметка переехала с new:60 на A-90
with pytest.raises(HubError) as e:

View File

@ -70,8 +70,13 @@ def test_formatters():
"sprint": None, "fact_status": None}])
assert "x" * 79 + "…" in text and "x" * 80 not in text and "—" in text
assert "очередь" in mcp_server.fmt_agent({"online": False, "agent_version": None, "current_command_id": None,
"rate_limited_24h": 0, "stuck": [], "warning": None, "last_seen_at": None,
"rate_limited_24h": 0, "redelivered": [], "warning": None, "last_seen_at": None,
"settings_version": None})
text = mcp_server.fmt_agent({"online": True, "agent_version": "2", "current_command_id": 40, "rate_limited_24h": 0,
"redelivered": [{"id": 40, "type": "ensure_fields", "claimed_at": "2026-10-13T10:00:00+00:00",
"deliveries": 2}],
"warning": None, "last_seen_at": "2026-10-13T10:00:00+00:00", "settings_version": 3})
assert "онлайн" in text and "#40" in text and "выдач 2" in text
def test_settings_set_bad_json(live):

View File

@ -42,7 +42,9 @@ def test_claim_order_pause_and_report():
b2 = queue.create_batch(conn, "второй", "claude", [fields("A-2")])
c = queue.claim(conn, agent_version=1, settings_version=1)
assert c["type"] == "ensure_fields" and c["params"]["key"] == "A-1" and c["pause_seconds"] == 120 and c["batch_id"] == b1["id"]
assert queue.claim(conn) is None # одна команда в полёте
again = queue.claim(conn) # ответ потерян — та же команда снова
assert again["id"] == c["id"] and again["pause_seconds"] == 120
assert queue.get_command(conn, c["id"])["deliveries"] == 2
assert queue.agent_view(conn)["current_command_id"] == c["id"]
done = queue.report_result(conn, c["id"], "done", result={"changed": ["duedate"]}, jira_calls=2)
assert done["status"] == "done" and done["jira_calls"] == 2
@ -75,8 +77,10 @@ def test_claim_is_exclusive(tmp_path):
threads = [threading.Thread(target=worker, args=(c,)) for c in conns]
[t.start() for t in threads]
[t.join() for t in threads]
assert sorted(r is None for r in results) == [False, True]
assert setup.execute("SELECT COUNT(*) AS n FROM command WHERE status = 'claimed'").fetchone()["n"] == 1
got = [r for r in results if r is not None]
assert got and len({r["id"] for r in got}) == 1 # один и тот же id при любом порядке потоков
row = setup.execute("SELECT COUNT(*) AS n, MAX(deliveries) AS d FROM command WHERE status = 'claimed'").fetchone()
assert row["n"] == 1 and row["d"] == len(got)
def test_backup_failure_does_not_block_claim(caplog):
@ -92,16 +96,6 @@ def test_backup_failure_does_not_block_claim(caplog):
assert "не создана" in caplog.text
def test_stale_claimed_does_not_block():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
c = queue.claim(conn)
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
nxt = queue.claim(conn)
assert nxt["params"]["key"] == "A-2"
assert queue.agent_view(conn)["stuck"][0]["id"] == c["id"]
def test_temp_key_substitution_and_skip():
conn = fresh()
mirror.set_note(conn, "new:60", {"fact_status": "В работе"}, "import")
@ -134,7 +128,7 @@ def test_cancel_and_retry():
queue.cancel_command(conn, c["id"])
view = queue.cancel_batch(conn, b["id"])
assert [x["status"] for x in view["commands"]] == ["claimed", "cancelled", "cancelled"]
assert queue.claim(conn) is None
assert queue.claim(conn)["id"] == c["id"] # взятая из отменённого пакета довыдаётся
forced = queue.cancel_command(conn, c["id"], force=True)
assert forced["status"] == "cancelled"
after = queue.report_result(conn, c["id"], "done", result={"changed": []})
@ -157,15 +151,13 @@ def test_reconcile_confirms_by_state():
conn = fresh()
queue.create_batch(conn, "x", "claude", [{"type": "ensure_sprint", "params": {"key": "A-1", "sprint": "Спринт 2"}}])
c = queue.claim(conn)
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (c["id"],))
b = queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
r = queue.claim(conn)
r = queue.batch_view(conn, b["id"])["commands"][0]
conn.execute("UPDATE command SET status = 'claimed', claimed_at = ?, deliveries = 1 WHERE id = ?", (db.now(), r["id"]))
mirror.apply_state(conn, [{"key": "A-1", "summary": "т", "sprint": "Спринт 2", "labels": []}], [])
queue.report_result(conn, r["id"], "done")
assert queue.get_command(conn, c["id"])["status"] == "confirmed_by_state"
assert r["batch_id"] == c["batch_id"] # выдался хвостовой refresh первого пакета
assert queue.batch_view(conn, c["batch_id"])["status"] == "done"
assert queue.batch_view(conn, b["id"])["status"] == "queued"
assert queue.batch_view(conn, b["id"])["status"] == "done"
def test_auto_refresh_conditions():
@ -277,21 +269,84 @@ def _two_batches_with_dependent(conn):
return a, b, first
def test_stale_definer_does_not_block_other_batches():
"""R-13: зависший ensure_issue с зависимыми не держит очередь целиком."""
conn = fresh()
a, b, first = _two_batches_with_dependent(conn)
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (first["id"],))
got = queue.claim(conn)
assert got is not None and got["batch_id"] == b["id"] and got["type"] == "refresh"
# остаток пакета A остаётся в очереди, не выдаётся и не пропускается
assert [c["status"] for c in queue.batch_view(conn, a["id"])["commands"]][1:] == ["queued", "queued"]
def test_fresh_definer_still_blocks_queue():
conn = fresh()
a, b, first = _two_batches_with_dependent(conn)
assert queue.claim(conn) is None # одна в полёте
assert queue.claim(conn)["id"] == first["id"] # довыдача определителя, зависимые ждут
conn.execute("UPDATE command SET status = 'done', result = '{\"key\": \"A-9\"}' WHERE id = ?", (first["id"],))
nxt = queue.claim(conn)
assert nxt["batch_id"] == a["id"] and nxt["params"]["source"] == "A-9"
def test_auto_refresh_waits_for_claimed_without_report():
conn = fresh()
queue.claim(conn, agent_version=1, settings_version=1) # агент на связи
queue.create_batch(conn, "y", "claude", [fields("A-3")])
queue.claim(conn) # выдана, отчёта нет
conn.execute("DELETE FROM command WHERE status = 'queued'") # хвостовой refresh не мешает проверке
conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE status = 'claimed'") # давно без отчёта
assert queue.maybe_auto_refresh(conn) is None # claimed без отчёта — очередь занята
def test_redelivery_of_lost_claim_before_queue():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
c = queue.claim(conn)
again = queue.claim(conn)
assert again["id"] == c["id"] and again["params"] == c["params"]
assert queue.agent_view(conn)["current_command_id"] == c["id"]
view = queue.agent_view(conn)["redelivered"]
assert [(x["id"], x["deliveries"]) for x in view] == [(c["id"], 2)]
queue.report_result(conn, c["id"], "done")
assert queue.claim(conn)["params"]["key"] == "A-2"
assert queue.agent_view(conn)["redelivered"] == []
def test_redelivery_limit_fails_command():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1"), fields("A-2")])
c = queue.claim(conn)
queue.claim(conn)
queue.claim(conn) # третья выдача
nxt = queue.claim(conn) # четвёртой нет — failed и дальше по очереди
assert nxt["params"]["key"] == "A-2"
failed = queue.get_command(conn, c["id"])
assert failed["status"] == "failed" and failed["error"] == "агент не отчитался после 3 выдач" and failed["finished_at"]
assert queue.MAX_DELIVERIES == 3
def test_redelivery_oldest_first():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
queue.create_batch(conn, "y", "claude", [fields("A-2")])
rows = conn.execute("SELECT id FROM command WHERE type = 'ensure_fields' ORDER BY id").fetchall()
conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-13T10:00:00+00:00', deliveries = 1 WHERE id = ?", (rows[1]["id"],))
conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-13T11:00:00+00:00', deliveries = 1 WHERE id = ?", (rows[0]["id"],))
assert queue.claim(conn)["id"] == rows[1]["id"]
def test_redelivered_refresh_keeps_one_sync_run():
conn = fresh()
queue.create_batch(conn, "обновить", "claude", [{"type": "refresh", "params": {"keys": ["A-1"]}}])
r = queue.claim(conn)
assert queue.claim(conn)["id"] == r["id"]
assert conn.execute("SELECT COUNT(*) AS n FROM sync_run WHERE command_id = ?", (r["id"],)).fetchone()["n"] == 1
def test_redelivery_does_not_repeat_backup():
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
labels = []
queue.claim(conn, backup=labels.append)
queue.claim(conn, backup=labels.append)
assert len(labels) == 1
def test_migrated_claimed_is_redelivered():
"""Команда, взятая до миграции (deliveries = 1 после неё), довыдаётся со счётчиком 2."""
conn = fresh()
queue.create_batch(conn, "x", "claude", [fields("A-1")])
cid = conn.execute("SELECT id FROM command WHERE type = 'ensure_fields'").fetchone()["id"]
conn.execute("UPDATE command SET status = 'claimed', claimed_at = '2026-10-12T00:00:00+00:00', deliveries = 1 WHERE id = ?", (cid,))
assert queue.claim(conn)["id"] == cid
assert queue.get_command(conn, cid)["deliveries"] == 2