From 6e5320c18bc78bc4faa47c063d2774c1fc790aca Mon Sep 17 00:00:00 2001 From: Layne Penney Date: Thu, 20 Aug 2026 07:59:41 -0500 Subject: [PATCH] fix: one torn-line-safe JSONL home for both lane prototypes Both lane prototypes carried a BYTE-IDENTICAL append_jsonl and a reader with the same body under two names. That is one defect with two addresses, not a class with two instances, so it gets one fix rather than two: a shared jsonl_store module both consume. Fixing them separately would have written the same fix and the same witnesses twice and left the copies free to diverge again, which is how the situation arose. The defect, in both copies: the writer appended with no terminator repair and no fsync, so a writer killed mid-write left a remnant and the NEXT record glued onto it - a record that had itself completed became part of one unparseable line and was lost. The reader parsed with a bare json.loads over read_text().splitlines(), so the remnant raised, and the whole-file decode meant one invalid byte anywhere destroyed every record in the file rather than its own line. WHERE THE COUNT LIVES, which was the open design question: these are module-level functions with no instance to hang health on, and a module-level accumulator would be hidden state and wrong under concurrent readers. So the count is a RETURN VALUE - read_jsonl returns rows AND the lines it could not read. The question dissolves rather than getting answered. And the count reaches a CONSUMER: both CLI callers report it on stderr, so a --json caller's stdout stays machine-readable. A count nobody surfaces is the same silence as no count at all, which is the defect this fix exists to close. The reporter lives in the shared module too. The first version of this change put a helper in one consumer and an inline copy of the same three lines in the other - byte-adjacent duplication of a reporting rule across two files, which is the exact shape being eliminated, committed inside the fix. Caught before the gate. Structure is validated; schema is not. A generic reader has no fields to check, so a line that is well-formed JSON and a well-formed object IS a row even when its values are odd. That boundary is pinned by its own witness, added after a witness failed against a wrong expectation of mine rather than against the code - the hostile-content table had been copied from a surface that does have a schema. 21 witnesses, 7 mutation rows, each mutation killing witnesses whose failure TYPE matches it: removing the decode guard kills seven by real UnicodeDecodeError; removing the parse guard kills six by real JSONDecodeError, RecursionError and the integer-literal ValueError; silencing the count kills fifteen; silencing the reporter kills exactly one, which is what proves the count reaches a consumer. The fsync witness is deliberately weaker than it sounds and says so: it pins that fsync is CALLED, not that durability holds, because a real crash is unwitnessable in-process. It exists because without it, deleting the fsync killed nothing at all - and a guard nothing checks is not a guard. Unbounded line length is a resource limit on the READ layer and is deliberately not defended, named in the module docstring rather than left to be discovered. Ref #897 Co-Authored-By: Claude --- gr2/prototypes/jsonl_store.py | 158 ++++++++++++++ gr2/prototypes/lane_workspace_prototype.py | 156 ++++++++------ gr2/prototypes/recall_lane_history.py | 40 ++-- gr2/tests/test_jsonl_store.py | 239 +++++++++++++++++++++ 4 files changed, 498 insertions(+), 95 deletions(-) create mode 100644 gr2/prototypes/jsonl_store.py create mode 100644 gr2/tests/test_jsonl_store.py diff --git a/gr2/prototypes/jsonl_store.py b/gr2/prototypes/jsonl_store.py new file mode 100644 index 0000000..6b280b0 --- /dev/null +++ b/gr2/prototypes/jsonl_store.py @@ -0,0 +1,158 @@ +"""Torn-line-safe JSONL primitives, shared by the lane prototypes. + +Both lane prototypes carried a BYTE-IDENTICAL ``append_jsonl`` and a reader with +the same body under two names. They were not two instances of a defect class; +they were one function, copied. Fixing them separately would have written the +same fix and the same witnesses twice and left the copies free to diverge again, +which is how the situation arose. One home, two consumers. + +The defect, measured on the sibling surface this shape came from: + +* the writer appended with no terminator repair and no ``fsync``, so a writer + killed mid-write left a remnant and the NEXT record glued onto it — a record + that had itself completed became part of one unparseable line and was lost; +* the reader parsed with a bare ``json.loads`` over ``read_text().splitlines()``, + so the remnant raised, and the whole-file decode meant one invalid byte + anywhere destroyed every record in the file rather than its own line. + +A line passes through four layers — read, decode, parse, shape-check — and the +last three are each guarded where they happen. Types are VALIDATED rather than +coerced, because a coercion over untrusted bytes forces the caller to enumerate +the exceptions it might raise, which is a denylist, and a denylist leaks. +Unbounded line length is a resource limit on the READ layer and is deliberately +NOT defended here; it is named rather than left to be discovered. +""" + +from __future__ import annotations + +import json +import os +import sys +from dataclasses import dataclass, field +from pathlib import Path + +from gr2.prototypes.propagation_state_machine import MalformedLine + + +@dataclass(frozen=True) +class JsonlRead: + """Rows that could be read, and the lines that could not. + + The count travels WITH the data. These are module-level functions with no + object to hang health on, and a module-level accumulator would be hidden + state and wrong under concurrent readers — so the health of the scan is a + return value instead. A caller that ignores ``malformed`` still gets correct + rows; a caller that wants to report health already holds it. Skipping alone + is silent, and silence is the defect: the remnant sits in the file while the + caller sees a healthy-looking result. + """ + + rows: list[dict] + malformed: tuple[MalformedLine, ...] = field(default=()) + + +def _decode_line(raw: bytes) -> tuple[str | None, str]: + """Bytes become text HERE, so the failure that can happen HERE is guarded here.""" + + try: + return raw.decode("utf-8"), "" + except UnicodeDecodeError as exc: + return None, str(exc) + + +def _read_object(line: str) -> tuple[dict | None, str]: + """``json.loads`` is the only operation over a decoded line that can raise.""" + + try: + obj = json.loads(line) + except (ValueError, RecursionError) as exc: + return None, str(exc) + if not isinstance(obj, dict): + return None, f"line is a {type(obj).__name__}, not an object" + return obj, "" + + +def append_jsonl(path: Path, payload: dict) -> None: + """Append one record, repairing a missing terminator first, then fsync. + + Without the repair a remnant GLUES the next record onto itself and the next + record is lost. Without the fsync the record is not durable against the very + crash that produces remnants. + """ + + path.parent.mkdir(parents=True, exist_ok=True) + unterminated = False + try: + with path.open("rb") as probe: + if probe.seek(0, os.SEEK_END): + probe.seek(-1, os.SEEK_END) + unterminated = probe.read(1) != b"\n" + except FileNotFoundError: + pass + with path.open("ab") as handle: + if unterminated: + handle.write(b"\n") + handle.write((json.dumps(payload) + "\n").encode("utf-8")) + handle.flush() + os.fsync(handle.fileno()) + + +def read_jsonl(path: Path) -> JsonlRead: + """Read every line that can be read, and report every line that cannot.""" + + if not path.exists(): + return JsonlRead(rows=[]) + rows: list[dict] = [] + malformed: list[MalformedLine] = [] + for index, raw in enumerate(path.read_bytes().split(b"\n")): + stripped = raw.strip() + if not stripped: + continue + line, reason = _decode_line(stripped) + if line is not None: + obj, reason = _read_object(line) + if obj is not None: + rows.append(obj) + continue + malformed.append( + MalformedLine( + index=index, + excerpt=stripped[:120].decode("utf-8", "replace"), + reason=reason, + ) + ) + return JsonlRead(rows=rows, malformed=tuple(malformed)) + + +def warn_unreadable(read: JsonlRead, what: str, stream: object = None) -> bool: + """Report unreadable lines on STDERR. Returns whether anything was reported. + + Lives HERE, beside the primitives it reports on, because the first version of + this fix put a helper in one consumer and an inline copy of the same three + lines in the other — which is byte-adjacent duplication of a reporting rule + across two files, the same shape as the copied ``append_jsonl`` this module + exists to eliminate. One home for the rule, or the message formats drift. + + STDERR specifically: a ``--json`` caller's stdout must stay machine-readable. + And it is reported at ALL because a count nobody surfaces is the same silence + as no count — the CLI is the layer that already looks. + """ + + if not read.malformed: + return False + first = read.malformed[0] + print( + f"warning: {len(read.malformed)} unreadable line(s) in {what}; " + f"first at line {first.index}: {first.reason}", + file=stream if stream is not None else sys.stderr, + ) + return True + + +__all__ = [ + "JsonlRead", + "MalformedLine", + "append_jsonl", + "read_jsonl", + "warn_unreadable", +] diff --git a/gr2/prototypes/lane_workspace_prototype.py b/gr2/prototypes/lane_workspace_prototype.py index 21749ef..44582bd 100644 --- a/gr2/prototypes/lane_workspace_prototype.py +++ b/gr2/prototypes/lane_workspace_prototype.py @@ -25,6 +25,12 @@ from pathlib import Path import tomli_w +from gr2.prototypes.jsonl_store import ( + JsonlRead, + append_jsonl, + read_jsonl, + warn_unreadable, +) LANE_SCHEMA_VERSION = 1 SCRATCHPAD_SCHEMA_VERSION = 1 @@ -140,9 +146,7 @@ def as_toml(self) -> str: def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Prototype gr2 lanes + shared scratchpads" - ) + parser = argparse.ArgumentParser(description="Prototype gr2 lanes + shared scratchpads") sub = parser.add_subparsers(dest="command", required=True) create = sub.add_parser("create-lane") @@ -221,7 +225,9 @@ def parse_args() -> argparse.Namespace: enter.add_argument("workspace_root", type=Path) enter.add_argument("owner_unit") enter.add_argument("lane_name") - enter.add_argument("--actor", required=True, help="actor label, e.g. human:layne or agent:atlas") + enter.add_argument( + "--actor", required=True, help="actor label, e.g. human:layne or agent:atlas" + ) enter.add_argument("--notify-channel", action="store_true") enter.add_argument("--recall", action="store_true") @@ -355,12 +361,7 @@ def workspace_edit_leases_lock_file(workspace_root: Path) -> Path: def shared_lane_access_file(workspace_root: Path, owner_unit: str, lane_name: str) -> Path: return ( - workspace_root - / ".grip" - / "state" - / "shared_lane_access" - / owner_unit - / f"{lane_name}.json" + workspace_root / ".grip" / "state" / "shared_lane_access" / owner_unit / f"{lane_name}.json" ) @@ -410,12 +411,6 @@ def load_current_lane_doc(workspace_root: Path, owner_unit: str) -> dict: return json.loads(path.read_text()) -def append_jsonl(path: Path, payload: dict) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("a", encoding="utf-8") as fh: - fh.write(json.dumps(payload) + "\n") - - def emit_lane_event(workspace_root: Path, payload: dict) -> None: append_jsonl(lane_events_file(workspace_root), payload) @@ -424,17 +419,10 @@ def emit_recall_lane_event(workspace_root: Path, payload: dict) -> None: append_jsonl(recall_lane_events_file(workspace_root), payload) -def iter_lane_events(workspace_root: Path) -> list[dict]: - path = lane_events_file(workspace_root) - if not path.exists(): - return [] - items: list[dict] = [] - for line in path.read_text().splitlines(): - line = line.strip() - if not line: - continue - items.append(json.loads(line)) - return items +def iter_lane_events(workspace_root: Path) -> JsonlRead: + """Rows AND the lines that could not be read. See ``jsonl_store.JsonlRead``.""" + + return read_jsonl(lane_events_file(workspace_root)) def write_json(path: Path, payload: dict) -> None: @@ -654,7 +642,9 @@ def lease_conflicts(existing_mode: str, requested_mode: str) -> bool: return requested_mode in matrix.get(existing_mode, set()) -def conflicting_leases(leases: list[dict], actor: str, requested_mode: str) -> tuple[list[dict], list[dict]]: +def conflicting_leases( + leases: list[dict], actor: str, requested_mode: str +) -> tuple[list[dict], list[dict]]: active: list[dict] = [] stale: list[dict] = [] for lease in leases: @@ -801,8 +791,8 @@ def enter_lane(args: argparse.Namespace) -> int: emit_lane_event(workspace_root, event) if args.notify_channel: event["channel_message"] = ( - f'{args.actor} entered {args.owner_unit}/{args.lane_name} ' - f'[{lane_doc["lane_type"]}] repos={",".join(lane_doc.get("repos", []))}' + f"{args.actor} entered {args.owner_unit}/{args.lane_name} " + f"[{lane_doc['lane_type']}] repos={','.join(lane_doc.get('repos', []))}" ) if args.recall: emit_recall_lane_event( @@ -842,8 +832,8 @@ def exit_lane(args: argparse.Namespace) -> int: emit_lane_event(workspace_root, event) if args.notify_channel: event["channel_message"] = ( - f'{args.actor} exited {args.owner_unit}/{current_doc["lane_name"]} ' - f'[{current_doc["lane_type"]}]' + f"{args.actor} exited {args.owner_unit}/{current_doc['lane_name']} " + f"[{current_doc['lane_type']}]" ) if args.recall: emit_recall_lane_event( @@ -867,7 +857,9 @@ def exit_lane(args: argparse.Namespace) -> int: "current": next_current, "recent": recent[1:] if next_current else [], } - current_lane_file(workspace_root, args.owner_unit).write_text(json.dumps(updated, indent=2) + "\n") + current_lane_file(workspace_root, args.owner_unit).write_text( + json.dumps(updated, indent=2) + "\n" + ) print(current_lane_file(workspace_root, args.owner_unit)) return 0 @@ -879,28 +871,29 @@ def current_lane(args: argparse.Namespace) -> int: return 0 current_doc = doc["current"] print("gr2 prototype current-lane") - print(f'owner={current_doc["owner_unit"]} lane={current_doc["lane_name"]} type={current_doc["lane_type"]} actor={current_doc["actor"]}') - print(f'entered_at={current_doc["entered_at"]}') + print( + f"owner={current_doc['owner_unit']} lane={current_doc['lane_name']} type={current_doc['lane_type']} actor={current_doc['actor']}" + ) + print(f"entered_at={current_doc['entered_at']}") recent = doc.get("recent", []) if recent: print("recent:") for item in recent: - print(f' - {item["owner_unit"]}/{item["lane_name"]} ({item["lane_type"]})') + print(f" - {item['owner_unit']}/{item['lane_name']} ({item['lane_type']})") return 0 def lane_history(args: argparse.Namespace) -> int: - rows = [ - event for event in iter_lane_events(args.workspace_root.resolve()) - if event.get("owner_unit") == args.owner_unit - ] + read = iter_lane_events(args.workspace_root.resolve()) + warn_unreadable(read, "the lane event log") + rows = [event for event in read.rows if event.get("owner_unit") == args.owner_unit] if args.json: print(json.dumps(rows, indent=2)) return 0 print("TIMESTAMP\tTYPE\tACTOR\tAGENT_ID\tLANE\tREPOS") for row in rows: print( - f'{row.get("timestamp","-")}\t{row.get("type","-")}\t{row.get("agent","-")}\t{row.get("agent_id","-")}\t{row.get("lane","-")}\t{",".join(row.get("repos", []))}' + f"{row.get('timestamp', '-')}\t{row.get('type', '-')}\t{row.get('agent', '-')}\t{row.get('agent_id', '-')}\t{row.get('lane', '-')}\t{','.join(row.get('repos', []))}" ) return 0 @@ -949,6 +942,7 @@ def acquire_lane_lease(args: argparse.Namespace) -> int: print(lane_leases_file(workspace_root, args.owner_unit, args.lane_name)) return 0 + def _acquire_lane_lease_mutation(leases: list[dict], args: argparse.Namespace) -> dict: retained = [lease for lease in leases if lease["actor"] != args.actor] active_conflicts, stale_conflicts = conflicting_leases(retained, args.actor, args.mode) @@ -1041,7 +1035,7 @@ def show_lane_leases(args: argparse.Namespace) -> int: for lease in leases: state = "stale" if is_stale_lease(lease) else "active" print( - f'{lease["actor"]}\t{lease["mode"]}\t{lease.get("ttl_seconds", "-")}\t{lease["acquired_at"]}\t{lease.get("expires_at", "-")}\t{state}' + f"{lease['actor']}\t{lease['mode']}\t{lease.get('ttl_seconds', '-')}\t{lease['acquired_at']}\t{lease.get('expires_at', '-')}\t{state}" ) return 0 @@ -1088,9 +1082,7 @@ def check_review_requirements(args: argparse.Namespace) -> int: workspace_root = args.workspace_root.resolve() ref = f"{args.repo}#{args.pr_number}" required = int( - workspace_constraints(workspace_root) - .get("required_reviewers", {}) - .get(args.repo, 0) + workspace_constraints(workspace_root).get("required_reviewers", {}).get(args.repo, 0) ) matching: list[dict] = [] for path in iter_lane_files(workspace_root): @@ -1190,21 +1182,33 @@ def plan_handoff(args: argparse.Namespace) -> int: source = load_lane_doc(workspace_root, args.source_owner_unit, args.source_lane_name) find_unit_spec(workspace_root, args.target_unit) if args.mode == "shared": - access_path = shared_lane_access_file(workspace_root, args.source_owner_unit, args.source_lane_name) + access_path = shared_lane_access_file( + workspace_root, args.source_owner_unit, args.source_lane_name + ) access = json.loads(access_path.read_text()) if access_path.exists() else None payload = { "mode": "shared", "source_owner_unit": args.source_owner_unit, "source_lane_name": args.source_lane_name, "target_unit": args.target_unit, - "shared_access_present": bool(access and args.target_unit in access.get("shared_with", [])), + "shared_access_present": bool( + access and args.target_unit in access.get("shared_with", []) + ), "exec_rows": [ { "acting_unit": args.target_unit, "owner_unit": args.source_owner_unit, "lane_name": args.source_lane_name, "repo": repo, - "cwd": str(workspace_root / "agents" / args.source_owner_unit / "lanes" / args.source_lane_name / "repos" / repo), + "cwd": str( + workspace_root + / "agents" + / args.source_owner_unit + / "lanes" + / args.source_lane_name + / "repos" + / repo + ), "lease_scope": f"{args.source_owner_unit}/{args.source_lane_name}", } for repo in source.get("repos", []) @@ -1228,7 +1232,15 @@ def plan_handoff(args: argparse.Namespace) -> int: "owner_unit": args.target_unit, "lane_name": target_lane_name, "repo": repo, - "cwd": str(workspace_root / "agents" / args.target_unit / "lanes" / target_lane_name / "repos" / repo), + "cwd": str( + workspace_root + / "agents" + / args.target_unit + / "lanes" + / target_lane_name + / "repos" + / repo + ), "lease_scope": f"{args.target_unit}/{target_lane_name}", } for repo in source.get("repos", []) @@ -1293,7 +1305,7 @@ def list_lanes(args: argparse.Namespace) -> int: doc = tomllib.loads(path.read_text()) refs = ",".join(item["ref"] for item in doc.get("pr_associations", [])) or "-" print( - f'{doc["owner_unit"]}\t{doc["lane_name"]}\t{doc["lane_type"]}\t{len(doc.get("repos", []))}\t{refs}' + f"{doc['owner_unit']}\t{doc['lane_name']}\t{doc['lane_type']}\t{len(doc.get('repos', []))}\t{refs}" ) return 0 @@ -1315,7 +1327,7 @@ def list_shared_scratchpads(args: argparse.Namespace) -> int: doc = tomllib.loads(path.read_text()) participants = ",".join(doc.get("participants", [])) or "-" print( - f'{doc["name"]}\t{doc["kind"]}\t{doc["lifecycle"]}\t{age_days(path)}\t{participants}\t{doc["purpose"]}' + f"{doc['name']}\t{doc['kind']}\t{doc['lifecycle']}\t{age_days(path)}\t{participants}\t{doc['purpose']}" ) return 0 @@ -1348,7 +1360,7 @@ def audit_shared_scratchpads(args: argparse.Namespace) -> int: issues.append("empty-docs") status = "ok" if not issues else "needs-attention" - print(f'{doc["name"]}\t{status}\t{days}\t{",".join(issues) or "-"}') + print(f"{doc['name']}\t{status}\t{days}\t{','.join(issues) or '-'}") return 0 @@ -1357,24 +1369,24 @@ def plan_promote_scratchpad(args: argparse.Namespace) -> int: doc = load_shared_scratchpad_doc(workspace_root, args.name) lane_name = args.lane or f"promote-{args.name}" print("gr2 prototype scratchpad-promotion plan") - print(f'scratchpad: {doc["name"]}') - print(f'kind: {doc["kind"]}') - print(f'lifecycle: {doc["lifecycle"]}') - print(f'target repo: {args.target_repo}') - print(f'target path: {args.target_path}') - print(f'owner unit: {args.owner_unit}') - print(f'suggested lane: {lane_name}') + print(f"scratchpad: {doc['name']}") + print(f"kind: {doc['kind']}") + print(f"lifecycle: {doc['lifecycle']}") + print(f"target repo: {args.target_repo}") + print(f"target path: {args.target_path}") + print(f"owner unit: {args.owner_unit}") + print(f"suggested lane: {lane_name}") print("recommended:") - print( - f" 1. create or reuse a feature lane for {args.target_repo} under {args.owner_unit}" - ) + print(f" 1. create or reuse a feature lane for {args.target_repo} under {args.owner_unit}") print( f" 2. copy content from shared/scratchpads/{doc['name']}/docs into {args.target_repo}:{args.target_path}" ) print(f" 3. branch and commit in lane {lane_name}") print(" 4. open a PR once the artifact is ready for formal review") if not doc.get("linked_refs"): - print("warning: scratchpad has no linked refs; traceability should be added before promotion") + print( + "warning: scratchpad has no linked refs; traceability should be added before promotion" + ) return 0 @@ -1415,9 +1427,9 @@ def next_step(args: argparse.Namespace) -> int: workspace_root = args.workspace_root.resolve() lane_doc = load_lane_doc(workspace_root, args.owner_unit, args.lane_name) print("gr2 prototype next-step") - print(f'lane: {args.owner_unit}/{lane_doc["lane_name"]}') - print(f'type: {lane_doc["lane_type"]}') - print(f'repos: {", ".join(lane_doc["repos"])}') + print(f"lane: {args.owner_unit}/{lane_doc['lane_name']}") + print(f"type: {lane_doc['lane_type']}") + print(f"repos: {', '.join(lane_doc['repos'])}") if lane_doc.get("pr_associations"): print("mode: review") print("recommended:") @@ -1463,7 +1475,9 @@ def plan_exec(args: argparse.Namespace) -> int: print("gr2 lane-exec prototype") print("status=blocked reason=conflicting-active-lease") for lease in active_conflicts: - print(f'conflict: actor={lease["actor"]} mode={lease["mode"]} acquired_at={lease["acquired_at"]}') + print( + f"conflict: actor={lease['actor']} mode={lease['mode']} acquired_at={lease['acquired_at']}" + ) return 0 if stale_conflicts: payload = { @@ -1481,7 +1495,9 @@ def plan_exec(args: argparse.Namespace) -> int: print("gr2 lane-exec prototype") print("status=blocked reason=stale-conflicting-lease") for lease in stale_conflicts: - print(f'stale-conflict: actor={lease["actor"]} mode={lease["mode"]} expires_at={lease.get("expires_at", "-")}') + print( + f"stale-conflict: actor={lease['actor']} mode={lease['mode']} expires_at={lease.get('expires_at', '-')}" + ) return 0 selected_repos = lane_doc["repos"] @@ -1519,12 +1535,12 @@ def plan_exec(args: argparse.Namespace) -> int: else: print("gr2 lane-exec prototype") print( - f'owner={lane_doc["owner_unit"]} lane={lane_doc["lane_name"]} type={lane_doc["lane_type"]} fail_fast={lane_doc["exec_defaults"]["fail_fast"]}' + f"owner={lane_doc['owner_unit']} lane={lane_doc['lane_name']} type={lane_doc['lane_type']} fail_fast={lane_doc['exec_defaults']['fail_fast']}" ) print("LANE\tREPO\tBRANCH\tCWD\tCOMMAND") for row in rows: print( - f'{row["lane"]}\t{row["repo"]}\t{row["branch"]}\t{row["cwd"]}\t{" ".join(row["command"])}' + f"{row['lane']}\t{row['repo']}\t{row['branch']}\t{row['cwd']}\t{' '.join(row['command'])}" ) return 0 diff --git a/gr2/prototypes/recall_lane_history.py b/gr2/prototypes/recall_lane_history.py index 323bf65..f54b5ce 100644 --- a/gr2/prototypes/recall_lane_history.py +++ b/gr2/prototypes/recall_lane_history.py @@ -10,11 +10,16 @@ from pathlib import Path from typing import Any +from gr2.prototypes.jsonl_store import ( + JsonlRead, + append_jsonl, + read_jsonl, + warn_unreadable, +) + def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser( - description="Prototype recall lane history surface" - ) + parser = argparse.ArgumentParser(description="Prototype recall lane history surface") sub = parser.add_subparsers(dest="command", required=True) demo = sub.add_parser("demo-data") @@ -40,22 +45,10 @@ def lane_events_file(workspace_root: Path) -> Path: return events_dir(workspace_root) / "lane_events.jsonl" -def append_jsonl(path: Path, payload: dict) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - with path.open("a", encoding="utf-8") as fh: - fh.write(json.dumps(payload) + "\n") +def load_jsonl(path: Path) -> JsonlRead: + """Rows AND the lines that could not be read. See ``jsonl_store.JsonlRead``.""" - -def load_jsonl(path: Path) -> list[dict]: - if not path.exists(): - return [] - rows: list[dict] = [] - for line in path.read_text().splitlines(): - line = line.strip() - if not line: - continue - rows.append(json.loads(line)) - return rows + return read_jsonl(path) def parse_ts(raw: str) -> datetime: @@ -133,11 +126,7 @@ def repo_activity(index: dict[str, Any], repo: str) -> dict[str, Any]: def time_range(index: dict[str, Any], start: str, end: str) -> dict[str, Any]: start_dt = parse_ts(start) end_dt = parse_ts(end) - rows = [ - event - for event in index["all"] - if start_dt <= parse_ts(event["timestamp"]) <= end_dt - ] + rows = [event for event in index["all"] if start_dt <= parse_ts(event["timestamp"]) <= end_dt] return { "query": {"start": start, "end": end}, "count": len(rows), @@ -263,8 +252,9 @@ def main() -> int: print(json.dumps(result, indent=2)) return 0 - events = load_jsonl(lane_events_file(args.workspace_root.resolve())) - index = build_index(events) + read = load_jsonl(lane_events_file(args.workspace_root.resolve())) + warn_unreadable(read, "the lane event log") + index = build_index(read.rows) if args.command == "query": if args.lane: diff --git a/gr2/tests/test_jsonl_store.py b/gr2/tests/test_jsonl_store.py new file mode 100644 index 0000000..6124c18 --- /dev/null +++ b/gr2/tests/test_jsonl_store.py @@ -0,0 +1,239 @@ +"""Witnesses for the shared torn-line-safe JSONL primitives. + +Both lane prototypes carried a BYTE-IDENTICAL ``append_jsonl`` and a reader with +the same body under two names. This is one defect with two addresses, not a class +with two instances, so it has one fix, one home, and one witness set. + +* W1 a writer killed mid-write leaves a remnant: the append point survives it, + the NEXT record is not GLUED onto it, and the remnant is skipped AND COUNTED +* W2 hostile line CONTENT neither bricks nor vanishes (validate, never coerce) +* W3 undecodable BYTES neither brick nor vanish, and a bad byte is CONFINED to + its own line rather than destroying every record around it +* W4 the count has a REAL CONSUMER: warn_unreadable reports it, because a count + nobody surfaces is the same silence as no count at all +""" + +from __future__ import annotations + +import io +import json +import os +from pathlib import Path + +import pytest +from gr2.prototypes.jsonl_store import ( + JsonlRead, + append_jsonl, + read_jsonl, + warn_unreadable, +) + + +def _tear(path: Path, remnant: bytes) -> None: + """A writer killed between its record and its terminator: bytes, no newline.""" + with path.open("ab") as handle: + handle.write(remnant) + handle.flush() + os.fsync(handle.fileno()) + + +# --------------------------------------------------------------------------- W1 + + +def test_w1_a_torn_remnant_does_not_brick_the_append_point(tmp_path: Path) -> None: + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + _tear(path, b'{"n": 2, "part') + + append_jsonl(path, {"n": 3}) # must not raise + append_jsonl(path, {"n": 4}) # and must not raise AGAIN: not a once-survivable event + + read = read_jsonl(path) + assert [row["n"] for row in read.rows] == [1, 3, 4] + assert len(read.malformed) == 1 + + +def test_w1_the_next_record_is_not_glued_onto_the_remnant(tmp_path: Path) -> None: + """Terminator repair. Without it the record after a torn write is swallowed.""" + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + _tear(path, b'{"n": 2, "part') + + append_jsonl(path, {"n": 3}) + + lines = path.read_bytes().split(b"\n") + assert lines[1] == b'{"n": 2, "part' # the remnant, still on its own line + assert json.loads(lines[2])["n"] == 3 # the new record survived WHOLE + assert len(read_jsonl(path).rows) == 2 + + +def test_w1_a_missing_file_reads_empty_and_reports_nothing(tmp_path: Path) -> None: + read = read_jsonl(tmp_path / "never" / "written.jsonl") + assert read == JsonlRead(rows=[], malformed=()) + + +def test_w1_the_write_is_fsynced(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Pins that fsync is CALLED on the appended handle. Deliberately weaker than + it sounds, and labelled so nobody reads more into it than it proves. + + It does NOT prove durability — that needs a real crash, which is unwitnessable + in-process. What it does prove is that the call is present and reached, which + is the difference between a guard and a guard-shaped comment: without it, + deleting the fsync kills no test at all, and a guard nothing checks is not a + guard. Durability against power loss remains ASSERTED, not demonstrated. + """ + synced: list[int] = [] + real = os.fsync + monkeypatch.setattr(os, "fsync", lambda fd: (synced.append(fd), real(fd))[1]) + + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + append_jsonl(path, {"n": 2}) + + assert len(synced) == 2, "each append must fsync its own write" + + +# --------------------------------------------------------------------------- W2 + +HOSTILE_LINES = { + "deeply nested (json.loads raises RecursionError)": '{"n": ' + + "[" * 100000 + + "]" * 100000 + + "}", + "integer literal past the conversion limit": '{"n": ' + "9" * 6000 + "}", + "truncated object": '{"n": 1', + "line is an array, not an object": "[1, 2, 3]", + "line is a bare string": '"just a string"', + "line is a bare number": "42", +} + + +# Deliberately NOT hostile here: this module validates STRUCTURE (is the line a +# JSON object?) and never SCHEMA (does it carry my fields?), because its rows are +# generic and the consumers own interpretation. Nothing coerces, so nothing raises. +# The sibling AppendSurface DOES have a record schema and rejects these — the +# difference is real and is pinned below rather than assumed. +STRUCTURALLY_VALID_BUT_ODD = { + "float infinity": ('{"n": 1e999}', float("inf")), + "not-a-number": ('{"n": NaN}', None), # NaN != NaN, checked by isnan +} + + +@pytest.mark.parametrize("label", sorted(STRUCTURALLY_VALID_BUT_ODD)) +def test_w2_structure_is_validated_but_schema_is_not(label: str, tmp_path: Path) -> None: + """These ARE accepted, deliberately. A generic reader has no schema to check. + + Found by a witness failing against my own expectation rather than against the + code: I copied a hostile-content table from a surface that HAS a record schema. + The row is well-formed JSON and a well-formed object, so it is a row. + """ + import math + + line, expected = STRUCTURALLY_VALID_BUT_ODD[label] + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + with path.open("ab") as handle: + handle.write(line.encode("utf-8") + b"\n") + append_jsonl(path, {"n": 2}) + + read = read_jsonl(path) + assert len(read.rows) == 3 + assert read.malformed == () # NOT malformed: structurally fine + value = read.rows[1]["n"] + if expected is None: + assert math.isnan(value) + else: + assert value == expected + + +@pytest.mark.parametrize("label", sorted(HOSTILE_LINES)) +def test_w2_hostile_line_content_neither_bricks_nor_vanishes(label: str, tmp_path: Path) -> None: + """A guard that LISTS exception types is a denylist over untrusted input, and a + denylist leaks: it passes the case you thought of and bricks on the next one.""" + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + with path.open("ab") as handle: + handle.write(HOSTILE_LINES[label].encode("utf-8") + b"\n") + + append_jsonl(path, {"n": 2}) # the WRITE path must not brick + + read = read_jsonl(path) + assert [row["n"] for row in read.rows] == [1, 2] + assert len(read.malformed) == 1 + assert read.malformed[0].reason + + +# --------------------------------------------------------------------------- W3 + +HOSTILE_BYTES = { + "invalid byte 0xff": b'{"n": "\xff"}', + "lone surrogate": b'{"n": "\xed\xa0\x80"}', + "truncated multi-byte sequence": b'{"n": "\xe2\x82"}', + "overlong encoding": b'{"n": "\xc0\xaf"}', + "continuation byte with no lead": b'{"n": "\x80\x80"}', + "line is pure binary": b"\x00\x01\xff\xfe\xfd", +} + + +@pytest.mark.parametrize("label", sorted(HOSTILE_BYTES)) +def test_w3_undecodable_bytes_neither_brick_nor_vanish(label: str, tmp_path: Path) -> None: + """The bytes-to-text boundary is content-dependent, so it is guarded there.""" + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + with path.open("ab") as handle: + handle.write(HOSTILE_BYTES[label] + b"\n") + + append_jsonl(path, {"n": 2}) + + read = read_jsonl(path) + assert [row["n"] for row in read.rows] == [1, 2] + assert len(read.malformed) == 1 + assert read.malformed[0].excerpt # reportable even when it is not valid text + assert "utf-8" in read.malformed[0].reason + + +def test_w3_an_undecodable_line_is_confined_to_itself(tmp_path: Path) -> None: + """One bad byte costs ONE line. A whole-file decode costs every record.""" + path = tmp_path / "log" / "events.jsonl" + for n in range(1, 6): + append_jsonl(path, {"n": n}) + with path.open("ab") as handle: + handle.write(b'{"n": "\xff"}\n') + for n in range(6, 11): + append_jsonl(path, {"n": n}) + + read = read_jsonl(path) + assert [row["n"] for row in read.rows] == list(range(1, 11)) + assert len(read.malformed) == 1 + + +# --------------------------------------------------------------------------- W4 + + +def test_w4_the_count_reaches_a_consumer(tmp_path: Path) -> None: + """A count nobody surfaces is the same silence as no count. Verify by fruit.""" + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + _tear(path, b'{"n": 2, "part') + append_jsonl(path, {"n": 3}) + + out = io.StringIO() + reported = warn_unreadable(read_jsonl(path), "the lane event log", stream=out) + + assert reported is True + text = out.getvalue() + assert "1 unreadable line(s)" in text + assert "the lane event log" in text + assert "first at line 1" in text # names WHERE, so it can be acted on + + +def test_w4_a_healthy_log_reports_nothing(tmp_path: Path) -> None: + """The negative case: silence when there is nothing to say, or the signal is noise.""" + path = tmp_path / "log" / "events.jsonl" + append_jsonl(path, {"n": 1}) + + out = io.StringIO() + reported = warn_unreadable(read_jsonl(path), "the lane event log", stream=out) + + assert reported is False + assert out.getvalue() == ""