fix(pfs,llm): ledger survives truncated lines, indexes once, counts pending failures

src/llm/structured.py

@@ -111,8 +111,8 @@ class JsonExtractor:
         reply = data.get("message", {}).get("content", "")
         try:
             obj = json.loads(reply)
-        except ValueError:
-            log.info("extract: non-JSON reply %r", reply[:60])
+        except (ValueError, TypeError):
+            log.info("extract: non-JSON reply %.60r", reply)
             return None
         errors = validate(obj, schema)
         if errors or not isinstance(obj, dict):

src/pfs/runlog.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+import logging
 import os
 from collections.abc import Callable, Iterable
 from dataclasses import asdict, dataclass
@@ -16,6 +17,8 @@ from pathlib import Path
 
 from conf import ROOT
 
+log = logging.getLogger(__name__)
+
 Unit = tuple[str, str, Callable[[], dict | None]]
 
 
@@ -42,29 +45,42 @@ def ledger_path(task: str, family: str) -> Path:
 
 
 class RunLog:
+    """Append-only JSON-lines ledger. The file is read lazily, once per
+    instance, into an index; an instance must not be shared across processes
+    (another process's appends are not seen). A truncated or malformed line
+    (e.g. a crash mid-write) is skipped with a warning, never fatal."""
+
     def __init__(self, path: Path) -> None:
         self.path = path
+        self._entries: list[Entry] | None = None
+        self._index: dict[tuple[str, str, str, str], Entry] = {}
 
-    def _entries(self) -> list[Entry]:
-        if not self.path.exists():
-            return []
-        out = []
-        for line in self.path.read_text().splitlines():
-            if line.strip():
-                out.append(Entry(**json.loads(line)))
-        return out
+    def _load(self) -> list[Entry]:
+        if self._entries is None:
+            self._entries = []
+            if self.path.exists():
+                text = self.path.read_text(encoding="utf-8")
+                for n, line in enumerate(text.splitlines(), 1):
+                    if not line.strip():
+                        continue
+                    try:
+                        entry = Entry(**json.loads(line))
+                    except (ValueError, TypeError):
+                        log.warning(
+                            "%s:%d: skipping unreadable ledger line", self.path, n
+                        )
+                        continue
+                    self._add(entry)
+        return self._entries
+
+    def _add(self, entry: Entry) -> None:
+        assert self._entries is not None
+        self._entries.append(entry)
+        self._index[(entry.unit, entry.input_hash, entry.model, entry.version)] = entry
 
     def get(self, unit: str, input_hash: str, model: str, version: str) -> Entry | None:
-        hit = None
-        for e in self._entries():
-            if (e.unit, e.input_hash, e.model, e.version) == (
-                unit,
-                input_hash,
-                model,
-                version,
-            ):
-                hit = e
-        return hit
+        self._load()
+        return self._index.get((unit, input_hash, model, version))
 
     def put(
         self,
@@ -74,19 +90,27 @@ class RunLog:
         version: str,
         output: dict | None,
     ) -> Entry:
+        self._load()
         entry = Entry(
             unit, input_hash, model, version, output, datetime.now(UTC).isoformat()
         )
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        with self.path.open("a") as f:
-            f.write(json.dumps(asdict(entry), sort_keys=True) + "\n")
+        lead = ""
+        if self.path.exists() and self.path.stat().st_size:
+            with self.path.open("rb") as f:
+                f.seek(-1, os.SEEK_END)
+                if f.read(1) != b"\n":
+                    lead = "\n"
+        with self.path.open("a", encoding="utf-8") as f:
+            f.write(lead + json.dumps(asdict(entry), sort_keys=True) + "\n")
             f.flush()
             os.fsync(f.fileno())
+        self._add(entry)
         return entry
 
     def outputs(self, model: str, version: str) -> dict[str, dict]:
         out: dict[str, dict] = {}
-        for e in self._entries():
+        for e in self._load():
             if e.model == model and e.version == version and e.output is not None:
                 out[e.unit] = e.output
         return out
@@ -102,13 +126,23 @@ def run_units(
     retry_failed: bool = False,
     progress: Callable[[str, str], None] | None = None,
 ) -> dict[str, int]:
-    counts = {"done": 0, "skipped": 0, "failed": 0}
+    """Run each unit not already logged for (unit, hash, model, version).
+
+    Counts: ``done`` (new non-None output), ``failed`` (new None output,
+    logged), ``skipped`` (already logged with output), ``failed_pending``
+    (already logged as failed and not retried; with ``retry_failed`` it is
+    re-run instead). ``limit`` caps new model calls; once it is hit the
+    remaining units are neither run nor counted."""
+    counts = {"done": 0, "skipped": 0, "failed": 0, "failed_pending": 0}
     calls = 0
     for unit, h, fn in units:
         prior = log.get(unit, h, model, version)
-        if prior is not None and (prior.output is not None or not retry_failed):
+        if prior is not None and prior.output is not None:
             counts["skipped"] += 1
             continue
