jira-agent/docs/superpowers/plans/2026-10-14-agent-self-update-agent.md

agent-self-update (агент) Implementation Plan

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: агент v5 сам обновляется с хаба в паузе между командами, а запускатель run.py подменяет файл, проверяет новую версию пробным периодом и откатывает плохую.

Architecture: в agent.py — функции update.json (read_update_state/write_update_state), класс Updater (проверка публикации, скачивание, sha256, компиляция, selftest, состояние staged) и точка в Runner.loop после claim без команды; run выходит с кодом 75. Новый файл run.py (без импорта agent.py: тот меняется) запускает agent.py run дочерним процессом, по коду 75 подменяет файл, ведёт пробный период и откат, перезапускает после падения с паузой.

Tech Stack: Python 3.9.6 (/usr/bin/python3), только stdlib, unittest.

Spec: /Users/andrey.epifancev/projects/awp-ca/jira-hub/.wt/self-update/openspec/changes/agent-self-update/ (design.md решения 7–24, specs/agent/spec.md — «Самообновление агента», «Запускатель run.py»). Решения агента при исполнении — openspec/changes/agent-self-update/design.md этого репозитория. План хаба — jira-hub/docs/superpowers/plans/2026-10-14-agent-self-update-hub.md; агенту хаб в тестах не нужен (FakeHub).

Worktree: /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update, ветка feat/19-agent-self-update. Команды тестов — из этого каталога; git — git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update ….

Global Constraints

• Python 3.9, только стандартная библиотека; tests/test_py39.py проверяет agent.py и run.py. Без X | Y в аннотациях, без match.
• AGENT_VERSION = 5; LAUNCHER_VERSION = 1; код выхода «обновление готово» — 75.
• Файлы — в каталоге agent.py/run.py: agent.py.new, agent.py.prev, update.json, run.py.lock. Общая папка для кода не используется.
• update.json = {state, from_version, to_version, sha256, error, at}; состояния staged, trial, ok, rolled_back, rejected; запись — временный файл + os.replace.
• selftest: печатает jira-agent v<N>, затем как check; код 0 — всё доступно, 3 — что-то недоступно; блокировку журнала не берёт.
• Подготовка: таймаут selftest 120 с; повтор после временного сбоя — не раньше 300 с; подсказка без запускателя — не чаще 3600 с; ошибка rejected от selftest — последние 500 символов вывода.
• Запускатель: пробный период 600 с; SIGTERM, через 60 с SIGKILL; пауза после падения 5 → ×2 → не больше 300 с, сброс после 600 с работы.
• Тексты лога: «доступна v; для самообновления запускайте через run.py», «v готова, перезапуск через run.py», «обновление до v подтверждено», «run: откат v → v: <причина>», «run: агент упал (код ), перезапуск через с», «запускатель уже работает».
• Причины отката: «код через с», «не вышла на связь за 10 мин»; отказ при подмене: «файл изменился после проверки»; sha: «sha256 не сходится».
• Язык сообщений, комментариев, коммитов — русский. Коммит заканчивается строкой Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>.
• Весь набор: /usr/bin/python3 -W error::ResourceWarning -m unittest discover -s tests -t . — зелёным считается только весь набор.

Review Focus

• Ctrl-C во время selftest новой версии: SIGINT получает и дочерний selftest (общая группа), он падает с кодом 1 — это НЕ должно стать rejected; агент удаляет .new и выходит как обычно (тест в Task 2: test_stop_during_selftest_is_not_reject).
• Ответ GET /agent/release с другой версией/sha, чем в claim (публикация сменилась между запросами) — временный сбой, а не rejected (тест в Task 2: test_release_changed_between_requests_is_transient).
• Мусор в update.json (обрезан, не объект) — агент и запускатель не падают, считают «состояния нет» (тесты в Task 1: test_read_update_state_garbage, в Task 3: test_garbage_update_json_is_no_state).
• Падение запускателя между выходом агента с 75 и подменой — при следующем старте подмена доделывается (тест в Task 3: test_staged_at_start_is_applied).
• Ctrl-C в паузе перед перезапуском после падения — запускатель выходит, а не висит (тест в Task 3: test_sigint_without_child_stops).

---

Task 1: Состояние обновления, selftest, поля claim, версия 5

Files:

• Modify: src/agent.py (константы, read_update_state, write_update_state, HubClient.claim/release, класс Updater — только state/confirm, Runner — передача полей в claim и подтверждение, cmd_selftest, main)
• Modify: tests/fake_hub.py (поле release, маршрут /agent/release, agent_release в ответе claim)
• Modify: tests/test_py39.py (разрешённые импорты hashlib, re, subprocess)
• Create: tests/test_update.py
• Test: tests/test_runner.py, tests/test_clients.py

Interfaces:

• Produces (используют Task 2 и Task 4):

  • agent.AGENT_VERSION = 5, agent.EXIT_UPDATE = 75
  • agent.read_update_state(path: str) -> Optional[dict]
  • agent.write_update_state(path: str, state: str, from_version, to_version, sha256, error=None) -> dict
  • agent.HubClient.claim(agent_version, settings_version, last_jira_call_at, update_state=None, launcher_version=None) -> dict
  • agent.HubClient.release() -> dict
  • agent.Updater(hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True, config_path: Optional[str] = None); атрибуты state_path, new_path, launcher_version; методы state() -> Optional[dict], confirm() -> None
  • agent.Runner.updater (по умолчанию None), agent.Runner.exit_code (по умолчанию 0)
  • agent.cmd_selftest(cfg) -> int (0 или 3); main([... "selftest"])
  • FakeHub.release — None или {"version", "sha256", "content"}; FakeHub.release_requests: int

• Step 1: Write the failing tests

tests/test_update.py:

import json
import os
import tempfile
import unittest

from _common import agent


class UpdateStateTest(unittest.TestCase):
    def setUp(self):
        self.dir = tempfile.TemporaryDirectory()
        self.path = os.path.join(self.dir.name, "update.json")

    def tearDown(self):
        self.dir.cleanup()

    def test_write_and_read(self):
        d = agent.write_update_state(self.path, "staged", 5, 6, "abc")
        self.assertEqual(agent.read_update_state(self.path), d)
        self.assertEqual((d["state"], d["from_version"], d["to_version"], d["sha256"], d["error"]), ("staged", 5, 6, "abc", None))
        self.assertTrue(d["at"])
        self.assertEqual(os.listdir(self.dir.name), ["update.json"])        # временный файл не остался

    def test_read_missing(self):
        self.assertIsNone(agent.read_update_state(self.path))

    def test_read_update_state_garbage(self):
        for text in ('{"state": "tri', "[1, 2]", ""):
            with open(self.path, "w", encoding="utf-8") as f:
                f.write(text)
            self.assertIsNone(agent.read_update_state(self.path))

    def test_confirm_trial_of_this_version(self):
        u = agent.Updater(None, self.dir.name, 1)
        agent.write_update_state(u.state_path, "trial", 4, agent.AGENT_VERSION, "s")
        u.confirm()
        st = agent.read_update_state(u.state_path)
        self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("ok", 4, agent.AGENT_VERSION, "s"))

    def test_confirm_ignores_other_states(self):
        u = agent.Updater(None, self.dir.name, 1)
        u.confirm()
        self.assertIsNone(u.state())                                       # нет файла — не создаётся
        agent.write_update_state(u.state_path, "trial", 4, agent.AGENT_VERSION + 1, "s")
        u.confirm()
        self.assertEqual(u.state()["state"], "trial")                      # пробная другая версия — не наша
        agent.write_update_state(u.state_path, "rolled_back", 5, 6, "s", "код 1 через 3 с")
        u.confirm()
        self.assertEqual(u.state()["state"], "rolled_back")


