feat(agent): журнал на диске, восстановление по нему, prune
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o
This commit is contained in:
Andrey Epifancev
parent
24cfdcf25c
commit
7fb592e6de
83
src/agent.py
83
src/agent.py
@ -199,6 +199,85 @@ class HubClient(Http):
|
||||
return self.call("GET", "/agent/known-keys")["keys"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Журнал: только дозапись; started → finished → acked по id команды
|
||||
# ---------------------------------------------------------------------------
|
||||
class Journal:
|
||||
def __init__(self, path: str):
|
||||
self.path = path
|
||||
|
||||
def append(self, record: dict) -> None:
|
||||
record = dict(record)
|
||||
record.setdefault("at", now_iso())
|
||||
with open(self.path, "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||||
|
||||
def records(self) -> List[dict]:
|
||||
if not os.path.exists(self.path):
|
||||
return []
|
||||
out = []
|
||||
with open(self.path, encoding="utf-8") as f:
|
||||
for n, line in enumerate(f, 1):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
out.append(json.loads(line))
|
||||
except ValueError:
|
||||
log("журнал: строка %d не читается, пропускаю" % n)
|
||||
return out
|
||||
|
||||
def started(self, command_id) -> None:
|
||||
self.append({"kind": "started", "command_id": int(command_id)})
|
||||
|
||||
def finished(self, command_id, status, result, error, jira_calls, rate_limited) -> None:
|
||||
self.append({"kind": "finished", "command_id": int(command_id), "status": status, "result": result,
|
||||
"error": error, "jira_calls": int(jira_calls or 0), "rate_limited": int(rate_limited or 0)})
|
||||
|
||||
def acked(self, command_id) -> None:
|
||||
self.append({"kind": "acked", "command_id": int(command_id)})
|
||||
|
||||
def unacked_finished(self) -> List[dict]:
|
||||
last, acked = {}, set()
|
||||
for r in self.records():
|
||||
if r.get("kind") == "finished":
|
||||
last[r["command_id"]] = r
|
||||
acked.discard(r["command_id"])
|
||||
elif r.get("kind") == "acked":
|
||||
acked.add(r["command_id"])
|
||||
return [r for cid, r in last.items() if cid not in acked]
|
||||
|
||||
def unfinished_started(self) -> List[int]:
|
||||
started, finished = [], set()
|
||||
for r in self.records():
|
||||
if r.get("kind") == "started":
|
||||
started.append(r["command_id"])
|
||||
elif r.get("kind") in ("finished", "acked"):
|
||||
finished.add(r["command_id"])
|
||||
return [cid for cid in started if cid not in finished]
|
||||
|
||||
def prune(self, days: int = 30) -> int:
|
||||
"""Удалить записи команд, подтверждённых (acked) раньше, чем days дней назад. Неподтверждённые — остаются."""
|
||||
records = self.records()
|
||||
cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
|
||||
old = set()
|
||||
for r in records:
|
||||
if r.get("kind") == "acked":
|
||||
try:
|
||||
ts = datetime.fromisoformat(r.get("at", "")).timestamp()
|
||||
except ValueError:
|
||||
continue
|
||||
if ts < cutoff:
|
||||
old.add(r["command_id"])
|
||||
keep = [r for r in records if r.get("command_id") not in old]
|
||||
tmp = self.path + ".tmp"
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
for r in keep:
|
||||
f.write(json.dumps(r, ensure_ascii=False) + "\n")
|
||||
os.replace(tmp, self.path)
|
||||
return len(records) - len(keep)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -228,6 +307,10 @@ def main(argv=None) -> int:
|
||||
cfg = load_config(args.config)
|
||||
if args.cmd == "check":
|
||||
return cmd_check(cfg)
|
||||
if args.cmd == "prune":
|
||||
removed = Journal(args.journal).prune(30)
|
||||
log("журнал: удалено записей %d" % removed)
|
||||
return 0
|
||||
log("команда %s ещё не реализована" % args.cmd)
|
||||
return 2
|
||||
|
||||
|
||||
77
tests/test_journal.py
Normal file
77
tests/test_journal.py
Normal file
@ -0,0 +1,77 @@
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from _common import agent
|
||||
|
||||
|
||||
class JournalTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.dir = tempfile.TemporaryDirectory()
|
||||
self.path = os.path.join(self.dir.name, "journal.jsonl")
|
||||
self.j = agent.Journal(self.path)
|
||||
|
||||
def tearDown(self):
|
||||
self.dir.cleanup()
|
||||
|
||||
def test_empty(self):
|
||||
self.assertEqual(self.j.records(), [])
|
||||
self.assertEqual(self.j.unacked_finished(), [])
|
||||
self.assertEqual(self.j.unfinished_started(), [])
|
||||
|
||||
def test_lifecycle(self):
|
||||
self.j.started(1)
|
||||
self.assertEqual(self.j.unfinished_started(), [1])
|
||||
self.j.finished(1, "done", {"key": "A-1"}, None, 2, 0)
|
||||
self.assertEqual(self.j.unfinished_started(), [])
|
||||
pending = self.j.unacked_finished()
|
||||
self.assertEqual(len(pending), 1)
|
||||
self.assertEqual(pending[0]["command_id"], 1)
|
||||
self.assertEqual(pending[0]["result"], {"key": "A-1"})
|
||||
self.assertEqual(pending[0]["jira_calls"], 2)
|
||||
self.j.acked(1)
|
||||
self.assertEqual(self.j.unacked_finished(), [])
|
||||
self.j.started(2)
|
||||
self.j.started(3)
|
||||
self.j.finished(3, "failed", None, "HTTP 400", 1, 0)
|
||||
self.assertEqual(self.j.unfinished_started(), [2])
|
||||
self.assertEqual([r["command_id"] for r in self.j.unacked_finished()], [3])
|
||||
|
||||
def test_broken_line_is_skipped(self):
|
||||
self.j.started(1)
|
||||
with open(self.path, "a", encoding="utf-8") as f:
|
||||
f.write("{не json\n")
|
||||
self.j.finished(1, "done", None, None, 1, 0)
|
||||
out = io.StringIO()
|
||||
with contextlib.redirect_stdout(out):
|
||||
records = self.j.records()
|
||||
self.assertEqual(len(records), 2)
|
||||
self.assertIn("строка 2 не читается", out.getvalue())
|
||||
|
||||
def test_prune_keeps_unacked_and_recent(self):
|
||||
self.j.started(1)
|
||||
self.j.finished(1, "done", None, None, 1, 0)
|
||||
self.j.append({"kind": "acked", "command_id": 1, "at": "2020-01-01T00:00:00+00:00"})
|
||||
self.j.started(2)
|
||||
self.j.finished(2, "done", None, None, 1, 0)
|
||||
self.j.acked(2)
|
||||
self.j.started(3)
|
||||
removed = self.j.prune(days=30)
|
||||
self.assertEqual(removed, 3)
|
||||
ids = [(r["kind"], r["command_id"]) for r in self.j.records()]
|
||||
self.assertEqual(ids, [("started", 2), ("finished", 2), ("acked", 2), ("started", 3)])
|
||||
self.assertEqual(self.j.prune(days=30), 0)
|
||||
|
||||
def test_no_secrets_in_records(self):
|
||||
self.j.finished(1, "done", {"key": "A-1"}, None, 1, 0)
|
||||
with open(self.path, encoding="utf-8") as f:
|
||||
text = f.read()
|
||||
self.assertNotIn("password", text)
|
||||
self.assertEqual(json.loads(text.splitlines()[0])["kind"], "finished")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
x
Reference in New Issue
Block a user