Spaces:
Sleeping
Sleeping
| """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 | |