6b3109863d
wait_child запоминает подтверждение (любое состояние update.json, кроме trial этой версии, пока агент жив), а итоговое решение учитывает и случай, когда ok и следующая запись уложились между опросами. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
320 lines
16 KiB
Python
320 lines
16 KiB
Python
#!/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 = 2
|
||
EXIT_UPDATE = 75 # агент: «agent.py.new проверен, update.json = staged»
|
||
EXIT_LOCKED = 73 # агент: «журнал занят другим процессом» — не сбой версии (design 6)
|
||
TRIAL_SECONDS = 600 # новая версия должна записать ok за это время
|
||
KILL_GRACE = 60 # после SIGTERM до SIGKILL
|
||
BACKOFF_START, BACKOFF_MAX = 5, 300
|
||
HEALTHY_UPTIME = 600 # проработал столько — пауза перезапуска с начала
|
||
STOP_CODES = (0, 130, -signal.SIGINT) # выход агента после Ctrl-C (-SIGINT — Ctrl-C до установки его обработчиков)
|
||
|
||
|
||
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) -> bool:
|
||
"""Атомарная запись update.json. False — не записано (диск полон): запускатель не падает (design 5)."""
|
||
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"
|
||
try:
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False)
|
||
os.replace(tmp, path)
|
||
except OSError as e:
|
||
log("не удалось записать update.json: %s" % e)
|
||
return False
|
||
return True
|
||
|
||
|
||
def sha256_file(path: str) -> Optional[str]:
|
||
try:
|
||
with open(path, "rb") as f:
|
||
return hashlib.sha256(f.read()).hexdigest()
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
def remove(path: str) -> None:
|
||
try:
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
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.interrupted = False # был Ctrl-C: выход агента с 0/130 — остановка человеком, а не сбой версии
|
||
self.child = None
|
||
|
||
# --- сигналы ---------------------------------------------------------------
|
||
def on_sigint(self, signum, frame):
|
||
"""Ctrl-C в терминале приходит и агенту (общая группа): пока агент жив — решает он, запускатель только запоминает,
|
||
что человек остановил (I-1). Без агента (пауза) — выход."""
|
||
self.interrupted = True
|
||
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 reconcile(self) -> None:
|
||
"""Сверка update.json с файлами при старте (design 1): запускатель мог упасть посреди подмены или отката."""
|
||
for tmp in (self.new + ".tmp", self.prev + ".tmp", self.state_path + ".tmp"):
|
||
remove(tmp)
|
||
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")
|
||
st = read_state(self.state_path)
|
||
state, sha = (st or {}).get("state"), (st or {}).get("sha256")
|
||
if state == "staged":
|
||
if sha and sha256_file(self.agent) == sha: # подмена сделана, trial не записан (M-5)
|
||
remove(self.new)
|
||
write_state(self.state_path, "trial", st)
|
||
log("v%s уже на месте — пробный запуск" % st.get("to_version"))
|
||
elif sha and os.path.exists(self.new) and sha256_file(self.new) == sha:
|
||
log("найдена подготовленная версия — ставлю")
|
||
self.apply_update()
|
||
else:
|
||
remove(self.new)
|
||
write_state(self.state_path, "rejected", st, "подготовленный файл потерян")
|
||
log("подготовленная v%s потеряна — отвергнута" % st.get("to_version"))
|
||
return
|
||
remove(self.new)
|
||
if state == "trial" and (not sha or sha256_file(self.agent) != sha):
|
||
self.drop_stale_trial(st)
|
||
|
||
def in_trial(self) -> bool:
|
||
"""Пробный период для этого запуска: trial и agent.py — именно эта версия (design 2)."""
|
||
st = read_state(self.state_path)
|
||
if not (st and st.get("state") == "trial"):
|
||
return False
|
||
if not st.get("sha256") or sha256_file(self.agent) != st.get("sha256"):
|
||
self.drop_stale_trial(st)
|
||
return False
|
||
return True
|
||
|
||
def drop_stale_trial(self, st: dict) -> None:
|
||
"""agent.py не та версия, что в trial (прерван откат, ручная подмена): пробного периода нет, версию не ставить (M-3)."""
|
||
reason = "пробный период прерван: agent.py не v%s" % st.get("to_version")
|
||
write_state(self.state_path, "rolled_back", st, reason)
|
||
log(reason)
|
||
|
||
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")):
|
||
remove(self.new)
|
||
write_state(self.state_path, "rejected", st, "файл изменился после проверки")
|
||
log("обновление отвергнуто: файл изменился после проверки")
|
||
return
|
||
tmp = self.prev + ".tmp"
|
||
try:
|
||
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 существует на каждом шаге
|
||
except OSError as e:
|
||
self.fail_update(st, e)
|
||
return
|
||
if not write_state(self.state_path, "trial", st):
|
||
try:
|
||
os.replace(self.prev, self.agent) # без trial новая версия осталась бы без пробного периода
|
||
except OSError as e:
|
||
log("не удалось вернуть agent.py.prev: %s" % e)
|
||
self.fail_update(st, "update.json не записан")
|
||
return
|
||
log("v%s → v%s: пробный запуск" % (st.get("from_version"), st.get("to_version")))
|
||
|
||
def fail_update(self, st: dict, err) -> None:
|
||
"""Подмена не удалась (диск полон, права): текущая версия остаётся, эта — rejected навсегда (design 4)."""
|
||
remove(self.prev + ".tmp")
|
||
remove(self.new)
|
||
write_state(self.state_path, "rejected", st, "подмена не удалась: %s" % err)
|
||
log("подмена v%s не удалась: %s" % (st.get("to_version"), err))
|
||
|
||
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.
|
||
Подтверждение — любое состояние update.json, кроме trial этой же версии, пока агент жив: запускатель сам
|
||
update.json при живом агенте не пишет, значит это сделал подтвердивший агент (ok, затем rejected/staged для
|
||
следующей версии — I-1)."""
|
||
start = time.monotonic()
|
||
st0 = read_state(self.state_path) if trial else None
|
||
trial_sha = (st0 or {}).get("sha256")
|
||
confirmed = False
|
||
while True:
|
||
try:
|
||
return child.wait(1), False, confirmed
|
||
except subprocess.TimeoutExpired:
|
||
pass
|
||
if trial and not confirmed:
|
||
st = read_state(self.state_path)
|
||
if st and not (st.get("state") == "trial" and st.get("sha256") == trial_sha):
|
||
confirmed = True
|
||
elif time.monotonic() - start >= self.trial_seconds:
|
||
self.terminate(child)
|
||
return child.returncode, True, False
|
||
|
||
def pause(self, seconds: float) -> bool:
|
||
"""Пауза перед перезапуском; False — остановлены во время паузы."""
|
||
end = time.monotonic() + seconds
|
||
while not self.stopping:
|
||
left = end - time.monotonic() # одно чтение часов: остаток не уйдёт в минус между проверкой и sleep (M-1)
|
||
if left <= 0:
|
||
break
|
||
time.sleep(min(1.0, left))
|
||
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.reconcile()
|
||
code = 0
|
||
while True:
|
||
if self.stopping: # сигнал пришёл между запусками — нового агента не запускаем (M-2)
|
||
return code
|
||
trial = self.in_trial()
|
||
trial_sha = (read_state(self.state_path) or {}).get("sha256") if trial else None
|
||
started = time.monotonic()
|
||
self.interrupted = False # Ctrl-C относится к этому запуску агента
|
||
self.child = self.spawn()
|
||
if self.stopping: # SIGTERM между Popen и присваиванием child до агента не дошёл
|
||
self.child.send_signal(signal.SIGTERM)
|
||
code, timed_out, confirmed = self.wait_child(self.child, trial)
|
||
self.child = None
|
||
uptime = time.monotonic() - started
|
||
st = read_state(self.state_path)
|
||
ok = bool(st and st.get("state") == "ok")
|
||
if trial and not (ok or confirmed) and st and not (st.get("state") == "trial" and st.get("sha256") == trial_sha) \
|
||
and sha256_file(self.agent) == trial_sha:
|
||
confirmed = True # ok и следующая запись агента случились между опросами (I-1)
|
||
ok = ok or confirmed
|
||
reason = "агент упал (код %s)" % code
|
||
if timed_out and ok: # ответ claim пришёл в последнюю секунду (M-4, design 3)
|
||
log("подтверждение пришло в последний момент — отката нет")
|
||
self.backoff = BACKOFF_START
|
||
continue
|
||
if timed_out:
|
||
if self.rollback(st, "не вышла на связь за 10 мин"):
|
||
continue
|
||
elif self.stopping or (self.interrupted and code in STOP_CODES):
|
||
return code
|
||
elif code == EXIT_UPDATE:
|
||
self.apply_update()
|
||
self.backoff = BACKOFF_START
|
||
if self.interrupted: # Ctrl-C во время подготовки: подмена сделана, trial — при следующем старте
|
||
return 0
|
||
continue
|
||
elif code == EXIT_LOCKED: # журнал держит другой процесс — версия ни при чём (M-7, design 6)
|
||
reason = "журнал занят другим процессом (агент уже запущен)"
|
||
elif trial and not ok:
|
||
if self.rollback(st, "код %s через %d с" % (code, uptime)):
|
||
continue
|
||
if self.stopping:
|
||
return code
|
||
wait = self.next_backoff(uptime)
|
||
log("%s, перезапуск через %d с" % (reason, 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())
|