fix(agent): журнал — дозапись после оборванной строки, порядковый учёт started, безопасный 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
7fb592e6de
commit
6ffcf2b92a
120
src/agent.py
120
src/agent.py
@ -206,26 +206,54 @@ class Journal:
|
||||
def __init__(self, path: str):
|
||||
self.path = path
|
||||
|
||||
def _torn_tail(self) -> bool:
|
||||
"""Файл не пуст и не заканчивается переводом строки (оборванная запись)."""
|
||||
try:
|
||||
if os.path.getsize(self.path) == 0:
|
||||
return False
|
||||
except FileNotFoundError:
|
||||
return False
|
||||
with open(self.path, "rb") as f:
|
||||
f.seek(-1, os.SEEK_END)
|
||||
return f.read(1) != b"\n"
|
||||
|
||||
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")
|
||||
data = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8")
|
||||
if self._torn_tail():
|
||||
data = b"\n" + data
|
||||
fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600)
|
||||
with os.fdopen(fd, "ab") as f:
|
||||
f.write(data)
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
|
||||
def records(self) -> List[dict]:
|
||||
def _scan(self):
|
||||
"""Возвращает (годные записи, номера битых строк). Годная — dict с command_id."""
|
||||
good, bad = [], []
|
||||
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()
|
||||
return good, bad
|
||||
with open(self.path, "rb") as f:
|
||||
for n, raw in enumerate(f, 1):
|
||||
line = raw.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
out.append(json.loads(line))
|
||||
r = json.loads(line.decode("utf-8"))
|
||||
except ValueError:
|
||||
log("журнал: строка %d не читается, пропускаю" % n)
|
||||
return out
|
||||
r = None
|
||||
if isinstance(r, dict) and "command_id" in r:
|
||||
good.append(r)
|
||||
else:
|
||||
bad.append(n)
|
||||
return good, bad
|
||||
|
||||
def records(self) -> List[dict]:
|
||||
good, bad = self._scan()
|
||||
for n in bad:
|
||||
log("журнал: строка %d не читается, пропускаю" % n)
|
||||
return good
|
||||
|
||||
def started(self, command_id) -> None:
|
||||
self.append({"kind": "started", "command_id": int(command_id)})
|
||||
@ -248,34 +276,64 @@ class Journal:
|
||||
return [r for cid, r in last.items() if cid not in acked]
|
||||
|
||||
def unfinished_started(self) -> List[int]:
|
||||
started, finished = [], set()
|
||||
"""Порядковый учёт: started открывает id, finished/acked закрывает; порядок — первое открытие."""
|
||||
opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия
|
||||
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]
|
||||
kind, cid = r.get("kind"), r["command_id"]
|
||||
if kind == "started":
|
||||
opened[cid] = True
|
||||
elif kind in ("finished", "acked") and cid in opened:
|
||||
opened[cid] = False
|
||||
return [cid for cid, is_open in opened.items() if is_open]
|
||||
|
||||
def prune(self, days: int = 30) -> int:
|
||||
"""Удалить записи команд, подтверждённых (acked) раньше, чем days дней назад. Неподтверждённые — остаются."""
|
||||
records = self.records()
|
||||
"""Удалить записи id до его последнего acked, если тот старше days дней (возраст — по acked)
|
||||
и после него у id нет started/finished. Возвращает число удалённых годных записей; битые строки
|
||||
отбрасываются при перезаписи и учитываются в логе."""
|
||||
good, bad = self._scan()
|
||||
if bad:
|
||||
log("журнал: удаляю битых строк %d" % len(bad))
|
||||
cutoff = datetime.now(timezone.utc).timestamp() - days * 86400
|
||||
old = set()
|
||||
for r in records:
|
||||
last_ack, last_open = {}, {}
|
||||
for i, r in enumerate(good):
|
||||
cid = r["command_id"]
|
||||
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]
|
||||
last_ack[cid] = i
|
||||
elif r.get("kind") in ("started", "finished"):
|
||||
last_open[cid] = i
|
||||
drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется
|
||||
for cid, i in last_ack.items():
|
||||
if last_open.get(cid, -1) > i:
|
||||
continue
|
||||
try:
|
||||
ts = _parse_at(good[i].get("at"))
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
if ts < cutoff:
|
||||
drop_upto[cid] = i
|
||||
keep = [r for i, r in enumerate(good)
|
||||
if not (r["command_id"] in drop_upto and i <= drop_upto[r["command_id"]])]
|
||||
tmp = self.path + ".tmp"
|
||||
with open(tmp, "w", encoding="utf-8") as f:
|
||||
try:
|
||||
os.remove(tmp)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
||||
with os.fdopen(fd, "wb") as f:
|
||||
for r in keep:
|
||||
f.write(json.dumps(r, ensure_ascii=False) + "\n")
|
||||
f.write((json.dumps(r, ensure_ascii=False) + "\n").encode("utf-8"))
|
||||
f.flush()
|
||||
os.fsync(f.fileno())
|
||||
os.replace(tmp, self.path)
|
||||
return len(records) - len(keep)
|
||||
return len(good) - len(keep)
|
||||
|
||||
|
||||
def _parse_at(value) -> float:
|
||||
"""Метка времени журнала в эпоху; наивное время трактуется как UTC."""
|
||||
dt = datetime.fromisoformat(value)
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt.timestamp()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@ -1,9 +1,8 @@
|
||||
import contextlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
from _common import agent
|
||||
|
||||
@ -13,6 +12,9 @@ class JournalTest(unittest.TestCase):
|
||||
self.dir = tempfile.TemporaryDirectory()
|
||||
self.path = os.path.join(self.dir.name, "journal.jsonl")
|
||||
self.j = agent.Journal(self.path)
|
||||
patcher = mock.patch.object(agent, "log")
|
||||
self.log = patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
def tearDown(self):
|
||||
self.dir.cleanup()
|
||||
@ -45,11 +47,8 @@ class JournalTest(unittest.TestCase):
|
||||
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())
|
||||
self.assertEqual(len(self.j.records()), 2)
|
||||
self.log.assert_called_with("журнал: строка 2 не читается, пропускаю")
|
||||
|
||||
def test_prune_keeps_unacked_and_recent(self):
|
||||
self.j.started(1)
|
||||
@ -73,5 +72,68 @@ class JournalTest(unittest.TestCase):
|
||||
self.assertEqual(json.loads(text.splitlines()[0])["kind"], "finished")
|
||||
|
||||
|
||||
def test_torn_tail_is_not_glued(self):
|
||||
with open(self.path, "wb") as f:
|
||||
f.write('{"kind": "star'.encode("utf-8"))
|
||||
self.j.started(1)
|
||||
recs = self.j.records()
|
||||
self.assertEqual([(r["kind"], r["command_id"]) for r in recs], [("started", 1)])
|
||||
self.assertEqual(self.j.unfinished_started(), [1])
|
||||
|
||||
def test_restart_after_acked_reopens(self):
|
||||
self.j.started(1)
|
||||
self.j.finished(1, "done", None, None, 1, 0)
|
||||
self.j.acked(1)
|
||||
self.assertEqual(self.j.unfinished_started(), [])
|
||||
self.j.started(1)
|
||||
self.assertEqual(self.j.unfinished_started(), [1])
|
||||
|
||||
def test_duplicate_started_is_deduplicated_in_first_open_order(self):
|
||||
self.j.started(2)
|
||||
self.j.started(1)
|
||||
self.j.started(1)
|
||||
self.assertEqual(self.j.unfinished_started(), [2, 1])
|
||||
self.j.finished(2, "done", None, None, 1, 0)
|
||||
self.assertEqual(self.j.unfinished_started(), [1])
|
||||
|
||||
def test_prune_keeps_id_with_newer_records(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(1)
|
||||
self.j.finished(1, "done", None, None, 1, 0)
|
||||
before = self.j.records()
|
||||
self.assertEqual(self.j.prune(days=30), 0)
|
||||
self.assertEqual(self.j.records(), before)
|
||||
|
||||
def test_non_dict_and_no_command_id_lines_are_skipped(self):
|
||||
with open(self.path, "wb") as f:
|
||||
f.write(b'[1, 2]\n5\n{"kind": "started"}\n')
|
||||
self.j.started(1)
|
||||
self.j.finished(1, "done", None, None, 1, 0)
|
||||
self.j.acked(1)
|
||||
self.assertEqual(len(self.j.records()), 3)
|
||||
self.assertEqual(self.j.unacked_finished(), [])
|
||||
self.assertEqual(self.j.unfinished_started(), [])
|
||||
self.assertEqual(self.j.prune(days=30), 0)
|
||||
self.log.assert_any_call("журнал: удаляю битых строк 3")
|
||||
self.assertEqual(len(self.j.records()), 3)
|
||||
with open(self.path, encoding="utf-8") as f:
|
||||
self.assertNotIn("[1, 2]", f.read())
|
||||
|
||||
def test_naive_and_invalid_at_in_prune(self):
|
||||
self.j.started(1)
|
||||
self.j.append({"kind": "acked", "command_id": 1, "at": "2020-01-01T00:00:00"})
|
||||
self.j.started(2)
|
||||
self.j.append({"kind": "acked", "command_id": 2, "at": "не дата"})
|
||||
self.assertEqual(self.j.prune(days=30), 2)
|
||||
self.assertEqual([(r["kind"], r["command_id"]) for r in self.j.records()], [("started", 2), ("acked", 2)])
|
||||
|
||||
def test_journal_file_is_private(self):
|
||||
self.j.started(1)
|
||||
mode = os.stat(self.path).st_mode & 0o777
|
||||
self.assertEqual(mode & 0o077, 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user