if __name__ == "__main__":
    unittest.main()

tests/test_runner.py, в RunnerTest:

    def test_claim_sends_update_state_and_launcher(self):
        self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1)
        agent.write_update_state(self.runner.updater.state_path, "trial", 4, agent.AGENT_VERSION, "s")
        self.runner.loop()
        body = self.hub.claim_bodies[0]
        self.assertEqual(body["launcher_version"], 1)
        self.assertEqual(body["update_state"]["state"], "trial")
        self.assertEqual(agent.read_update_state(self.runner.updater.state_path)["state"], "ok")   # первый успешный claim
        self.assertEqual(self.hub.claim_bodies[0]["agent_version"], 5)

    def test_claim_without_updater_sends_nulls(self):
        self.runner.loop()
        self.assertIsNone(self.hub.claim_bodies[0]["update_state"])
        self.assertIsNone(self.hub.claim_bodies[0]["launcher_version"])

tests/test_clients.py — после test_check_command (в том же классе, где self.fake — FakeJira):

    def test_selftest_command(self):
        import contextlib
        import io
        import json
        import os
        import tempfile
        with tempfile.TemporaryDirectory() as d:
            path = os.path.join(d, "agent.json")
            with open(path, "w", encoding="utf-8") as f:
                json.dump(jira_cfg(self.fake), f)
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                # хаб недоступен (порт 1), Jira доступен — selftest печатает версию и возвращает 3, а не 1
                self.assertEqual(agent.main(["--config", path, "selftest"]), 3)
        self.assertIn("jira-agent v5", out.getvalue())
        self.assertIn("Jira: доступен", out.getvalue())

• Step 2: Run tests to verify they fail

Run: /usr/bin/python3 -m unittest tests.test_update tests.test_runner tests.test_clients -v 2>&1 | tail -20 Expected: FAIL/ERROR — AttributeError: module 'agent' has no attribute 'write_update_state', KeyError: 'update_state', invalid choice: 'selftest'.

• Step 3: Implement

tests/test_py39.py: в allowed добавить "hashlib", "re", "subprocess".

src/agent.py:

1. Docstring модуля — строки команд дополнить:

    python3 agent.py selftest  # самопроверка новой версии перед обновлением: версия + check; код 0 или 3

и абзац: «Самообновление — через запускатель run.py (спека agent, «Самообновление агента»).»

2. Импорты: добавить hashlib, re, subprocess (по алфавиту в существующем списке).

3. Константы:

AGENT_VERSION = 5
EXIT_UPDATE = 75              # «обновление готово»: run.py подменит agent.py на agent.py.new

4. После load_config — функции состояния обновления:

# ---------------------------------------------------------------------------
# Самообновление: update.json рядом с agent.py (спека agent «Самообновление агента», design 7)
# ---------------------------------------------------------------------------
def read_update_state(path: str) -> Optional[dict]:
    """Состояние обновления; нет файла или мусор — None."""
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        return None
    except (OSError, ValueError):
        log("update.json не читается — считаю, что состояния нет")
        return None
    return data if isinstance(data, dict) else None


def write_update_state(path: str, state: str, from_version, to_version, sha256, error=None) -> dict:
    data = {"state": state, "from_version": from_version, "to_version": to_version, "sha256": sha256,
            "error": error, "at": now_iso()}
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    os.replace(tmp, path)
    return data

5. HubClient:

    def claim(self, agent_version, settings_version, last_jira_call_at, update_state=None, launcher_version=None) -> dict:
        return self.call("POST", "/agent/claim", {"agent_version": agent_version, "settings_version": settings_version,
                                                  "last_jira_call_at": last_jira_call_at, "update_state": update_state,
                                                  "launcher_version": launcher_version}, timeout=max(self.timeout, 45))

    def release(self) -> dict:
        return self.call("GET", "/agent/release")

6. Класс Updater (перед class Runner), пока только состояние и подтверждение — check/prepare добавит Task 2:

class Updater:
    """Самообновление агента (спека agent «Самообновление агента», design 7–12). Работает только под run.py."""

    def __init__(self, hub, base_dir: str, launcher_version: Optional[int], enabled: bool = True,
                 config_path: Optional[str] = None):
        self.hub, self.dir, self.launcher_version, self.enabled = hub, base_dir, launcher_version, enabled
        self.config_path = config_path or os.path.join(base_dir, "agent.json")
        self.state_path = os.path.join(base_dir, "update.json")
        self.new_path = os.path.join(base_dir, "agent.py.new")
        self.agent_path = os.path.join(base_dir, "agent.py")

    def state(self) -> Optional[dict]:
        return read_update_state(self.state_path)

    def confirm(self) -> None:
        """Первый успешный claim новой версии: trial этой версии → ok (запускатель перестаёт ждать отката)."""
        st = self.state()
        if st and st.get("state") == "trial" and st.get("to_version") == AGENT_VERSION:
            write_update_state(self.state_path, "ok", st.get("from_version"), AGENT_VERSION, st.get("sha256"))
            log("обновление до v%d подтверждено" % AGENT_VERSION)

7. Runner.__init__: добавить

        self.updater = None           # type: Optional[Updater]
        self.exit_code = 0            # EXIT_UPDATE — выход для подмены версии

8. Runner.loop — вызов claim и подтверждение:

            try:
                upd = self.updater
                resp = self.hub.claim(AGENT_VERSION, self.settings.get("settings_version"), self.jira.last_call_iso,
                                      upd.state() if upd else None, upd.launcher_version if upd else None)
            except (ConnectionError, HttpError) as e:
                self.sleep(self.hub_wait(e, max(self.poll, 1)))
                continue
            if self.updater:
                self.updater.confirm()

9. cmd_selftest после cmd_check:

def cmd_selftest(cfg: dict) -> int:
    """Самопроверка версии перед обновлением: её запускает предыдущая версия. Контракт кодов MUST сохраняться:
    0 — хаб и Jira доступны, 3 — что-то недоступно (временно); любой другой код — файл негоден (design 10)."""
    log("jira-agent v%d" % AGENT_VERSION)
    return 0 if cmd_check(cfg) == 0 else 3

main: choices=["run", "check", "selftest", "prune"] и ветка if args.cmd == "selftest": return cmd_selftest(cfg).

tests/fake_hub.py: в __init__ — self.release = None и self.release_requests = 0; в route для /agent/claim оба ответа {"command": …, "settings_version": …} дополнить "agent_release": self._release_ref(); метод и маршрут:

    def _release_ref(self):
        r = self.release
        return {"version": r["version"], "sha256": r["sha256"]} if r else None

        if path == "/agent/release":
            self.release_requests += 1
            return (200, self.release) if self.release else (404, {"error": "no_release"})

(маршрут — до разбора /agent/commands/...).

• Step 4: Run tests to verify they pass

Run: /usr/bin/python3 -W error::ResourceWarning -m unittest discover -s tests -t . 2>&1 | tail -3 Expected: OK.

• Step 5: Commit

git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update add src/agent.py tests/
git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update commit -m "feat(agent): update.json, selftest, update_state и launcher_version в claim, версия 5 (#19)

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

---

Task 2: Обновление в паузе — Updater.check/prepare, выход 75

