fix: an unreadable file must not read as an absent one

postpeer_pilot/config.py

@@ -12,6 +12,7 @@
   scheduled.jsonl   local ledger of what this tool scheduled (series caps)
 """
 import json
+from . import safe_read
 import os
 from pathlib import Path
 
@@ -51,17 +52,17 @@
 def load() -> dict:
     cfg = dict(DEFAULTS)
     f = HOME / "config.json"
-    if f.exists():
-        cfg.update(json.loads(f.read_text()))
+    raw = safe_read.read_text_if_present(f)
+    if raw is not None:
+        cfg.update(json.loads(raw))
     return cfg
 
 
 def api_key() -> str:
     if os.environ.get("POSTPEER_API_KEY"):
         return os.environ["POSTPEER_API_KEY"]
     env = HOME / ".env"
-    if env.exists():
-        for line in env.read_text().splitlines():
+    for line in safe_read.read_lines_if_present(env):
             if line.startswith("POSTPEER_API_KEY="):
                 return line.split("=", 1)[1].strip()
     raise RuntimeError(f"POSTPEER_API_KEY not set (env var or {env})")

postpeer_pilot/perf.py

@@ -20,7 +20,7 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
-from . import config
+from . import config, safe_read
 
 STORE = config.HOME / "performance.jsonl"
 
@@ -42,10 +42,8 @@ def _append(rows: list):
 
 def latest() -> list:
     """Latest snapshot per (source, title): [{'title', 'views', 'source'}]."""
-    if not STORE.exists():
-        return []
     best = {}
-    for ln in STORE.read_text().splitlines():
+    for ln in safe_read.read_lines_if_present(STORE):
         try:
             r = json.loads(ln)
             best[(r.get("source"), r["title"])] = r    # append-only: later line wins
@@ -128,8 +126,7 @@ def pull_meta(limit: int = 50) -> int:
         return 0
     token = None
     env = Path(m.get("env", "")).expanduser()
-    if env.exists():
-        for ln in env.read_text().splitlines():
+    for ln in safe_read.read_lines_if_present(env):
             if ln.startswith("META_PAGE_TOKEN="):
                 token = ln.split("=", 1)[1].strip()
     if not token:

postpeer_pilot/plan.py

@@ -7,7 +7,7 @@
 from collections import defaultdict
 from datetime import date, datetime, timedelta
 
-from . import api, config
+from . import api, config, safe_read
 
 PLAN_FILE = config.HOME / "plan.json"
 LEDGER = config.HOME / "scheduled.jsonl"
@@ -19,8 +19,9 @@ def active() -> dict:
     plan = {"per_day_by_wd": {int(k): int(v) for k, v in cfg["per_day_by_wd"].items()},
             "slot_hours": list(cfg["slot_hours"]),
             "series_day_cap": int(cfg["series_day_cap"]), "source": "config defaults"}
-    if PLAN_FILE.exists():
-        p = json.loads(PLAN_FILE.read_text())
+    raw = safe_read.read_text_if_present(PLAN_FILE)
+    if raw is not None:
+        p = json.loads(raw)
         plan.update({"per_day_by_wd": {int(k): int(v) for k, v in p["per_day_by_wd"].items()},
                      "slot_hours": list(p.get("slot_hours", plan["slot_hours"])),
                      "source": "plan.json"})
@@ -49,8 +50,7 @@ def _series_by_day(exclude_slots: set | None = None) -> dict:
     handed out AND ledger-recorded in the same run must not count twice toward the cap."""
     out = defaultdict(list)
     exclude_slots = exclude_slots or set()
-    if LEDGER.exists():
-        for ln in LEDGER.read_text().splitlines():
+    for ln in safe_read.read_lines_if_present(LEDGER):
             try:
                 r = json.loads(ln)
                 if r.get("series") and r["when"] not in exclude_slots:
