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

`Path.exists()` returns False when the OS denies access — macOS TCC, unix
permissions, an unmounted share. Used as a guard before a read, it turns
"I may not read this" into "this does not exist" and the caller falls back
to defaults with nothing in the output to say so.

Found in the field on 2026-07-20: a sibling deployment reported an empty
review folder while nine finished videos sat in it, because the folder scan
was guarded by `exists()` and the process had lost Documents access. The
number 0 was indistinguishable from a real empty folder. Acting on it would
have meant reporting "nothing to schedule" against a full queue.

The same guard sits in front of this repo's own state: plan.json, the damping
ledger, the performance store and config.json. An unreadable plan looks like
no plan, so the scheduler books against built-in defaults; an unreadable store
looks like no history, so the damped planner sees an empty record. Both give
confident, wrong output.

Adds postpeer_pilot/safe_read.py: a read either returns content, returns None
because the file genuinely is absent, or raises Unreadable. Callers that want
a default now choose it explicitly, so the fallback is a decision in the code
rather than an accident of the filesystem.

Applied to plan.active, plan.series_slots, plan.ledger_entries,
plan.captions_by_post, perf.latest, perf meta-token lookup, config.load and
config.api_key. Regression tests cover absent, present and chmod-000 paths,
including that an unreadable plan or config raises instead of defaulting.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: benmfzen

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)