Files:

• Modify: src/agent.py (Updater.check, prepare, _run_selftest, _reject, _transient, _drop_new; Runner.loop; cmd_run/_run; main)
• Test: tests/test_update.py, tests/test_runner.py

Interfaces:

• Consumes: Task 1 (Updater, read_update_state, write_update_state, HubClient.release, FakeHub.release).

• Produces:

  • Updater.check(release: Optional[dict]) -> bool — True: agent.py.new готов, update.json = staged, run должен выйти с 75
  • Updater.run_selftest: Callable[[str], Tuple[Optional[int], str]] — подменяется в тестах; None — таймаут
  • Updater.clock (по умолчанию time.monotonic), Updater.should_stop: Callable[[], bool]
  • Updater.RETRY_SECONDS = 300, HINT_SECONDS = 3600, SELFTEST_TIMEOUT = 120
  • cmd_run(cfg, journal_path, config_path=None) -> int — 0 или EXIT_UPDATE

• Step 1: Write the failing tests — tests/test_update.py, новый класс:

import hashlib

from fake_hub import FakeHub
from _common import SETTINGS


def make_release(version, body="print('ok')\n"):
    content = "AGENT_VERSION = %d\n%s" % (version, body)
    return {"version": version, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), "content": content}


class UpdaterCheckTest(unittest.TestCase):
    def setUp(self):
        self.dir = tempfile.TemporaryDirectory()
        self.hub = FakeHub(SETTINGS)
        cfg = {"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5}
        self.u = agent.Updater(agent.HubClient(cfg), self.dir.name, 1)
        self.selftests = []
        self.code = 0
        self.u.run_selftest = lambda path: (self.selftests.append(path), (self.code, "вывод selftest"))[1]
        self.now = [1000.0]
        self.u.clock = lambda: self.now[0]

    def tearDown(self):
        self.hub.stop()
        self.dir.cleanup()

    def ref(self, rel):
        return {"version": rel["version"], "sha256": rel["sha256"]}

    def test_good_version_is_staged(self):
        rel = make_release(6)
        self.hub.release = rel
        self.assertTrue(self.u.check(self.ref(rel)))
        with open(self.u.new_path, encoding="utf-8") as f:
            self.assertEqual(f.read(), rel["content"])
        st = self.u.state()
        self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("staged", 5, 6, rel["sha256"]))
        self.assertEqual(self.selftests, [self.u.new_path])

    def test_not_newer_or_disabled_or_no_release(self):
        self.hub.release = make_release(5)
        self.assertFalse(self.u.check(self.ref(self.hub.release)))
        self.assertFalse(self.u.check(None))
        self.hub.release = make_release(6)
        self.u.enabled = False
        self.assertFalse(self.u.check(self.ref(self.hub.release)))
        self.assertEqual(self.hub.release_requests, 0)

    def test_sha_mismatch_rejected_once(self):
        rel = make_release(6)
        self.hub.release = dict(rel, content=rel["content"] + "# подмена\n")
        ref = self.ref(rel)
        self.hub.release["sha256"] = rel["sha256"]
        self.assertFalse(self.u.check(ref))
        st = self.u.state()
        self.assertEqual((st["state"], st["to_version"], st["error"]), ("rejected", 6, "sha256 не сходится"))
        self.assertFalse(os.path.exists(self.u.new_path))
        self.assertFalse(self.u.check(ref))
        self.assertEqual(self.hub.release_requests, 1)            # отвергнутую версию больше не качает

    def test_syntax_error_and_wrong_version_rejected(self):
        bad = make_release(6, body="def (:\n")
        self.hub.release = bad
        self.assertFalse(self.u.check(self.ref(bad)))
        self.assertIn("не компилируется", self.u.state()["error"])
        wrong = make_release(7)
        wrong = dict(wrong, version=8)
        self.hub.release = wrong
        self.assertFalse(self.u.check(self.ref(wrong)))
        self.assertIn("AGENT_VERSION", self.u.state()["error"])
        self.assertEqual(self.selftests, [])

    def test_selftest_bad_code_rejected(self):
        self.hub.release = make_release(6)
        self.code = 1
        self.assertFalse(self.u.check(self.ref(self.hub.release)))
        st = self.u.state()
        self.assertEqual(st["state"], "rejected")
        self.assertTrue(st["error"].startswith("selftest: код 1"))
        self.assertIn("вывод selftest", st["error"])
        self.assertFalse(os.path.exists(self.u.new_path))

    def test_selftest_unavailable_is_transient(self):
        self.hub.release = make_release(6)
        self.code = 3
        ref = self.ref(self.hub.release)
        self.assertFalse(self.u.check(ref))
        self.assertIsNone(self.u.state())                          # update.json не тронут
        self.assertFalse(os.path.exists(self.u.new_path))
        self.now[0] += 299
        self.assertFalse(self.u.check(ref))
        self.assertEqual(len(self.selftests), 1)                   # раньше 5 минут не повторяет
        self.now[0] += 2
        self.code = 0
        self.assertTrue(self.u.check(ref))

    def test_selftest_timeout_is_transient(self):
        self.hub.release = make_release(6)
        self.code = None
        self.assertFalse(self.u.check(self.ref(self.hub.release)))
        self.assertIsNone(self.u.state())

    def test_download_failure_is_transient(self):
        rel = make_release(6)                                      # в claim есть, а /agent/release — 404
        self.assertFalse(self.u.check(self.ref(rel)))
        self.assertIsNone(self.u.state())

    def test_release_changed_between_requests_is_transient(self):
        old, new = make_release(6), make_release(7)
        self.hub.release = new
        self.assertFalse(self.u.check(self.ref(old)))
        self.assertIsNone(self.u.state())
        self.assertEqual(self.selftests, [])

    def test_rolled_back_same_sha_not_retried(self):
        rel = make_release(6)
        self.hub.release = rel
        agent.write_update_state(self.u.state_path, "rolled_back", 5, 6, rel["sha256"], "код 1 через 3 с")
        self.assertFalse(self.u.check(self.ref(rel)))
        fixed = make_release(6, body="print('fixed')\n")
        self.hub.release = fixed
        self.assertTrue(self.u.check(self.ref(fixed)))             # тот же номер, другой sha — пробуем

    def test_without_launcher_only_hint(self):
        self.u.launcher_version = None
        self.hub.release = make_release(6)
        logs = []
        orig = agent.log
        agent.log = logs.append
        try:
            self.assertFalse(self.u.check(self.ref(self.hub.release)))
            self.assertFalse(self.u.check(self.ref(self.hub.release)))
            self.now[0] += 3600
            self.assertFalse(self.u.check(self.ref(self.hub.release)))
        finally:
            agent.log = orig
        self.assertEqual(logs.count("доступна v6; для самообновления запускайте через run.py"), 2)
        self.assertEqual(self.hub.release_requests, 0)

    def test_stop_during_selftest_is_not_reject(self):
        self.hub.release = make_release(6)
        stop = [False]
        self.u.should_stop = lambda: stop[0]

        def selftest(path):
            stop[0] = True                                         # Ctrl-C пришёл и selftest упал с 1
            return 1, "KeyboardInterrupt"
        self.u.run_selftest = selftest
        self.assertFalse(self.u.check(self.ref(self.hub.release)))
        self.assertIsNone(self.u.state())
        self.assertFalse(os.path.exists(self.u.new_path))

    def test_real_selftest_runs_new_file(self):
        u = agent.Updater(None, self.dir.name, 1, config_path=os.path.join(self.dir.name, "agent.json"))
        path = os.path.join(self.dir.name, "agent.py.new")
        with open(path, "w", encoding="utf-8") as f:
            f.write("import sys\nprint('args', sys.argv[1:])\nsys.exit(3)\n")
        code, out = u._run_selftest(path)
        self.assertEqual(code, 3)
        self.assertIn("'--config', '%s', 'selftest'" % os.path.join(self.dir.name, "agent.json"), out)