@@ -70,10 +70,8 @@ def record(when: str, video: str, series: str | None, post_id: str = "",
 
 
 def ledger_entries() -> list:
-    if not LEDGER.exists():
-        return []
     out = []
-    for ln in LEDGER.read_text().splitlines():
+    for ln in safe_read.read_lines_if_present(LEDGER):
         try:
             out.append(json.loads(ln))
         except json.JSONDecodeError:
@@ -86,8 +84,7 @@ def ledger_by_post_id() -> dict:
     performance matching: a published post whose id is in here needs no fuzzy text
     match to know its canonical caption."""
     out = {}
-    if LEDGER.exists():
-        for ln in LEDGER.read_text().splitlines():
+    for ln in safe_read.read_lines_if_present(LEDGER):
             try:
                 r = json.loads(ln)
                 if r.get("post_id"):

postpeer_pilot/safe_read.py

@@ -0,0 +1,44 @@
+"""Absent and unreadable are not the same thing.
+
+`Path.exists()` returns **False** when the OS refuses access — macOS TCC, unix
+permissions, an unmounted network share. Used as a guard before reading a file,
+it silently turns "I am not allowed to read this" into "this does not exist",
+and the caller falls back to defaults without anyone noticing.
+
+For a publishing tool that failure is not cosmetic. A plan file that cannot be
+read looks like *no plan*, so the scheduler quietly books against built-in
+defaults; a store that cannot be read looks like *no history*, so the damped
+planner sees an empty performance record. Both produce confident, wrong output.
+
+The rule this module enforces: a read either returns content, returns ``None``
+because the file genuinely is not there, or raises. It never degrades quietly.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+
+class Unreadable(OSError):
+    """The path exists (or its status is unknowable) but could not be read."""
+
+
+def read_text_if_present(path: Path) -> str | None:
+    """Return file contents, or ``None`` if the file genuinely does not exist.
+
+    Raises :class:`Unreadable` for every other failure — permission denied,
+    unreadable mount, a directory where a file was expected. Callers that want
+    a default must choose it explicitly after catching, so the fallback is a
+    decision in the code rather than an accident of the filesystem.
+    """
+    try:
+        return Path(path).read_text()
+    except FileNotFoundError:
+        return None
+    except OSError as exc:                       # PermissionError, NotADirectoryError, EIO …
+        raise Unreadable(f"{path} exists but could not be read: {exc}") from exc
+
+
+def read_lines_if_present(path: Path) -> list[str]:
+    """``read_text_if_present`` split into lines; ``[]`` when the file is absent."""
+    raw = read_text_if_present(path)
+    return raw.splitlines() if raw is not None else []

tests/test_safe_read.py

@@ -0,0 +1,67 @@
+"""Unreadable must never look like absent.
+
+Regression for the failure found on 2026-07-20: `queue_status` reported an empty
+review folder while nine finished videos sat in it. The guard was
+`if PATH.exists()` — and `exists()` returns False when the OS denies access, so a
+permission problem was indistinguishable from "nothing there". Anything that
+schedules against such a reading books work it cannot see.
+"""
+import json
+import os
+import stat
+
+import pytest
+
+from postpeer_pilot import config, plan, safe_read
+
+
+def test_absent_file_reads_as_none(tmp_path):
+    assert safe_read.read_text_if_present(tmp_path / "nope.json") is None
+    assert safe_read.read_lines_if_present(tmp_path / "nope.jsonl") == []
+
+
+def test_present_file_reads_through(tmp_path):
+    f = tmp_path / "there.txt"
+    f.write_text("a\nb\n")
+    assert safe_read.read_text_if_present(f) == "a\nb\n"
+    assert safe_read.read_lines_if_present(f) == ["a", "b"]
+
+
+@pytest.mark.skipif(os.geteuid() == 0, reason="root bypasses file permissions")
+def test_unreadable_file_raises_instead_of_defaulting(tmp_path):
+    f = tmp_path / "locked.json"
+    f.write_text("{}")
+    f.chmod(0)
+    try:
+        with pytest.raises(safe_read.Unreadable):
+            safe_read.read_text_if_present(f)
+    finally:
+        f.chmod(stat.S_IRUSR | stat.S_IWUSR)
+
+
+@pytest.mark.skipif(os.geteuid() == 0, reason="root bypasses file permissions")
+def test_unreadable_plan_does_not_silently_fall_back(tmp_path, monkeypatch):
+    """The important one: an unreadable plan must NOT masquerade as 'no plan',
+    because the scheduler would then quietly book against built-in defaults."""
+    pf = tmp_path / "plan.json"
+    pf.write_text(json.dumps({"per_day_by_wd": {str(i): 1 for i in range(7)}}))
+    pf.chmod(0)
+    monkeypatch.setattr(plan, "PLAN_FILE", pf)
+    try:
+        with pytest.raises(safe_read.Unreadable):
+            plan.active()
+    finally:
+        pf.chmod(stat.S_IRUSR | stat.S_IWUSR)
+
+
+@pytest.mark.skipif(os.geteuid() == 0, reason="root bypasses file permissions")
+def test_unreadable_config_does_not_silently_fall_back(tmp_path, monkeypatch):
+    f = tmp_path / "config.json"
+    f.write_text("{}")
+    f.chmod(0)
+    monkeypatch.setattr(config, "HOME", tmp_path)
+    try:
+        with pytest.raises(safe_read.Unreadable):
+            config.load()
+    finally:
+        f.chmod(stat.S_IRUSR | stat.S_IWUSR)