+        if prior is not None and not retry_failed:
+            counts["failed_pending"] += 1
+            continue
         if limit and calls >= limit:
             break
         calls += 1

tests/llm/test_structured.py

@@ -148,3 +148,10 @@ def test_validate_types_and_arrays():
     assert len(errs) == 2 and "more than 2" in errs[0] and "[1]" in errs[1]
     assert validate([1, 2], {"items": {"type": "string"}})
     assert validate("x", {"type": "array"})
+
+
+def test_extractor_non_string_content_is_none():
+    ask = json_extractor(
+        Cfg(), Pool(["h"]), post=lambda url, json=None: {"message": {"content": None}}
+    )
+    assert ask("s", "u", SCHEMA) is None

tests/pfs/test_runlog.py

@@ -30,10 +30,16 @@ def test_runlog_resume_skips_logged_units(tmp_path):
         version="v1",
         progress=lambda u, s: seen.append(s),
     )
-    assert r1 == {"done": 3, "skipped": 0, "failed": 0} and seen == ["done"] * 3
+    assert (
+        r1 == {"done": 3, "skipped": 0, "failed": 0, "failed_pending": 0}
+        and seen == ["done"] * 3
+    )
     calls.clear()
     r2 = run_units(_units(calls), log, model="m", version="v1")
-    assert r2 == {"done": 0, "skipped": 3, "failed": 0} and calls == []
+    assert (
+        r2 == {"done": 0, "skipped": 3, "failed": 0, "failed_pending": 0}
+        and calls == []
+    )
 
 
 def test_runlog_new_hash_reruns(tmp_path):
@@ -61,10 +67,15 @@ def test_failed_units_retry_only_when_asked(tmp_path):
         version="v",
         progress=lambda u, s: seen.append((u, s)),
     )
-    assert r == {"done": 2, "skipped": 0, "failed": 1} and ("u1", "failed") in seen
+    assert (
+        r == {"done": 2, "skipped": 0, "failed": 1, "failed_pending": 0}
+        and ("u1", "failed") in seen
+    )
     calls.clear()
     r = run_units(_units(calls, outs), log, model="m", version="v")
-    assert r == {"done": 0, "skipped": 3, "failed": 0} and calls == []
+    assert (
+        r == {"done": 0, "skipped": 2, "failed": 0, "failed_pending": 1} and calls == []
+    )
     r = run_units(
         _units(calls, ({"v": 1}, {"v": 2}, {"v": 3})),
         log,
@@ -72,7 +83,12 @@ def test_failed_units_retry_only_when_asked(tmp_path):
         version="v",
         retry_failed=True,
     )
-    assert r == {"done": 1, "skipped": 2, "failed": 0} and calls == [1]
+    assert r == {
+        "done": 1,
+        "skipped": 2,
+        "failed": 0,
+        "failed_pending": 0,
+    } and calls == [1]
     assert log.get("u1", "h2", "m", "v").output == {"v": 2}
     assert log.outputs("m", "v") == {"u0": {"v": 1}, "u1": {"v": 2}, "u2": {"v": 3}}
 
@@ -94,7 +110,12 @@ def test_limit_caps_new_calls(tmp_path):
     assert r["done"] == 2 and calls == [0, 1]
     calls.clear()
     r = run_units(_units(calls), log, model="m", version="v", limit=2)
-    assert r == {"done": 1, "skipped": 2, "failed": 0} and calls == [2]
+    assert r == {
+        "done": 1,
+        "skipped": 2,
+        "failed": 0,
+        "failed_pending": 0,
+    } and calls == [2]
 
 
 def test_put_appends_json_lines_and_blank_lines_ignored(tmp_path):
@@ -120,3 +141,19 @@ def test_ledger_path_under_data():
     assert ledger_path("origin", "HOME-VISITS") == (
         ROOT / "data/pfs/ledger" / "origin-HOME-VISITS.jsonl"
     )
+
+
+def test_truncated_and_malformed_lines_are_skipped_and_put_recovers(tmp_path, caplog):
+    p = tmp_path / "l.jsonl"
+    RunLog(p).put("a", "h", "m", "v", {"k": 1})
+    with p.open("a") as f:
+        f.write('\n{"unit": "x", "missing": 1}\n')  # blank + valid JSON, wrong keys
+        f.write('{"unit": "b", "input_ha')  # cut mid-write, no newline
+    log = RunLog(p)
+    with caplog.at_level("WARNING"):
+        assert log.outputs("m", "v") == {"a": {"k": 1}}
+    assert caplog.text.count("skipping unreadable") == 2
+    log.put("c", "h", "m", "v", {"k": 3})
+    fresh = RunLog(p)
+    assert fresh.outputs("m", "v") == {"a": {"k": 1}, "c": {"k": 3}}
+    assert fresh.get("c", "h", "m", "v").output == {"k": 3}