tests/test_runner.py, в RunnerTest:

    def test_idle_with_release_exits_for_update(self):
        rel = {"version": 6, "sha256": "s", "content": "x"}
        self.hub.release = rel
        self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1)
        self.runner.updater.check = lambda r: r == {"version": 6, "sha256": "s"}
        self.runner.idle_limit = 5
        self.runner.loop()
        self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE)
        self.assertEqual(self.hub.claims, 1)

    def test_command_first_then_update(self):
        self.hub.release = {"version": 6, "sha256": "s", "content": "x"}
        self.hub.enqueue("ensure_fields", {"key": "AURORA-1", "duedate": "2026-10-23"})
        self.runner.updater = agent.Updater(self.runner.hub, self.dir.name, 1)
        seen = []
        self.runner.updater.check = lambda r: (seen.append(r), True)[1]
        self.runner.loop()
        self.assertEqual(self.hub.results[1]["status"], "done")             # команда выполнена и подтверждена
        self.assertEqual(len(seen), 1)                                       # проверка — только после claim без команды
        self.assertEqual(self.runner.exit_code, agent.EXIT_UPDATE)

• Step 2: Run tests to verify they fail

Run: /usr/bin/python3 -m unittest tests.test_update tests.test_runner -v 2>&1 | tail -20 Expected: ERROR — AttributeError: 'Updater' object has no attribute 'check'.

• Step 3: Implement — в class Updater:

    RETRY_SECONDS = 300       # временный сбой подготовки — повтор не раньше (R-8)
    HINT_SECONDS = 3600       # подсказка «запускайте через run.py» — не чаще (R-10)
    SELFTEST_TIMEOUT = 120
    VERSION_RE = re.compile(r"^AGENT_VERSION = (\d+)[ \t]*\r?$", re.M)

в __init__ дописать:

        self.retry_after = {}         # (version, sha256) -> clock(), раньше которого не повторять
        self.hint_at = None           # type: Optional[float]
        self.clock = time.monotonic
        self.should_stop = lambda: False
        self.run_selftest = self._run_selftest

