diff --git a/.gitignore b/.gitignore index ca0fe32..fb103f0 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ data.json tests/fixtures/*.m4a tests/fixtures/*.webm .DS_Store +journal.jsonl.lock diff --git a/README.md b/README.md index 7430e72..8d232b2 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ выполняет команды «желаемый результат» в Jira по правилу «проверь, потом сделай», ведёт журнал на диске и возвращает хабу состояние задач. Один файл `src/agent.py`, только стандартная библиотека, Python 3.9. -Спека: `../jira-hub/openspec/changes/bridge-core/specs/agent/spec.md` (вместе с `commands/` и `hub-api/`). +Спека: `../jira-hub/openspec/specs/agent/spec.md` (вместе с `commands/` и `hub-api/`). Хаб, очередь команд и зеркало задач живут в репозитории `jira-hub`; агент с ним общается только по HTTP. ## Доставка на виртуалку @@ -70,7 +70,7 @@ nohup python3 -u agent.py > agent.log 2>&1 & Остановка: Ctrl-C или `kill `. Первый сигнал — агент доделывает текущую команду, отправляет отчёт и выходит. Повторный сигнал — немедленный выход (код 130). Остановка во время паузы перед командой безопасна: команда уже -записана в журнал как `claimed`, при следующем запуске агент запрашивает её у хаба и выполняет через проверку (без паузы). +записана в журнал как `claimed`, при следующем запуске агент запрашивает её у хаба и выполняет через проверку, выдержав паузу темпа от последней записи журнала. ## Автозапуск через launchd (опционально) @@ -140,8 +140,11 @@ python3 agent.py prune Правила работы: - **Проверь, потом сделай.** Каждый тип сначала читает текущее состояние и записывает только разницу. - **Паузы.** Хаб выдаёт команду с `pause_seconds` — агент досыпает до зазора от последнего вызова Jira. Внутри команды - с несколькими записями зазор берётся из `pacing` настроек хаба. + с несколькими записями зазор берётся из `pacing` настроек хаба. После перезапуска время последнего вызова Jira + берётся из журнала (самая поздняя запись), поэтому первая команда тоже ждёт паузу. - **Одна команда в полёте.** Агент берёт следующую команду, только когда предыдущая выполнена и отчёт подтверждён. + На этом строится довыдача: если ответ на `claim` потерян, хаб выдаст ту же команду снова при следующем `claim` + (до 3 выдач, затем `failed`). - **Отчёт до подтверждения.** Отчёт повторяется каждые 10 с, пока хаб не подтвердит. Ответы 404/409/422 хаб выдаёт, когда команда уже отменена или удалена: агент считает её закрытой и не зацикливается. - **429 и 503.** Агент ждёт `Retry-After` из ответа Jira и повторяет запрос: до 5 запросов, то есть не более 4 ожиданий; @@ -150,7 +153,7 @@ python3 agent.py prune `finished`, ждёт не меньше 30 с и повторяет команду через проверку («проверь, потом сделай») до 3 раз. Счётчики суммируются. После исчерпания — `failed` «результат неизвестен (обрыв связи с Jira), повтор безопасен». При остановке по сигналу команду подхватит восстановление. -- **Версия.** Константа `AGENT_VERSION` в `agent.py` (сейчас 2) отправляется в каждом `claim`, в каждом отчёте +- **Версия.** Константа `AGENT_VERSION` в `agent.py` (сейчас 3) отправляется в каждом `claim`, в каждом отчёте (`agent_version`) и в `/agent/state`. Если хаб назначил `min_agent_version` выше, он пишет в `agent_status.warning`: «обновите агента»; команды при этом выдаются по-прежнему. @@ -167,7 +170,10 @@ python3 agent.py prune ## Обновление агента -Скопировать новый `agent.py` поверх старого и перезапустить процесс (или `launchctl unload` + `load`). +Остановить старый процесс (Ctrl-C или `kill `, дождаться «остановлен»), скопировать новый `agent.py` поверх +старого и запустить снова (или `launchctl unload` + `load`). Два процесса с одним журналом работать не могут: хаб +довыдаёт команду без отчёта, и второй процесс выполнял бы её одновременно с первым. Начиная с версии 3 второй процесс +не стартует («агент уже запущен…», код 1) — блокировка `journal.jsonl.lock` рядом с журналом. `agent.json` и `journal.jsonl` не трогать: журнал переживает обновление, а неподтверждённые отчёты досылаются при старте. ## Известные ограничения / эксплуатация @@ -177,9 +183,9 @@ python3 agent.py prune 1. **`prune` не запускать при работающем агенте.** Блокировки нет. Запись, добавленная агентом во время `prune`, может пропасть при перезаписи файла. Сначала остановить агента. 2. **Остановка во время паузы или между claim и выполнением (R-9).** Команда остаётся `claimed` в хабе, но в журнале - есть запись `claimed`: при перезапуске агент запрашивает её у хаба и выполняет без паузы. Отмена с force и повтор - больше не нужны. Если журнал потерян, команду придётся отменить с force (`POST /commands/{id}/cancel`, - `{"force": true}`) и повторить (`POST /commands/{id}/retry`). + есть запись `claimed`: при перезапуске агент запрашивает её у хаба и выполняет с паузой темпа (время последнего + вызова Jira берётся из журнала). Отмена с force и повтор не нужны. Если журнал потерян, хаб довыдаст команду при + следующем `claim` — агент выполнит её через проверку. 3. **HTTPS хаба с корпоративным CA не поддержан.** `HubClient` работает без `ca_bundle` и `verify_ssl`: `hub_url` должен быть `http://`. Для Jira `verify_ssl` и `ca_bundle` работают. 4. **Спринт в `refresh` — только активные и будущие спринты доски.** Поле `sprint` берётся из Agile API по diff --git a/docs/superpowers/plans/2026-10-13-bridge-reliability-agent.md b/docs/superpowers/plans/2026-10-13-bridge-reliability-agent.md new file mode 100644 index 0000000..37a79fd --- /dev/null +++ b/docs/superpowers/plans/2026-10-13-bridge-reliability-agent.md @@ -0,0 +1,265 @@ +# bridge-reliability — агент: план реализации + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** после перезапуска агент соблюдает темп Jira (время последнего вызова восстанавливается из журнала, пауза при +восстановлении не обнуляется) и не пишет `claimed` в журнал повторно; версия агента 3. + +**Architecture:** всё в `src/agent.py`: `Journal.last_at()` даёт самое позднее время записи; `Runner.restore_pace()` +один раз в начале `loop()` переводит его в `JiraClient.last_call_at`; `Runner.execute(cmd, journal_claimed=True)`; +`Runner.recover()` больше не обнуляет `pause_seconds` и вызывает `execute(cmd, journal_claimed=False)`. + +**Tech Stack:** Python 3.9, только стандартная библиотека, `unittest`. + +**Spec:** `../jira-hub/openspec/changes/bridge-reliability/` (ветка `feat/9-bridge-reliability` в jira-hub; +`specs/agent/spec.md` — требования «Цикл выполнения», «Журнал на диске и восстановление», «Ограничение темпа и 429», +«Доставка и запуск»; `design.md` — решения 9–11). + +## Global Constraints + +- Python ≥ 3.9, только stdlib (`tests/test_py39.py` проверяет синтаксис и импорты). +- Тесты: `/usr/bin/python3 -W error::ResourceWarning -m unittest discover -s tests -t .` из корня worktree — зелёным + считается только весь набор (до начала: 133 OK). +- Тела запросов к хабу не меняются. +- Комментарии, сообщения лога и коммитов — по-русски; коммиты через + `git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/reliability …`, суффикс `(#9)`, строка + `Co-Authored-By: Claude Opus 5.5 `. +- Решения без обсуждения — в `openspec/changes/bridge-reliability/design.md` этого репозитория (создать при первом + решении: заголовок `# bridge-reliability — решения агента`, ссылка на спеку хаба, таблица + `## Decisions made without discussion` с колонками Дата | Решение | Обоснование | Цена ошибки), маркер `R-n`. + +## Review Focus + +1. Журнал со временем записи в будущем (часы виртуалки сдвинулись назад после записи) — не ставить паузу больше + положенной и не падать: оценка игнорируется — тест в Task 1. +2. Журнал с битой строкой или записью без `at` — восстановление темпа берёт годные записи и не падает — тест в Task 1. +3. Тесты, где `last_call_at` задан вручную до `loop()`, — восстановление из журнала его не перетирает — тест в Task 1. +4. Восстановление команды, выданной до остановки, при пустом журнале кроме этой записи — пауза отсчитывается от + `claimed`/`started` этой записи, а не обнуляется — тест в Task 2. +5. Обычная выдача (не восстановление) по-прежнему пишет `claimed` до паузы (R-9) — существующий + `test_processes_queue` остаётся зелёным, Task 2. + +--- + +### Task 1: Время последнего вызова Jira восстанавливается из журнала + +**Files:** +- Modify: `src/agent.py` (`Journal.last_at`, `Runner.restore_pace`, `Runner.loop`) +- Test: `tests/test_runner.py`, `tests/test_journal.py` + +**Interfaces:** +- Produces: `Journal.last_at() -> Optional[datetime]` (aware UTC, самое позднее `at` среди годных записей или `None`); + `Runner.restore_pace() -> None` (ставит `jira.last_call_at`/`last_call_iso`, если они пусты и в журнале есть время не + из будущего). + +- [ ] **Step 1: Write the failing tests** + +`tests/test_journal.py` (в существующий класс тестов журнала; имена `self.path`/`self.journal` взять из его `setUp`; +при необходимости `from datetime import datetime, timedelta, timezone`): + +```python + def test_last_at_is_latest_record_time(self): + self.assertIsNone(self.journal.last_at()) + self.journal.append({"kind": "claimed", "command_id": 1, "at": "2026-10-13T10:00:00+00:00"}) + self.journal.append({"kind": "finished", "command_id": 1, "at": "2026-10-13T10:05:00+00:00"}) + self.journal.append({"kind": "acked", "command_id": 1, "at": "2026-10-13T10:04:00+00:00"}) + self.assertEqual(self.journal.last_at(), datetime(2026, 10, 13, 10, 5, tzinfo=timezone.utc)) + + def test_last_at_skips_bad_and_missing_times(self): + self.journal.append({"kind": "claimed", "command_id": 1, "at": "2026-10-13T10:00:00+00:00"}) + self.journal.append({"kind": "started", "command_id": 1, "at": "не время"}) + self.journal.append({"kind": "started", "command_id": 1, "at": "2026-10-13T11:00:00"}) # без зоны — пропуск + with open(self.journal.path, "a", encoding="utf-8") as f: + f.write("битая строка\n") + self.assertEqual(self.journal.last_at(), datetime(2026, 10, 13, 10, 0, tzinfo=timezone.utc)) +``` + +`tests/test_runner.py` (класс `RunnerTest`; импорт `from datetime import datetime, timedelta, timezone` в начало файла): + +```python + def test_restart_pace_from_old_finished(self): + at = (datetime.now(timezone.utc) - timedelta(seconds=10)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "finished", "command_id": 99, "status": "done", "result": None, "error": None, + "jira_calls": 1, "rate_limited": 0, "at": at}) + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=30) + self.runner.loop() + self.assertAlmostEqual(sum(self.sleeps), 20, delta=1.5) + + def test_future_journal_time_ignored(self): + at = (datetime.now(timezone.utc) + timedelta(hours=1)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=30) + self.runner.loop() + self.assertEqual(self.sleeps, []) # время из будущего — как пустой журнал + self.assertEqual(self.hub.results[1]["status"], "done") + + def test_restore_pace_keeps_existing_last_call(self): + at = (datetime.now(timezone.utc) - timedelta(seconds=100)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + mark = time.monotonic() + self.runner.jira.last_call_at = mark + self.runner.restore_pace() + self.assertEqual(self.runner.jira.last_call_at, mark) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `/usr/bin/python3 -W error::ResourceWarning -m unittest tests.test_journal tests.test_runner` +Expected: FAIL — нет `Journal.last_at`, нет `Runner.restore_pace`, пауза после перезапуска 0. + +- [ ] **Step 3: Implement** — `src/agent.py`: + +В класс `Journal` после `records`: + +```python + def last_at(self) -> Optional[datetime]: + """Самое позднее время записи журнала (UTC). Записи без времени или с непонятным временем пропускаются.""" + latest = None + for r in self.records(): + try: + t = datetime.fromisoformat(r["at"]) + except (KeyError, TypeError, ValueError): + continue + if t.tzinfo is None: + continue + if latest is None or t > latest: + latest = t + return latest +``` + +В класс `Runner` после `load_settings`: + +```python + def restore_pace(self) -> None: + """После перезапуска время последнего вызова Jira неизвестно (живёт в памяти). Берём самое позднее время записи + журнала: finished пишется сразу после последнего вызова команды, оценка не раньше настоящего вызова (M-9).""" + if self.jira.last_call_at is not None: + return + at = self.journal.last_at() + if at is None: + return + ago = (datetime.now(timezone.utc) - at).total_seconds() + if ago < 0: + log("журнал: время последней записи в будущем (%s) — темп не восстанавливаю" % at.isoformat()) + return + self.jira.last_call_at = time.monotonic() - ago + self.jira.last_call_iso = at.astimezone(timezone.utc).replace(microsecond=0).isoformat() + log("темп: последний вызов Jira не позже %s (по журналу)" % self.jira.last_call_iso) +``` + +В `loop()` первой строкой тела (до `idle, recovered = 0, False` или сразу после неё): `self.restore_pace()`. + +- [ ] **Step 4: Run tests** + +Run: полный набор (см. Global Constraints) — OK. Существующие тесты перезапуска +(`test_stop_during_pause_then_restart_executes_once`, `test_recover_unfinished_started`) остаются зелёными: `recover` +до Task 2 обнуляет `pause_seconds`, поэтому восстановленное время последнего вызова на них не влияет. + +- [ ] **Step 5: Commit** + +```bash +git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/reliability add src/agent.py tests +git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/reliability commit -m "feat(agent): время последнего вызова Jira восстанавливается из журнала после перезапуска (M-9, #9) + +Co-Authored-By: Claude Opus 5.5 " +``` + +--- + +### Task 2: Восстановление без обнуления паузы и без повторного `claimed`; версия 3 + +**Files:** +- Modify: `src/agent.py` (`AGENT_VERSION`, `Runner.execute`, `Runner.recover`, docstring про спеку в начале файла, + если он ссылается на `changes/bridge-core`) +- Modify: `README.md` (строки про «выполняет через проверку (без паузы)», «Версия … (сейчас 2)», раздел про R-9) +- Test: `tests/test_runner.py`, `tests/test_folder.py` + +**Interfaces:** +- Consumes: `Runner.restore_pace` (Task 1). +- Produces: `Runner.execute(cmd: dict, journal_claimed: bool = True) -> None`; `AGENT_VERSION = 3`. + +- [ ] **Step 1: Write the failing tests** — `tests/test_runner.py`: + +1. `test_recover_unfinished_started` — заменить проверку пауз и добавить проверку журнала: + +```python + def test_recover_unfinished_started(self): + cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"}, pause=99) + self.journal.started(cid) + self.runner.loop() + self.assertEqual(self.hub.results[cid]["result"], {"changed": []}) # уже так в Jira — записи нет + self.assertEqual(self.journal.unfinished_started(), []) + self.assertAlmostEqual(sum(self.sleeps), 99, delta=1) # пауза из хаба, от записи started (M-9) + self.assertEqual([r[0] for r in self.fake.writes()], []) +``` + +2. Новый тест FM-4: + +```python + def test_recover_does_not_journal_claimed_twice(self): + cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"}, pause=0) + self.journal.claimed(cid) + self.runner.loop() + self.assertEqual(self.kinds(), [("claimed", cid), ("started", cid), ("finished", cid), ("acked", cid)]) +``` + +3. `test_stop_during_pause_then_restart_executes_once` — последнюю проверку заменить на + `self.assertAlmostEqual(sum(self.sleeps), 5, delta=1) # пауза из хаба от записи claimed (M-9)`. +4. `tests/test_folder.py`: `test_version_is_2` → `test_version_is_3` с `self.assertEqual(agent.AGENT_VERSION, 3)`. + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `/usr/bin/python3 -W error::ResourceWarning -m unittest tests.test_runner tests.test_folder` +Expected: FAIL — пауза при восстановлении 0; `claimed` в журнале дважды; версия 2. + +- [ ] **Step 3: Implement** — `src/agent.py`: + +1. `AGENT_VERSION = 3`. +2. `execute`: + +```python + def execute(self, cmd: dict, journal_claimed: bool = True) -> None: + cid = cmd["id"] + log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200])) + if journal_claimed: + self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду + if not self.wait_pause(cmd.get("pause_seconds")): + return + # дальше без изменений +``` + +3. `recover` — конец цикла: + +```python + log("#%s выдана до перезапуска — выполняю через проверку" % cid) + self.execute(cmd, journal_claimed=False) # claimed/started для неё уже в журнале (FM-4); пауза — из хаба (M-9) +``` + + (строка `cmd["pause_seconds"] = 0` удаляется.) +4. Если docstring модуля/справка `--help` ссылается на `jira-hub/openspec/changes/bridge-core/specs/agent/spec.md` — + заменить на `jira-hub/openspec/specs/agent/spec.md`. + +`README.md`: +- «…при следующем запуске агент запрашивает её у хаба и выполняет через проверку (без паузы).» → + «…выполняет через проверку, выдержав паузу темпа от последней записи журнала.» +- «Константа `AGENT_VERSION` в `agent.py` (сейчас 2)» → «(сейчас 3)». +- В разделе ограничений, пункт 2 (R-9): «выполняет без паузы» → «выполняет с паузой темпа (время последнего вызова Jira + берётся из журнала)». +- В раздел про паузы (строка «**Паузы.** …») дописать предложение: «После перезапуска время последнего вызова Jira + берётся из журнала (самая поздняя запись), поэтому первая команда тоже ждёт паузу.» +- Если в README есть упоминание, что хаб ждёт 10 минут / «нет ответа от агента», — заменить на довыдачу: «Если ответ на + `claim` потерян, хаб выдаст ту же команду снова при следующем `claim` (до 3 выдач).» + +- [ ] **Step 4: Run tests** + +Run: полный набор (см. Global Constraints) — OK. + +- [ ] **Step 5: Commit** + +```bash +git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/reliability add -A src tests README.md +git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/reliability commit -m "feat(agent): восстановление с паузой из хаба и без повторного claimed; версия 3 (M-9, FM-4, #9) + +Co-Authored-By: Claude Opus 5.5 " +``` diff --git a/openspec/changes/bridge-reliability/design.md b/openspec/changes/bridge-reliability/design.md new file mode 100644 index 0000000..f98b87a --- /dev/null +++ b/openspec/changes/bridge-reliability/design.md @@ -0,0 +1,11 @@ +# bridge-reliability — решения агента + +Спека и дизайн изменения — в `jira-hub/openspec/changes/bridge-reliability/` (issue aep/jira-hub#9, aep/jira-agent#9). +Здесь — решения, принятые при исполнении плана агента без обсуждения с владельцем. + +## Decisions made without discussion + +| Дата | Решение | Обоснование | Цена ошибки | +|---|---|---|---| +| 2026-10-13 | `tests/test_folder.py::test_runner_through_folder` сверяет `agent_version` с `agent.AGENT_VERSION`, а не с литералом (R-51) | версию закрепляет отдельный `test_version_is_3`; литерал ломался бы при каждом подъёме версии | нет | +| 2026-10-13 | `fcntl` добавлен в белый список stdlib `tests/test_py39.py` (R-52) | блокировка одного процесса на журнал (финальное ревью I-1, решение 13 спеки хаба) — `fcntl.flock`, stdlib 3.9 | нет: на виртуалке macOS `fcntl` есть | diff --git a/src/agent.py b/src/agent.py index ca8a7ee..79aeaae 100644 --- a/src/agent.py +++ b/src/agent.py @@ -3,7 +3,7 @@ Опрашивает хаб (jira-hub) на Mac mini, выполняет команды «желаемый результат» в Jira по правилу «проверь, потом сделай», ведёт журнал на диске и возвращает хабу состояние задач. Только стандартная библиотека, -Python 3.9. Спека: jira-hub/openspec/changes/bridge-core/specs/agent/spec.md. +Python 3.9. Спека: jira-hub/openspec/specs/agent/spec.md. Связь с хабом — HTTP (`hub_url = http://…`) или общая папка (`hub_url = folder:<путь>`), если сеть закрыта VPN. python3 agent.py check # доступ к хабу и Jira, без записи @@ -12,6 +12,7 @@ Python 3.9. Спека: jira-hub/openspec/changes/bridge-core/specs/agent/spec.m """ import argparse import base64 +import fcntl import http.client import json import os @@ -26,7 +27,7 @@ import uuid from datetime import datetime, timezone from typing import Any, Dict, List, Optional -AGENT_VERSION = 2 +AGENT_VERSION = 3 HERE = os.path.dirname(os.path.abspath(__file__)) API = "/rest/api/2" AGILE = "/rest/agile/1.0" @@ -345,6 +346,20 @@ class Journal: log("журнал: строка %d не читается, пропускаю" % n) return good + def last_at(self) -> Optional[datetime]: + """Самое позднее время записи журнала (UTC). Записи без времени или с непонятным временем пропускаются.""" + latest = None + for r in self.records(): + try: + t = datetime.fromisoformat(r["at"]) + except (KeyError, TypeError, ValueError): + continue + if t.tzinfo is None: + continue + if latest is None or t > latest: + latest = t + return latest + def claimed(self, command_id) -> None: """Команда получена из claim (до паузы): с этого момента хаб считает её выданной агенту.""" self.append({"kind": "claimed", "command_id": int(command_id)}) @@ -916,6 +931,22 @@ class Runner: self.executor.sleep = self.sleep log("настройки хаба v%s (проект %s)" % (self.settings.get("settings_version"), self.settings.get("project_key"))) + def restore_pace(self) -> None: + """После перезапуска время последнего вызова Jira неизвестно (живёт в памяти). Берём самое позднее время записи + журнала: finished пишется сразу после последнего вызова команды, оценка не раньше настоящего вызова (M-9).""" + if self.jira.last_call_at is not None: + return + at = self.journal.last_at() + if at is None: + return + ago = (datetime.now(timezone.utc) - at).total_seconds() + if ago < 0: + log("журнал: время последней записи в будущем (%s) — темп не восстанавливаю" % at.isoformat()) + return + self.jira.last_call_at = time.monotonic() - ago + self.jira.last_call_iso = at.astimezone(timezone.utc).replace(microsecond=0).isoformat() + log("темп: последний вызов Jira не позже %s (по журналу)" % self.jira.last_call_iso) + def send_result(self, rec: dict) -> bool: """Отчёт хабу; повторяется каждые 10 с, пока хаб не подтвердит. 404/409/422 — хаб отверг (команда отменена или удалена): считаем закрытой, чтобы не зациклиться. При остановке по сигналу делается одна попытка.""" @@ -969,10 +1000,11 @@ class Runner: "error": "внутренняя ошибка агента: %s: %s" % (type(e).__name__, str(e)[:300]), "jira_calls": self.jira.calls, "rate_limited": self.jira.rate_limited} - def execute(self, cmd: dict) -> None: + def execute(self, cmd: dict, journal_claimed: bool = True) -> None: cid = cmd["id"] log("#%s %s %s" % (cid, cmd["type"], json.dumps(cmd.get("params"), ensure_ascii=False)[:200])) - self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду + if journal_claimed: + self.journal.claimed(cid) # сразу после claim, до паузы (R-9): остановка не теряет команду if not self.wait_pause(cmd.get("pause_seconds")): return self.journal.started(cid) # started — только после паузы @@ -1034,8 +1066,7 @@ class Runner: self.journal.acked(cid) continue log("#%s выдана до перезапуска — выполняю через проверку" % cid) - cmd["pause_seconds"] = 0 - self.execute(cmd) + self.execute(cmd, journal_claimed=False) # claimed/started для неё уже в журнале (FM-4); пауза — из хаба (M-9) def reload_settings_before(self, cmd) -> bool: """Настройки сменились: перечитать ДО выполнения. False — остановка, команда не должна выполняться @@ -1056,6 +1087,7 @@ class Runner: self.sleep(wait) def loop(self) -> None: + self.restore_pace() idle, recovered = 0, False while not self.stop: if self.settings is None: @@ -1087,7 +1119,30 @@ class Runner: self.sleep(self.poll) +def acquire_lock(journal_path: str): + """Исключительная блокировка <журнал>.lock: один процесс агента на журнал. Хаб довыдаёт команду без отчёта при + следующем claim, поэтому второй процесс выполнял бы ту же команду одновременно с первым. None — уже занято.""" + f = open(journal_path + ".lock", "a") + try: + fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + f.close() + return None + return f + + def cmd_run(cfg: dict, journal_path: str) -> int: + lock = acquire_lock(journal_path) + if lock is None: + log("агент уже запущен с журналом %s — второй процесс не стартует" % journal_path) + return 1 + try: + return _run(cfg, journal_path) + finally: + lock.close() + + +def _run(cfg: dict, journal_path: str) -> int: runner = Runner(cfg, HubClient(cfg), JiraClient(cfg), Journal(journal_path)) def on_signal(signum, frame): diff --git a/tests/test_folder.py b/tests/test_folder.py index 8f4fd72..d9474c5 100644 --- a/tests/test_folder.py +++ b/tests/test_folder.py @@ -135,8 +135,8 @@ class FolderIntegrationTest(unittest.TestCase): def cfg(self): return jira_cfg(self.fake, hub_url="folder:" + self.root, hub_token=self.hub.token) - def test_version_is_2(self): - self.assertEqual(agent.AGENT_VERSION, 2) + def test_version_is_3(self): + self.assertEqual(agent.AGENT_VERSION, 3) def test_runner_through_folder(self): self.fake.add("AURORA-1", duedate="2026-10-20") @@ -149,7 +149,7 @@ class FolderIntegrationTest(unittest.TestCase): runner.idle_limit = 1 runner.loop() self.assertEqual(self.hub.results[cid]["status"], "done") - self.assertEqual(self.hub.results[cid]["agent_version"], 2) + self.assertEqual(self.hub.results[cid]["agent_version"], agent.AGENT_VERSION) self.assertEqual(self.fake.issues["AURORA-1"]["duedate"], "2026-10-23") self.assertIn(("acked", cid), [(r["kind"], r["command_id"]) for r in journal.records()]) diff --git a/tests/test_journal.py b/tests/test_journal.py index 9a99603..5e63fef 100644 --- a/tests/test_journal.py +++ b/tests/test_journal.py @@ -1,5 +1,6 @@ import json import os +from datetime import datetime, timezone import tempfile import unittest from unittest import mock @@ -134,6 +135,20 @@ class JournalTest(unittest.TestCase): mode = os.stat(self.path).st_mode & 0o777 self.assertEqual(mode & 0o077, 0) + def test_last_at_is_latest_record_time(self): + self.assertIsNone(self.j.last_at()) + self.j.append({"kind": "claimed", "command_id": 1, "at": "2026-10-13T10:00:00+00:00"}) + self.j.append({"kind": "finished", "command_id": 1, "at": "2026-10-13T10:05:00+00:00"}) + self.j.append({"kind": "acked", "command_id": 1, "at": "2026-10-13T10:04:00+00:00"}) + self.assertEqual(self.j.last_at(), datetime(2026, 10, 13, 10, 5, tzinfo=timezone.utc)) + + def test_last_at_skips_bad_and_missing_times(self): + self.j.append({"kind": "claimed", "command_id": 1, "at": "2026-10-13T10:00:00+00:00"}) + self.j.append({"kind": "started", "command_id": 1, "at": "не время"}) + self.j.append({"kind": "started", "command_id": 1, "at": "2026-10-13T11:00:00"}) # без зоны — пропуск + with open(self.path, "a", encoding="utf-8") as f: + f.write("битая строка\n") + self.assertEqual(self.j.last_at(), datetime(2026, 10, 13, 10, 0, tzinfo=timezone.utc)) if __name__ == "__main__": unittest.main() diff --git a/tests/test_py39.py b/tests/test_py39.py index 3b1a0cb..1520da6 100644 --- a/tests/test_py39.py +++ b/tests/test_py39.py @@ -17,7 +17,7 @@ class Py39Test(unittest.TestCase): def test_no_third_party_imports(self): with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f: tree = ast.parse(f.read()) - allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "http", "sys", "time", "urllib", "datetime", "typing", "uuid"} + allowed = {"argparse", "base64", "json", "os", "signal", "ssl", "http", "sys", "time", "urllib", "datetime", "typing", "uuid", "fcntl"} for node in ast.walk(tree): if isinstance(node, ast.Import): for a in node.names: diff --git a/tests/test_runner.py b/tests/test_runner.py index 10c2a70..a6e8f5f 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -2,6 +2,7 @@ import os import tempfile import time import unittest +from datetime import datetime, timedelta, timezone from _common import SETTINGS, agent, jira_cfg, no_sleep from fake_hub import FakeHub @@ -50,6 +51,31 @@ class RunnerTest(unittest.TestCase): self.assertEqual(self.kinds(), [("claimed", 1), ("started", 1), ("finished", 1), ("acked", 1), ("claimed", 2), ("started", 2), ("finished", 2), ("acked", 2)]) self.assertEqual(self.sleeps, []) # пауз не было: last_call_at пуст, потом pause 0 + def test_restart_pace_from_old_finished(self): + at = (datetime.now(timezone.utc) - timedelta(seconds=10)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "finished", "command_id": 99, "status": "done", "result": None, "error": None, + "jira_calls": 1, "rate_limited": 0, "at": at}) + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=30) + self.runner.loop() + self.assertAlmostEqual(sum(self.sleeps), 20, delta=1.5) + + def test_future_journal_time_ignored(self): + at = (datetime.now(timezone.utc) + timedelta(hours=1)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=30) + self.runner.loop() + self.assertEqual(self.sleeps, []) # время из будущего — как пустой журнал + self.assertEqual(self.hub.results[1]["status"], "done") + + def test_restore_pace_keeps_existing_last_call(self): + at = (datetime.now(timezone.utc) - timedelta(seconds=100)).replace(microsecond=0).isoformat() + self.journal.append({"kind": "acked", "command_id": 99, "at": at}) + mark = time.monotonic() + self.runner.jira.last_call_at = mark + self.runner.restore_pace() + self.assertEqual(self.runner.jira.last_call_at, mark) + def test_pause_waits_from_last_jira_call(self): self.runner.jira.last_call_at = time.monotonic() self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}, pause=3) @@ -103,9 +129,15 @@ class RunnerTest(unittest.TestCase): self.runner.loop() self.assertEqual(self.hub.results[cid]["result"], {"changed": []}) # уже так в Jira — записи нет self.assertEqual(self.journal.unfinished_started(), []) - self.assertEqual(self.sleeps, []) # пауза при восстановлении не применяется + self.assertAlmostEqual(sum(self.sleeps), 99, delta=1) # пауза из хаба, от записи started (M-9) self.assertEqual([r[0] for r in self.fake.writes()], []) + def test_recover_does_not_journal_claimed_twice(self): + cid = self.claim_in_flight("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-20"}, pause=0) + self.journal.claimed(cid) + self.runner.loop() + self.assertEqual(self.kinds(), [("claimed", cid), ("started", cid), ("finished", cid), ("acked", cid)]) + def test_recover_ensure_issue_finds_duplicate(self): self.fake.add("AURORA-7", reporter="me", summary="Новая задача") before = len(self.fake.issues) @@ -159,7 +191,7 @@ class RunnerTest(unittest.TestCase): self.assertEqual(self.hub.results[1]["status"], "done") self.assertEqual(len(self.fake.writes()), 1) self.assertEqual(self.journal.unfinished_started(), []) - self.assertEqual(self.sleeps, []) # при восстановлении пауза не применяется + self.assertAlmostEqual(sum(self.sleeps), 5, delta=1) # пауза из хаба от записи claimed (M-9) def test_stop_between_claim_and_execute_is_recorded(self): self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"}) @@ -396,5 +428,21 @@ class RunnerTest(unittest.TestCase): self.assertEqual(seen["code"], 130) + def test_second_process_refused(self): + """I-1 финального ревью: второй run с тем же журналом не стартует — иначе довыдача хаба отдаст ему ту же команду.""" + from unittest import mock + held = agent.acquire_lock(self.journal.path) + self.assertIsNotNone(held) + try: + self.assertIsNone(agent.acquire_lock(self.journal.path)) + cfg = jira_cfg(self.fake, hub_url=self.hub.base_url, hub_token=self.hub.token) + with mock.patch.object(agent.Runner, "loop", lambda r: self.fail("цикл не должен стартовать")): + self.assertEqual(agent.cmd_run(cfg, self.journal.path), 1) + finally: + held.close() + again = agent.acquire_lock(self.journal.path) + self.assertIsNotNone(again) # после выхода первого — снова можно + again.close() + if __name__ == "__main__": unittest.main()