fix(ledger): report malformed entries during verification

src/flightdeck/ledger.py

@@ -14,11 +14,13 @@
 
 import hashlib
 import json
+from collections.abc import Iterator
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
 
 GENESIS = "0" * 64
+_ENTRY_FIELDS = ("seq", "at", "event", "data", "prev", "hash")
 
 
 def _canonical(data: dict) -> str:
@@ -30,6 +32,22 @@ def _entry_hash(seq: int, at: str, event: str, data: dict, prev: str) -> str:
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()
 
 
+def _entry_shape_error(entry: object) -> str | None:
+    if not isinstance(entry, dict):
+        return "invalid entry: expected a JSON object"
+    missing = [field for field in _ENTRY_FIELDS if field not in entry]
+    if missing:
+        return f"invalid entry: missing field(s): {', '.join(missing)}"
+    if type(entry["seq"]) is not int:
+        return "invalid entry: field 'seq' must be an integer"
+    for field in ("at", "event", "prev", "hash"):
+        if not isinstance(entry[field], str):
+            return f"invalid entry: field '{field}' must be a string"
+    if not isinstance(entry["data"], dict):
+        return "invalid entry: field 'data' must be a JSON object"
+    return None
+
+
 @dataclass
 class VerifyResult:
     entries: int
@@ -75,28 +93,59 @@ def append(self, event: str, data: dict, at: datetime | None = None) -> dict:
         return entry
 
     def entries(self) -> list[dict]:
+        return [json.loads(line.decode("utf-8")) for line in self._entry_lines()]
+
+    def _entry_lines(self) -> Iterator[bytes]:
         if not self.path.exists():
-            return []
-        out = []
-        with self.path.open(encoding="utf-8") as fh:
-            for line in fh:
-                line = line.strip()
-                if line:
-                    out.append(json.loads(line))
-        return out
+            return
+        with self.path.open("rb") as fh:
+            for raw_line in fh:
+                if line := raw_line.strip():
+                    yield line
 
     def verify(self) -> VerifyResult:
         """Re-walk the chain: sequence must be gapless from 0, each prev must match
         the previous hash, each hash must recompute. First break wins."""
-        entries = self.entries()
+        entry_count = 0
+        failure: tuple[int, str] | None = None
         prev = GENESIS
-        for index, entry in enumerate(entries):
+        for index, line in enumerate(self._entry_lines()):
+            entry_count += 1
+            if failure is not None:
+                continue
+            try:
+                text = line.decode("utf-8")
+            except UnicodeDecodeError as exc:
+                failure = (index, f"invalid UTF-8: {exc.reason}")
+                continue
+            try:
+                entry = json.loads(text)
+            except json.JSONDecodeError as exc:
+                failure = (index, f"invalid JSON: {exc.msg}")
+                continue
+            except (RecursionError, ValueError) as exc:
+                failure = (index, f"invalid JSON: {exc}")
+                continue
+            if shape_error := _entry_shape_error(entry):
+                failure = (index, shape_error)
+                continue
             if entry["seq"] != index:
-                return VerifyResult(len(entries), False, entry["seq"], "sequence gap or reorder")
+                failure = (entry["seq"], "sequence gap or reorder")
+                continue
             if entry["prev"] != prev:
-                return VerifyResult(len(entries), False, entry["seq"], "broken chain link")
-            expected = _entry_hash(entry["seq"], entry["at"], entry["event"], entry["data"], entry["prev"])
+                failure = (entry["seq"], "broken chain link")
+                continue
+            try:
+                expected = _entry_hash(
+                    entry["seq"], entry["at"], entry["event"], entry["data"], entry["prev"]
+                )
+            except (RecursionError, UnicodeEncodeError, ValueError) as exc:
+                failure = (entry["seq"], f"invalid entry: {exc}")
+                continue
             if entry["hash"] != expected:
-                return VerifyResult(len(entries), False, entry["seq"], "entry hash mismatch")
+                failure = (entry["seq"], "entry hash mismatch")
+                continue
             prev = entry["hash"]
-        return VerifyResult(len(entries), True)
+        if failure is not None:
+            return VerifyResult(entry_count, False, *failure)
+        return VerifyResult(entry_count, True)

tests/test_cli_and_html.py

@@ -4,6 +4,7 @@
 import json
 from pathlib import Path
 
+import pytest
 import yaml
 from typer.testing import CliRunner
 
@@ -122,6 +123,43 @@ def test_run_feedback_report_loop_offline(tmp_path):
     assert "chain intact" in result.output
 
 