методы:

    def check(self, release) -> bool:
        """Вызывается после claim без команды. True — agent.py.new проверен, update.json = staged: run выходит с 75."""
        if not self.enabled or not isinstance(release, dict):
            return False
        try:
            version, sha = int(release.get("version")), str(release.get("sha256"))
        except (TypeError, ValueError):
            return False
        if version <= AGENT_VERSION:
            return False
        st = self.state()
        if st and st.get("state") in ("rejected", "rolled_back") and st.get("to_version") == version \
                and st.get("sha256") == sha:
            return False
        if self.launcher_version is None:
            now = self.clock()
            if self.hint_at is None or now - self.hint_at >= self.HINT_SECONDS:
                self.hint_at = now
                log("доступна v%d; для самообновления запускайте через run.py" % version)
            return False
        until = self.retry_after.get((version, sha))
        if until is not None and self.clock() < until:
            return False
        return self.prepare(version, sha)

    def prepare(self, version: int, sha: str) -> bool:
        log("доступна v%d — готовлю обновление" % version)
        try:
            rel = self.hub.release()
        except (ConnectionError, HttpError) as e:
            return self._transient(version, sha, "скачивание: %s" % e)
        content = rel.get("content") if isinstance(rel, dict) else None
        if not isinstance(content, str) or rel.get("version") != version or rel.get("sha256") != sha:
            return self._transient(version, sha, "публикация сменилась между запросами")
        data = content.encode("utf-8")
        if hashlib.sha256(data).hexdigest() != sha:
            return self._reject(version, sha, "sha256 не сходится")
        try:
            compile(content, "agent.py.new", "exec")
        except (SyntaxError, ValueError) as e:
            return self._reject(version, sha, "не компилируется: %s" % e)
        if self.VERSION_RE.findall(content) != [str(version)]:
            return self._reject(version, sha, "AGENT_VERSION в файле не %d" % version)
        tmp = self.new_path + ".tmp"
        with open(tmp, "wb") as f:
            f.write(data)
        try:
            os.chmod(tmp, os.stat(self.agent_path).st_mode & 0o777)
        except OSError:
            pass
        os.replace(tmp, self.new_path)
        if self.should_stop():
            self._drop_new()
            return False
        code, out = self.run_selftest(self.new_path)
        if self.should_stop():                     # Ctrl-C уронил и selftest — это не отказ версии
            self._drop_new()
            return False
        if code == 0:
            write_update_state(self.state_path, "staged", AGENT_VERSION, version, sha)
            log("v%d готова, перезапуск через run.py" % version)
            return True
        if code is None:
            return self._transient(version, sha, "selftest: таймаут %d с" % self.SELFTEST_TIMEOUT)
        if code == 3:
            return self._transient(version, sha, "selftest: хаб или Jira недоступны")
        return self._reject(version, sha, "selftest: код %s: %s" % (code, out[-500:]))

    def _run_selftest(self, path: str):
        try:
            p = subprocess.run([sys.executable, path, "--config", self.config_path, "selftest"], cwd=self.dir,
                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=self.SELFTEST_TIMEOUT)
        except subprocess.TimeoutExpired:
            return None, ""
        return p.returncode, p.stdout.decode("utf-8", "replace")

    def _drop_new(self) -> None:
        for path in (self.new_path, self.new_path + ".tmp"):
            try:
                os.remove(path)
            except FileNotFoundError:
                pass

    def _transient(self, version, sha, reason) -> bool:
        self._drop_new()
        self.retry_after[(version, sha)] = self.clock() + self.RETRY_SECONDS
        log("обновление до v%d отложено на %d мин: %s" % (version, self.RETRY_SECONDS // 60, reason))
        return False

    def _reject(self, version, sha, reason) -> bool:
        self._drop_new()
        write_update_state(self.state_path, "rejected", AGENT_VERSION, version, sha, reason)
        log("v%d отвергнута: %s" % (version, reason))
        return False

Runner.loop — после if cmd: … continue, до idle += 1:

            if self.updater and not self.stop and self.updater.check(resp.get("agent_release")):
                self.exit_code = EXIT_UPDATE
                return

cmd_run/_run — передать путь настроек и собрать Updater:

def cmd_run(cfg: dict, journal_path: str, config_path: Optional[str] = None) -> int:
    ...
        return _run(cfg, journal_path, config_path)

def launcher_version() -> Optional[int]:
    """Версия запускателя из окружения (run.py ставит JIRA_AGENT_LAUNCHER); без запускателя — None."""
    try:
        return int(os.environ["JIRA_AGENT_LAUNCHER"])
    except (KeyError, ValueError):
        return None

в _run после создания runner:

    hub = runner.hub
    runner.updater = Updater(hub, HERE, launcher_version(), cfg.get("auto_update", True) is not False, config_path)
    runner.updater.should_stop = lambda: runner.stop

конец _run:

    runner.loop()
    log("выход для обновления (код %d)" % EXIT_UPDATE if runner.exit_code == EXIT_UPDATE else "остановлен")
    return runner.exit_code

main: return cmd_run(cfg, args.journal, args.config).

• Step 4: Run tests to verify they pass

Run: /usr/bin/python3 -W error::ResourceWarning -m unittest discover -s tests -t . 2>&1 | tail -3 Expected: OK.

• Step 5: Commit

git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update add src/agent.py tests/
git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update commit -m "feat(agent): самообновление в паузе — скачивание, sha256, selftest, выход 75 (#19)

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

---

Task 3: Запускатель run.py

Files:

• Create: src/run.py
• Create: tests/test_launcher.py
• Modify: tests/test_py39.py (проверка run.py)

Interfaces:

• Consumes: контракт агента — код 75 + update.json = staged + agent.py.new; JIRA_AGENT_LAUNCHER; агент пишет ok (Task 1–2). run.py НЕ импортирует agent.py.

• Produces:

  • run.LAUNCHER_VERSION = 1, run.EXIT_UPDATE = 75
  • run.Launcher(base_dir: str, args: List[str]); атрибуты trial_seconds (600), kill_grace (60), pause (метод pause(seconds) -> bool, подменяется в тестах), stopping, child; методы run() -> int, on_sigint(signum, frame), on_sigterm(signum, frame)
  • run.acquire_lock(base_dir) -> Optional[file]; run.main(argv=None) -> int

• Step 1: Write the failing tests — tests/test_launcher.py:

import hashlib
import json
import os
import signal
import subprocess
import sys
import tempfile
import textwrap
import unittest

from _common import SRC

if SRC not in sys.path:
    sys.path.insert(0, SRC)
import run  # noqa: E402

# Поддельные агенты: маленькие скрипты, которые ведут себя как agent.py по контракту (коды выхода, update.json).
STATE_HELPERS = textwrap.dedent('''
    import json, os, sys
    HERE = os.path.dirname(os.path.abspath(__file__))
    STATE = os.path.join(HERE, "update.json")
    def state():
        try:
            return json.load(open(STATE, encoding="utf-8"))
        except Exception:
            return None
    def write(**kw):
        json.dump(kw, open(STATE, "w", encoding="utf-8"))
    def mark(name):
        open(os.path.join(HERE, "runs.log"), "a").write(name + "\\n")
''')


def fake(name, body):
    return STATE_HELPERS + "mark(%r)\n" % name + textwrap.dedent(body)


# v5 «нашёл» новую версию: кладёт agent.py.new из next.py, пишет staged и выходит с 75; после отката/отказа — выход 0.
V5_UPDATES = fake("v5", '''
    import hashlib
    st = state()
    if st is None:
        data = open(os.path.join(HERE, "next.py"), "rb").read()
        open(os.path.join(HERE, "agent.py.new"), "wb").write(data)
        write(state="staged", from_version=5, to_version=6, sha256=hashlib.sha256(data).hexdigest())
        sys.exit(75)
    sys.exit(0)
''')
# v5 с подменённым после проверки файлом: sha в staged не сходится с agent.py.new.
V5_BAD_STAGE = fake("v5", '''
    st = state()
    if st is None:
        open(os.path.join(HERE, "agent.py.new"), "w").write("мусор")
        write(state="staged", from_version=5, to_version=6, sha256="чужой")
        sys.exit(75)
    sys.exit(0)
''')
V6_OK_THEN_STOP = fake("v6", '''
    st = state()
    write(**dict(st, state="ok"))
    sys.exit(0)
''')
V6_CRASH = fake("v6", 'sys.exit(1)\n')
V6_HANG = fake("v6", 'import time\ntime.sleep(60)\n')


class LauncherTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.dir = self.tmp.name
        self.l = run.Launcher(self.dir, ["--journal", "j.jsonl"])
        self.pauses = []
        self.l.pause = lambda s: (self.pauses.append(s), True)[1]

    def tearDown(self):
        self.tmp.cleanup()

    def put(self, name, text):
        with open(os.path.join(self.dir, name), "w", encoding="utf-8") as f:
            f.write(text)

    def read(self, name):
        with open(os.path.join(self.dir, name), encoding="utf-8") as f:
            return f.read()

    def state(self):
        return json.loads(self.read("update.json"))

    def runs(self):
        return self.read("runs.log").split()

    def next_version(self, text):
        self.put("next.py", text)
        return hashlib.sha256(text.encode("utf-8")).hexdigest()

    def stage(self, new_text, to_version=6):
        self.put("agent.py.new", new_text)
        sha = hashlib.sha256(new_text.encode("utf-8")).hexdigest()
        self.put("update.json", json.dumps({"state": "staged", "from_version": 5, "to_version": to_version, "sha256": sha}))
        return sha

    def test_update_applied_and_confirmed(self):
        self.put("agent.py", V5_UPDATES)
        sha = self.next_version(V6_OK_THEN_STOP)
        self.assertEqual(self.l.run(), 0)
        self.assertEqual(self.read("agent.py"), V6_OK_THEN_STOP)
        self.assertEqual(self.read("agent.py.prev"), V5_UPDATES)
        self.assertFalse(os.path.exists(os.path.join(self.dir, "agent.py.new")))
        st = self.state()
        self.assertEqual((st["state"], st["from_version"], st["to_version"], st["sha256"]), ("ok", 5, 6, sha))
        self.assertEqual(self.runs(), ["v5", "v6"])

    def test_crash_in_trial_rolls_back(self):
        self.put("agent.py", V5_UPDATES)
        sha = self.next_version(V6_CRASH)
        self.assertEqual(self.l.run(), 0)
        self.assertEqual(self.read("agent.py"), V5_UPDATES)
        st = self.state()
        self.assertEqual((st["state"], st["to_version"], st["sha256"]), ("rolled_back", 6, sha))
        self.assertRegex(st["error"], r"^код 1 через \d+ с$")
        self.assertEqual(self.runs(), ["v5", "v6", "v5"])
        self.assertEqual(self.pauses, [])                                    # откат без паузы

    def test_hang_in_trial_rolls_back(self):
        self.l.trial_seconds = 1
        self.l.kill_grace = 5
        self.put("agent.py", V5_UPDATES)
        self.next_version(V6_HANG)
        self.assertEqual(self.l.run(), 0)
        self.assertEqual(self.read("agent.py"), V5_UPDATES)
        self.assertEqual(self.state()["error"], "не вышла на связь за 10 мин")

    def test_crash_after_ok_restarts_with_backoff(self):
        self.put("update.json", json.dumps({"state": "ok", "from_version": 5, "to_version": 6, "sha256": "s"}))
        self.put("agent.py", fake("v6", '''
            n = len(open(os.path.join(HERE, "runs.log")).read().split())
            sys.exit(1 if n <= 2 else 0)
        '''))
        self.assertEqual(self.l.run(), 0)
        self.assertEqual(self.pauses, [5, 10])
        self.assertEqual(self.state()["state"], "ok")                        # отката нет

    def test_backoff_caps_and_resets(self):
        self.assertEqual([self.l.next_backoff(10) for _ in range(8)], [5, 10, 20, 40, 80, 160, 300, 300])
        self.assertEqual(self.l.next_backoff(600), 5)                        # проработал 10 минут — с начала

    def test_75_without_staged_rejected(self):
        self.put("agent.py", V5_BAD_STAGE)
        self.assertEqual(self.l.run(), 0)
        self.assertEqual(self.read("agent.py"), V5_BAD_STAGE)
        self.assertEqual(self.runs(), ["v5", "v5"])
        self.assertFalse(os.path.exists(os.path.join(self.dir, "agent.py.new")))
        st = self.state()
        self.assertEqual((st["state"], st["error"]), ("rejected", "файл изменился после проверки"))

    def test_staged_at_start_is_applied(self):
        self.put("agent.py", fake("v5", "sys.exit(0)\n"))
        self.stage(V6_OK_THEN_STOP)
        self.assertEqual(self.l.run(), 0)
        self.assertEqual(self.runs(), ["v6"])
        self.assertEqual(self.state()["state"], "ok")

    def test_stray_new_removed_at_start(self):
        self.put("agent.py", fake("v5", "sys.exit(0)\n"))
        self.put("agent.py.new", "остаток")
        self.assertEqual(self.l.run(), 0)
        self.assertFalse(os.path.exists(os.path.join(self.dir, "agent.py.new")))

    def test_missing_agent_restored_from_prev(self):
        self.put("agent.py.prev", fake("v5", "sys.exit(0)\n"))
        self.assertEqual(self.l.run(), 0)
        self.assertEqual(self.runs(), ["v5"])

    def test_garbage_update_json_is_no_state(self):
        self.put("update.json", '{"state": "tri')
        self.put("agent.py", fake("v5", "sys.exit(0)\n"))
        self.assertEqual(self.l.run(), 0)

    def test_env_and_args_passed(self):
        self.put("agent.py", fake("v5", '''
            open(os.path.join(HERE, "env.txt"), "w").write(os.environ.get("JIRA_AGENT_LAUNCHER", "") + " " + " ".join(sys.argv[1:]))
            sys.exit(130)
        '''))
        self.assertEqual(self.l.run(), 130)
        self.assertEqual(self.read("env.txt"), "1 run --journal j.jsonl")

    def test_sigint_without_child_stops(self):
        self.l.on_sigint(signal.SIGINT, None)
        self.assertTrue(self.l.stopping)

    def test_sigint_with_child_ignored(self):
        self.l.child = object()
        self.l.on_sigint(signal.SIGINT, None)
        self.assertFalse(self.l.stopping)

    def test_second_launcher_refused(self):
        with open(os.path.join(SRC, "run.py"), encoding="utf-8") as f:
            self.put("run.py", f.read())
        self.put("agent.py", fake("v5", "sys.exit(0)\n"))
        lock = run.acquire_lock(self.dir)
        try:
            p = subprocess.run([sys.executable, os.path.join(self.dir, "run.py")], stdout=subprocess.PIPE,
                               stderr=subprocess.STDOUT, timeout=30)
        finally:
            lock.close()
        self.assertEqual(p.returncode, 1)
        self.assertIn("запускатель уже работает", p.stdout.decode("utf-8"))
        self.assertFalse(os.path.exists(os.path.join(self.dir, "runs.log")))


if __name__ == "__main__":
    unittest.main()

tests/test_py39.py — оба теста по циклу for name in ("agent.py", "run.py") (через self.subTest(name)), allowed для run.py тот же набор плюс subprocess, hashlib уже есть с Task 1.

• Step 2: Run tests to verify they fail

Run: /usr/bin/python3 -m unittest tests.test_launcher -v 2>&1 | tail -5 Expected: ERROR — ModuleNotFoundError: No module named 'run'.

• Step 3: Implement — src/run.py:

#!/usr/bin/env python3
"""run.py — запускатель jira-agent: запускает agent.py дочерним процессом, ставит новую версию после самообновления
агента, откатывает её, если она упала или не вышла на связь, и поднимает агента после падения.
Только стандартная библиотека, Python 3.9. agent.py не импортирует: тот меняется, запускатель — нет.
Спека: jira-hub/openspec/specs/agent/spec.md, «Запускатель run.py».

    python3 run.py [--config agent.json] [--journal journal.jsonl]   # аргументы уходят в agent.py run

Обновляется вручную (редко): остановить, заменить файл, запустить.
"""
import fcntl
import hashlib
import json
import os
import signal
import subprocess
import sys
import time
from datetime import datetime, timezone
from typing import List, Optional

LAUNCHER_VERSION = 1
EXIT_UPDATE = 75                 # агент: «agent.py.new проверен, update.json = staged»
TRIAL_SECONDS = 600              # новая версия должна записать ok за это время
KILL_GRACE = 60                  # после SIGTERM до SIGKILL
BACKOFF_START, BACKOFF_MAX = 5, 300
HEALTHY_UPTIME = 600             # проработал столько — пауза перезапуска с начала


def log(msg: str) -> None:
    print("[%s] run: %s" % (datetime.now().strftime("%H:%M:%S"), msg), flush=True)


def read_state(path: str) -> Optional[dict]:
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
    except (OSError, ValueError):
        return None
    return data if isinstance(data, dict) else None


def write_state(path: str, state: str, base: Optional[dict], error=None) -> None:
    base = base or {}
    data = {"state": state, "from_version": base.get("from_version"), "to_version": base.get("to_version"),
            "sha256": base.get("sha256"), "error": error,
            "at": datetime.now(timezone.utc).replace(microsecond=0).isoformat()}
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    os.replace(tmp, path)


def sha256_file(path: str) -> str:
    with open(path, "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()


def acquire_lock(base_dir: str):
    """Один запускатель на каталог. None — уже занято."""
    f = open(os.path.join(base_dir, "run.py.lock"), "a")
    try:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        f.close()
        return None
    return f


class Launcher:
    def __init__(self, base_dir: str, args: List[str]):
        self.dir, self.args = base_dir, list(args)
        self.agent = os.path.join(base_dir, "agent.py")
        self.new = os.path.join(base_dir, "agent.py.new")
        self.prev = os.path.join(base_dir, "agent.py.prev")
        self.state_path = os.path.join(base_dir, "update.json")
        self.trial_seconds, self.kill_grace = TRIAL_SECONDS, KILL_GRACE
        self.backoff = BACKOFF_START
        self.stopping = False
        self.child = None

    # --- сигналы ---------------------------------------------------------------
    def on_sigint(self, signum, frame):
        """Ctrl-C в терминале приходит и агенту (общая группа): пока агент жив — решает он. Без агента (пауза) — выход."""
        if self.child is None:
            self.stopping = True

    def on_sigterm(self, signum, frame):
        self.stopping = True
        child = self.child
        if child is not None and hasattr(child, "send_signal"):
            child.send_signal(signal.SIGTERM)

    # --- файлы версии ----------------------------------------------------------
    def recover_files(self) -> None:
        if not os.path.exists(self.agent) and os.path.exists(self.prev):
            os.replace(self.prev, self.agent)
            log("agent.py не найден — восстановлен из agent.py.prev")
        if os.path.exists(self.new):
            st = read_state(self.state_path)
            if st and st.get("state") == "staged":
                log("найдена подготовленная версия — ставлю")
                self.apply_update()
            else:
                os.remove(self.new)

    def apply_update(self) -> None:
        st = read_state(self.state_path)
        if not (st and st.get("state") == "staged" and os.path.exists(self.new) and sha256_file(self.new) == st.get("sha256")):
            if os.path.exists(self.new):
                os.remove(self.new)
            write_state(self.state_path, "rejected", st, "файл изменился после проверки")
            log("обновление отвергнуто: файл изменился после проверки")
            return
        tmp = self.prev + ".tmp"
        with open(self.agent, "rb") as src, open(tmp, "wb") as dst:
            dst.write(src.read())
        os.replace(tmp, self.prev)
        os.replace(self.new, self.agent)                 # agent.py существует на каждом шаге
        write_state(self.state_path, "trial", st)
        log("v%s → v%s: пробный запуск" % (st.get("from_version"), st.get("to_version")))

    def rollback(self, st: Optional[dict], reason: str) -> bool:
        if not os.path.exists(self.prev):
            log("откат невозможен: нет agent.py.prev")
            return False
        os.replace(self.prev, self.agent)
        write_state(self.state_path, "rolled_back", st, reason)
        log("откат v%s → v%s: %s" % ((st or {}).get("to_version"), (st or {}).get("from_version"), reason))
        return True

    # --- процесс агента --------------------------------------------------------
    def spawn(self):
        env = dict(os.environ, JIRA_AGENT_LAUNCHER=str(LAUNCHER_VERSION))
        return subprocess.Popen([sys.executable, "-u", self.agent, "run"] + self.args, env=env, cwd=self.dir)

    def terminate(self, child) -> None:
        child.send_signal(signal.SIGTERM)
        try:
            child.wait(self.kill_grace)
        except subprocess.TimeoutExpired:
            child.kill()
            child.wait()

    def wait_child(self, child, trial: bool):
        """Код выхода и признак «пробный период истёк». В пробный период ждём ok не дольше trial_seconds."""
        start = time.monotonic()
        while True:
            try:
                return child.wait(1), False
            except subprocess.TimeoutExpired:
                pass
            if trial:
                st = read_state(self.state_path)
                if st and st.get("state") == "ok":
                    trial = False
                elif time.monotonic() - start >= self.trial_seconds:
                    self.terminate(child)
                    return child.returncode, True

    def pause(self, seconds: float) -> bool:
        """Пауза перед перезапуском; False — остановлены во время паузы."""
        end = time.monotonic() + seconds
        while not self.stopping and time.monotonic() < end:
            time.sleep(min(1.0, end - time.monotonic()))
        return not self.stopping

    def next_backoff(self, uptime: float) -> int:
        if uptime >= HEALTHY_UPTIME:
            self.backoff = BACKOFF_START
        wait = self.backoff
        self.backoff = min(self.backoff * 2, BACKOFF_MAX)
        return wait

    def run(self) -> int:
        self.recover_files()
        while True:
            st = read_state(self.state_path)
            trial = bool(st and st.get("state") == "trial")
            started = time.monotonic()
            self.child = self.spawn()
            code, timed_out = self.wait_child(self.child, trial)
            self.child = None
            uptime = time.monotonic() - started
            st = read_state(self.state_path)
            if timed_out:
                if self.rollback(st, "не вышла на связь за 10 мин"):
                    continue
            elif code in (0, 130) or self.stopping:
                return code
            elif code == EXIT_UPDATE:
                self.apply_update()
                self.backoff = BACKOFF_START
                continue
            elif trial and not (st and st.get("state") == "ok"):
                if self.rollback(st, "код %s через %d с" % (code, uptime)):
                    continue
            if self.stopping:
                return code
            wait = self.next_backoff(uptime)
            log("агент упал (код %s), перезапуск через %d с" % (code, wait))
            if not self.pause(wait):
                return code


def main(argv=None) -> int:
    base = os.path.dirname(os.path.abspath(__file__))
    lock = acquire_lock(base)
    if lock is None:
        log("запускатель уже работает в %s — второй не стартует" % base)
        return 1
    try:
        launcher = Launcher(base, sys.argv[1:] if argv is None else argv)
        signal.signal(signal.SIGINT, launcher.on_sigint)
        signal.signal(signal.SIGTERM, launcher.on_sigterm)
        log("запускатель v%d: агент %s" % (LAUNCHER_VERSION, launcher.agent))
        return launcher.run()
    finally:
        lock.close()


if __name__ == "__main__":
    sys.exit(main())

Примечание к test_hang_in_trial_rolls_back: таймаут пробного периода в тесте 1 с, а причина в update.json — всегда «не вышла на связь за 10 мин» (текст спеки, не зависит от trial_seconds). Поддельный v6 без обработчика SIGTERM умирает с кодом -15 — это не 0/130, но timed_out проверяется первым.

• Step 4: Run tests to verify they pass

Run: /usr/bin/python3 -W error::ResourceWarning -m unittest discover -s tests -t . 2>&1 | tail -3 Expected: OK. Если ResourceWarning на Popen — в run() после wait_child объект Popen уже дождан; на open(...) в поддельных агентах предупреждения в дочернем процессе не ловятся.

• Step 5: Commit

git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update add src/run.py tests/test_launcher.py tests/test_py39.py
git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update commit -m "feat(agent): запускатель run.py — подмена версии, пробный период, откат, перезапуск (#19)

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

---

Task 4: Сквозной тест run.py + настоящий agent.py и документация

Files:

• Create: tests/test_self_update_e2e.py
• Modify: src/agent.example.json ("auto_update": true)
• Modify: README.md («Доставка на виртуалку», «Запуск», «Автозапуск через launchd», «Обновление агента», раздел «Самообновление»)
• Modify: ARCHITECTURE.md (раздел Updater, run.py, границы)
• Modify: CLAUDE.md (архитектура, команды, текущий change, пункт публикации в закрытии)

Interfaces:

• Consumes: всё из Task 1–3; FakeHub.release, FakeJira (/rest/api/2/myself).

• Step 1: Write the failing test — tests/test_self_update_e2e.py:

import hashlib
import json
import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import unittest

from _common import SETTINGS, SRC
from fake_hub import FakeHub
from fake_jira import FakeJira


def wait_for(pred, timeout=30):
    end = time.monotonic() + timeout
    while time.monotonic() < end:
        if pred():
            return True
        time.sleep(0.1)
    return False


class SelfUpdateE2E(unittest.TestCase):
    """run.py + настоящий agent.py: v5 видит публикацию v6, ставит её, v6 подтверждает; Ctrl-C группе — выход 0."""

    def setUp(self):
        self.jira = FakeJira()
        self.hub = FakeHub(SETTINGS)
        self.tmp = tempfile.TemporaryDirectory()
        d = self.tmp.name
        shutil.copy(os.path.join(SRC, "run.py"), d)
        shutil.copy(os.path.join(SRC, "agent.py"), d)
        with open(os.path.join(d, "agent.json"), "w", encoding="utf-8") as f:
            json.dump({"hub_url": self.hub.base_url, "hub_token": self.hub.token, "timeout": 5, "poll_seconds": 0.2,
                       "jira": {"base_url": self.jira.base_url,
                                "auth": {"type": "basic", "user": "me", "password": "secret"}}}, f)
        with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f:
            src = f.read()
        content = src.replace("AGENT_VERSION = 5\n", "AGENT_VERSION = 6\n", 1)
        self.assertNotEqual(content, src)
        self.hub.release = {"version": 6, "sha256": hashlib.sha256(content.encode("utf-8")).hexdigest(), "content": content}
        self.content = content
        self.log = open(os.path.join(d, "out.log"), "wb")
        self.proc = subprocess.Popen([sys.executable, os.path.join(d, "run.py")], stdout=self.log,
                                     stderr=subprocess.STDOUT, start_new_session=True)

    def tearDown(self):
        if self.proc.poll() is None:
            os.killpg(self.proc.pid, signal.SIGKILL)
            self.proc.wait()
        self.log.close()
        self.hub.stop()
        self.jira.stop()
        self.tmp.cleanup()

    def state(self):
        try:
            with open(os.path.join(self.tmp.name, "update.json"), encoding="utf-8") as f:
                return json.load(f)
        except (OSError, ValueError):
            return None

    def test_update_to_v6_and_stop_by_ctrl_c(self):
        ok = wait_for(lambda: (self.state() or {}).get("state") == "ok", 60)
        with open(os.path.join(self.tmp.name, "out.log"), encoding="utf-8", errors="replace") as f:
            out = f.read()
        self.assertTrue(ok, out)
        with open(os.path.join(self.tmp.name, "agent.py"), encoding="utf-8") as f:
            self.assertEqual(f.read(), self.content)
        self.assertTrue(os.path.exists(os.path.join(self.tmp.name, "agent.py.prev")))
        self.assertTrue(wait_for(lambda: any(b.get("agent_version") == 6 for b in self.hub.claim_bodies)))
        last = [b for b in self.hub.claim_bodies if b.get("agent_version") == 6][-1]
        self.assertEqual(last["launcher_version"], 1)
        os.killpg(self.proc.pid, signal.SIGINT)                               # Ctrl-C в терминале — всей группе
        self.assertEqual(self.proc.wait(30), 0)

• Step 2: Run test to verify it fails or passes

Run: /usr/bin/python3 -m unittest tests.test_self_update_e2e -v 2>&1 | tail -15 Expected: PASS, если Task 1–3 выполнены верно (это интеграционный тест, а не TDD новой логики). Если FAIL — разбирать вывод out.log из сообщения теста (superpowers:systematic-debugging), чинить в agent.py/run.py с отдельным тестом на найденный случай в test_update.py/test_launcher.py.

• Step 3: Documentation

src/agent.example.json — после "poll_seconds": 5 строка "auto_update": true.

README.md:

• первая строка описания: «Один файл src/agent.py и запускатель src/run.py, только стандартная библиотека, Python 3.9.»
• «Доставка на виртуалку», шаг 1: копировать src/agent.py, src/run.py и src/agent.example.json в локальный каталог виртуалки (~/jira-agent-bundle), не запускать из общей папки; в список настроек — auto_update (по умолчанию true; false — не обновляться самому).
• «Проверка»: добавить абзац про python3 agent.py selftest («то же, что check, плюс строка jira-agent v<N>; код 3 — что-то недоступно; её вызывает предыдущая версия перед обновлением»).
• «Запуск»: python3 run.py в терминале, nohup python3 -u run.py > agent.log 2>&1 & в фоне; Ctrl-C — как раньше (первый — доделать команду, второй — выйти сразу); запуск python3 agent.py без запускателя возможен, но тогда агент себя не обновляет и раз в час пишет «доступна v…».
• «Автозапуск через launchd»: в ProgramArguments — /Users/USER/jira-agent/run.py.
• новый раздел «Самообновление» (после «Запуск»):

## Самообновление

С версии 5 агент обновляется сам. После `jira-hub agent publish` на Mac mini хаб сообщает новую версию в ответе на
`claim`. Агент в паузе между командами (очередь пуста, отчёты подтверждены) скачивает файл по `GET /agent/release`,
сверяет sha256, компилирует, проверяет `AGENT_VERSION`, кладёт рядом `agent.py.new`, запускает его `selftest` и выходит
с кодом 75. `run.py` копирует текущий `agent.py` в `agent.py.prev`, ставит новый и запускает его в пробном режиме:
версия подтверждена, когда она получила первый ответ хаба (`update.json` → `ok`). Если новая версия упала или не вышла
на связь за 10 минут — `run.py` возвращает `agent.py.prev` и пишет `rolled_back`; эту версию агент больше не ставит,
хаб показывает откат в `agent_status`. Код в общую папку не попадает: файлы пишутся только в каталог `run.py`.

Файлы рядом с `agent.py`: `update.json` (состояние: `staged`, `trial`, `ok`, `rolled_back`, `rejected`),
`agent.py.prev` (прежняя версия), `run.py.lock`. Не удалять во время работы.

Выключить: `"auto_update": false` в `agent.json`. `run.py` обновляется вручную (меняется редко).

• «Обновление агента» переписать: «С версии 5 — само (раздел «Самообновление»). Вручную — только run.py или переход с v4: остановить старый процесс, скопировать agent.py и run.py в каталог агента, python3 agent.py check, запустить python3 run.py. agent.json, journal.jsonl, update.json не трогать.» (абзац о двух процессах и блокировке журнала сохранить).
• «Версия» в «Правила работы»: «сейчас 5 — самообновление через run.py».

ARCHITECTURE.md:

• первая строка: агент — src/agent.py и запускатель src/run.py.
• строка таблицы перед «Цикл»:

| Самообновление | `read_update_state`, `write_update_state`, `Updater`, `launcher_version` | `update.json`; после `claim` без команды — проверка `agent_release`, скачивание, sha256, компиляция, `selftest`, `staged`; подтверждение `ok` новой версией; без запускателя — только подсказка в лог | `HubClient`, `subprocess` |

• у «Цикл» — «после claim без команды спрашивает Updater; готова версия — выход с EXIT_UPDATE (75)»; у CLI — selftest.
• новый абзац «## Запускатель src/run.py»: не импортирует agent.py; Launcher — дочерний agent.py run, коды 0/130 → выход, 75 → apply_update (.prev, os.replace, trial), падение в trial или 10 мин без ok → rollback, иначе пауза 5…300 с; блокировка run.py.lock; SIGINT при живом агенте игнорирует, SIGTERM пересылает.
• «Границы»: «Код новой версии приходит только через API хаба и пишется в каталог агента; общая папка для кода не используется.»

CLAUDE.md:

• «Архитектура»: src/run.py — запускатель (не импортирует agent.py); раздел Updater между Runner и CLI; CLI — run|check|selftest|prune.
• «Команды»: /usr/bin/python3 src/run.py — запуск через запускатель (самообновление); /usr/bin/python3 src/agent.py selftest.
• «Тесты»: test_update.py (самообновление агента), test_launcher.py (запускатель на поддельных агентах), test_self_update_e2e.py (run.py + настоящий agent.py + фейки).
• «Первая подсистема», последний абзац: текущий change — ../jira-hub/openspec/changes/agent-self-update/, план — docs/superpowers/plans/2026-10-14-agent-self-update-agent.md.
• новый раздел:

## Выкатка новой версии
Самообновление (с v5): после merge PR агента в main — поднять `AGENT_VERSION` (если не поднят в PR) и опубликовать:
`uv run --project ../jira-hub jira-hub agent publish src/agent.py` (из основного checkout `jira-agent` на main).
Через несколько минут MCP `agent_status` должен показать «обновление: ok v<старая> → v<новая>»; откат — «откат с v<N>…»
(причина — в логе агента на виртуалке). Это пункт чеклиста «закрывай» для PR агента. `run.py` публикацией не
обновляется — только вручную через общую папку.

• Step 4: Run all tests

Run: /usr/bin/python3 -W error::ResourceWarning -m unittest discover -s tests -t . 2>&1 | tail -3 Expected: OK.

• Step 5: Commit

git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update add tests/test_self_update_e2e.py src/agent.example.json README.md ARCHITECTURE.md CLAUDE.md
git -C /Users/andrey.epifancev/projects/awp-ca/jira-agent/.wt/self-update commit -m "test(agent): сквозной тест самообновления; README, ARCHITECTURE, CLAUDE.md — run.py и публикация (#19)

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