Spaces:
Sleeping
Sleeping
File size: 2,894 Bytes
2e818da | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """Generic append-only JSONL log with a byte-offset cursor, shared by the
student-interaction log and the approved-profile-staging log. Never
truncates or deletes the JSONL file — only the cursor advances, and only
after the caller confirms the batch read up to that offset is durable.
See docs/commit-memory-adaptation-architecture.md.
"""
from __future__ import annotations
import json
import os
import tempfile
from pathlib import Path
def append_line(jsonl_path: Path, payload: dict) -> None:
jsonl_path.parent.mkdir(parents=True, exist_ok=True)
with open(jsonl_path, "a", encoding="utf-8") as handle:
handle.write(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n")
def _read_cursor(cursor_path: Path, cursor_field: str) -> int:
if not cursor_path.exists():
return 0
try:
return int(json.loads(cursor_path.read_text(encoding="utf-8")).get(cursor_field, 0))
except (ValueError, json.JSONDecodeError):
return 0
def read_exact_range(jsonl_path: Path, from_offset: int, through_offset: int) -> list[str]:
"""Read exactly [from_offset, through_offset), independent of the
cursor. Used to retry a pending-review batch against the precise
historical range it was built from, rather than "since cursor to
current EOF" (which may have grown if new entries were appended
while the batch was in flight)."""
if not jsonl_path.exists() or through_offset <= from_offset:
return []
with open(jsonl_path, "rb") as handle:
handle.seek(from_offset)
chunk = handle.read(through_offset - from_offset)
return [line.decode("utf-8") for line in chunk.splitlines() if line.strip()]
def read_pending_range(
jsonl_path: Path, cursor_path: Path, cursor_field: str,
) -> tuple[list[str], int, int]:
from_offset = _read_cursor(cursor_path, cursor_field)
if not jsonl_path.exists():
return [], from_offset, from_offset
with open(jsonl_path, "rb") as handle:
handle.seek(0, os.SEEK_END)
through_offset = handle.tell()
if through_offset <= from_offset:
return [], from_offset, through_offset
handle.seek(from_offset)
chunk = handle.read(through_offset - from_offset)
lines = [line.decode("utf-8") for line in chunk.splitlines() if line.strip()]
return lines, from_offset, through_offset
def advance_cursor(cursor_path: Path, cursor_field: str, through_offset: int) -> None:
cursor_path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=str(cursor_path.parent), suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump({cursor_field: through_offset}, handle)
os.replace(tmp_path, cursor_path)
except BaseException:
if os.path.exists(tmp_path):
os.unlink(tmp_path)
raise
|