+@pytest.mark.parametrize(
+    ("corruption", "reason"),
+    [
+        ("shape", "invalid entry"),
+        ("utf16", "invalid UTF-8"),
+        ("huge_integer", "invalid JSON"),
+        ("surrogate", "invalid entry"),
+    ],
+)
+def test_audit_and_report_survive_an_invalid_ledger_entry(tmp_path, corruption, reason):
+    root = _init(tmp_path)
+    org = load_org(root)
+    if corruption == "utf16":
+        Ledger(org.ledger_path).append("event", {"n": 0})
+        record = org.ledger_path.read_text(encoding="utf-8").strip()
+        org.ledger_path.write_bytes(record.encode("utf-16"))
+    elif corruption == "huge_integer":
+        org.ledger_path.write_text('{"seq":' + "9" * 5000 + "}\n", encoding="utf-8")
+    elif corruption == "surrogate":
+        org.ledger_path.write_text(
+            '{"seq":0,"at":"2026-08-29T20:00:00+00:00","event":"\\ud800",'
+            '"data":{},"prev":"' + "0" * 64 + '","hash":"x"}\n',
+            encoding="utf-8",
+        )
+    else:
+        org.ledger_path.write_text("[]\n", encoding="utf-8")
+
+    audit = invoke("audit", "verify", "--dir", str(root))
+    assert audit.exit_code == 1
+    assert "INTEGRITY FAILURE" in audit.output
+    assert reason in audit.output
+
+    report = invoke("report", "--dir", str(root))
+    assert report.exit_code == 0, report.output
+    assert "LEDGER INTEGRITY FAILED" in report.output
+
+
 def test_report_json_with_html_keeps_stdout_valid_json(tmp_path):
     # --json is documented "for pipelines": `report --json --html x > data.json` must
     # leave stdout a single valid JSON document. The "wrote dashboard/statement"

tests/test_ledger.py

@@ -1,5 +1,7 @@
 import json
 
+import pytest
+
 from flightdeck.ledger import Ledger
 
 
@@ -16,6 +18,108 @@ def test_empty_ledger_verifies(tmp_path):
     assert result.ok and result.entries == 0
 
 
+def test_malformed_json_is_reported_as_integrity_failure(tmp_path):
+    path = tmp_path / "ledger.jsonl"
+    Ledger(path).append("event", {"n": 0})
+    path.write_text(path.read_text(encoding="utf-8") + '\n{"seq": 1\n', encoding="utf-8")
+
+    result = Ledger(path).verify()
+
+    assert not result.ok
+    assert result.entries == 2
+    assert result.broken_at == 1
+    assert result.reason.startswith("invalid JSON:")
+
+
+@pytest.mark.parametrize(
+    "entry",
+    [
+        None,
+        [],
+        {"seq": 0},
+        {
+            "seq": 0,
+            "at": "2026-08-29T20:00:00+00:00",
+            "event": "event",
+            "data": [],
+            "prev": "0" * 64,
+            "hash": "0" * 64,
+        },
+    ],
+)
+def test_invalid_entry_shape_is_reported_as_integrity_failure(tmp_path, entry):
+    path = tmp_path / "ledger.jsonl"
+    path.write_text(json.dumps(entry) + "\n", encoding="utf-8")
+
+    result = Ledger(path).verify()
+
+    assert not result.ok
+    assert result.entries == 1
+    assert result.broken_at == 0
+    assert result.reason.startswith("invalid entry:")
+
+
+def test_invalid_utf8_is_reported_as_integrity_failure(tmp_path):
+    path = tmp_path / "ledger.jsonl"
+    path.write_bytes(b"\xff\n")
+
+    result = Ledger(path).verify()
+
+    assert not result.ok
+    assert result.entries == 1
+    assert result.broken_at == 0
+    assert result.reason.startswith("invalid UTF-8:")
+
+
+def test_utf16_record_is_not_accepted_as_utf8(tmp_path):
+    path = tmp_path / "ledger.jsonl"
+    Ledger(path).append("event", {"n": 0})
+    record = path.read_text(encoding="utf-8").strip()
+    path.write_bytes(record.encode("utf-16"))
+
+    result = Ledger(path).verify()
+
+    assert not result.ok
+    assert result.broken_at == 0
+    assert result.reason.startswith("invalid UTF-8:")
+
+
+def test_entries_rejects_utf16_records(tmp_path):
+    path = tmp_path / "ledger.jsonl"
+    Ledger(path).append("event", {"n": 0})
+    record = path.read_text(encoding="utf-8").strip()
+    path.write_bytes(record.encode("utf-16"))
+
+    with pytest.raises(UnicodeDecodeError):
+        Ledger(path).entries()
+
+
+def test_oversized_json_integer_is_reported_as_integrity_failure(tmp_path):
+    path = tmp_path / "ledger.jsonl"
+    path.write_text('{"seq":' + "9" * 5000 + "}\n", encoding="utf-8")
+
+    result = Ledger(path).verify()
+
+    assert not result.ok
+    assert result.broken_at == 0
+    assert result.reason.startswith("invalid JSON:")
+
+
+def test_invalid_unicode_is_reported_as_integrity_failure(tmp_path):
+    path = tmp_path / "ledger.jsonl"
+    path.write_text(
+        '{"seq":0,"at":"2026-08-29T20:00:00+00:00","event":"\\ud800",'
+        '"data":{},"prev":"' + "0" * 64 + '","hash":"x"}\n',
+        encoding="utf-8",
+    )
+
+    result = Ledger(path).verify()
+
+    assert not result.ok
+    assert result.broken_at == 0
+    assert result.reason.startswith("invalid entry:")
+
+
 def test_tampered_data_breaks_at_that_entry(tmp_path):
     path = tmp_path / "ledger.jsonl"
     ledger = Ledger(path)