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() == ""