feat(agent): запускатель run.py — подмена версии, пробный период, откат, перезапуск (#19)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
3aa8823929
commit
7512acb083
223
src/run.py
Normal file
223
src/run.py
Normal file
@ -0,0 +1,223 @@
|
||||
#!/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())
|
||||
212
tests/test_launcher.py
Normal file
212
tests/test_launcher.py
Normal file
@ -0,0 +1,212 @@
|
||||
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()
|
||||
@ -10,21 +10,24 @@ from _common import SRC
|
||||
|
||||
|
||||
class Py39Test(unittest.TestCase):
|
||||
def test_agent_parses_as_python_39(self):
|
||||
with open(os.path.join(SRC, "agent.py"), encoding="utf-8") as f:
|
||||
ast.parse(f.read(), feature_version=(3, 9))
|
||||
FILES = ("agent.py", "run.py")
|
||||
|
||||
def test_parses_as_python_39(self):
|
||||
for name in self.FILES:
|
||||
with self.subTest(name), open(os.path.join(SRC, name), encoding="utf-8") as f:
|
||||
ast.parse(f.read(), feature_version=(3, 9))
|
||||
|
||||
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", "fcntl", "hashlib", "re", "subprocess"}
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for a in node.names:
|
||||
self.assertIn(a.name.split(".")[0], allowed, a.name)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
self.assertIn((node.module or "").split(".")[0], allowed, node.module)
|
||||
|
||||
for name in self.FILES:
|
||||
with self.subTest(name), open(os.path.join(SRC, name), encoding="utf-8") as f:
|
||||
tree = ast.parse(f.read())
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Import):
|
||||
for a in node.names:
|
||||
self.assertIn(a.name.split(".")[0], allowed, "%s: %s" % (name, a.name))
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
self.assertIn((node.module or "").split(".")[0], allowed, "%s: %s" % (name, node.module))
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user