fix(status): fail closed on unreadable retry journals

loopx/control_plane/post_writeback_composition_retry.py

@@ -188,8 +188,12 @@ def _iter_composition_retry_rows(
 
     try:
         lines = log_path.read_text(encoding="utf-8").splitlines()
-    except OSError:
+    except FileNotFoundError:
         return []
+    except OSError as error:
+        raise RuntimeError(
+            "composition retry receipt journal is unreadable"
+        ) from error
     selected = lines if row_limit is None else lines[-max(0, row_limit) :]
     rows: list[dict[str, Any]] = []
     for line in selected:
@@ -456,8 +460,6 @@ def collect_pending_composition_retry_projection(
     pending: list[dict[str, Any]] = []
     total_matching = 0
     for journal_path in journal_paths:
-        if not journal_path.is_file():
-            continue
         for row in pending_composition_retry_receipts_for_path(journal_path):
             if (
                 selector_agent_id

tests/control_plane/test_post_writeback_composition_retry.py

@@ -538,6 +538,44 @@ def test_pending_receipts_skip_foreign_and_malformed_rows(
     ]
 
 
+def test_pending_receipts_fail_closed_when_journal_is_unreadable(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    runtime_root = tmp_path
+    journal_path = composition_retry_receipt_log_path(runtime_root, "goal-1")
+    receipt = build_composition_retry_receipt(
+        goal_id="goal-1",
+        event_kind="todo_complete",
+        identity=_identity(),
+        state_version="vision-revision-2",
+        committed_at="2026-09-06T00:00:00Z",
+        hook_identities=[{"hook_id": "h.one", "capability_id": "cap-one"}],
+        error_code="source_projection_failed",
+    )
+    append_composition_retry_receipt(journal_path, receipt)
+    assert pending_composition_retry_receipts(runtime_root, "goal-1") == [receipt]
+
+    real_read_text = Path.read_text
+
+    def deny_journal_read(
+        path: Path, *args: object, **kwargs: object
+    ) -> str:
+        if path == journal_path:
+            raise PermissionError("injected journal read denial")
+        return real_read_text(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_text", deny_journal_read)
+
+    with pytest.raises(
+        RuntimeError, match="composition retry receipt journal is unreadable"
+    ):
+        pending_composition_retry_receipts(runtime_root, "goal-1")
+    with pytest.raises(
+        RuntimeError, match="composition retry receipt journal is unreadable"
+    ):
+        collect_pending_composition_retry_projection(runtime_root, "goal-1")
+
+
 def test_journal_unavailable_degrades_to_identity_without_receipt_ref(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/control_plane/test_status_receipt_retry_projection.py

@@ -4,17 +4,28 @@
 from copy import deepcopy
 from datetime import UTC, datetime, timedelta
 import json
+import os
 from pathlib import Path
+import stat
 import subprocess
 import sys
 from threading import Thread
 from urllib.request import urlopen
 
 import pytest
-from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from canonical_authority_fixture import (
+    initialize_canonical_authority,
+    isolate_sqlite_runtime,
+    promoted_create_fixture,
+)
 
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 from loopx.control_plane.effect_runtime import restart_effect_runtime
+from loopx.control_plane.post_writeback_composition_retry import (
+    append_composition_retry_receipt,
+    build_composition_retry_receipt,
+    composition_retry_receipt_log_path,
+)
 from loopx.control_plane.quota.blocked_retry import overlay_status_turn_retries
 from loopx.status import active_state_todo_fields
 from loopx.status_server import (
@@ -168,3 +179,99 @@ def test_real_provider_status_projects_wait_without_writes(tmp_path: Path, monke
         assert not list(runtime.rglob("*spend*"))
     finally:
         restart_effect_runtime()
+
+
+@pytest.mark.skipif(
+    os.name == "nt" or getattr(os, "geteuid", lambda: -1)() == 0,
+    reason="requires POSIX directory permissions for an unprivileged user",
+)
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_real_provider_status_fails_closed_when_retry_journal_parent_is_inaccessible(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str
+) -> None:
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    goal_id = "goal-a"
+    registry, runtime, _ = promoted_create_fixture(
+        tmp_path,
+        provider=provider,
+    )
+    registry_payload = json.loads(registry.read_text(encoding="utf-8"))
+    registry_payload["goals"][0].update(
+        domain="software", adapter={"kind": "read_only_project_map_v0"}
+    )
+    registry.write_text(json.dumps(registry_payload), encoding="utf-8")
+    receipt = build_composition_retry_receipt(
+        goal_id=goal_id,
+        event_kind="todo_complete",
+        identity={
+            "agent_id": "agent-a",
+            "todo_id": "todo_work",
+            "turn_instance_id": "turn-a",
+            "effect_id": "goal-a:agent-a:todo_work:turn-a",
+        },
+        state_version="revision-a",
+        committed_at="2030-01-01T00:00:00Z",
+        hook_identities=[
+            {"hook_id": "h.one", "capability_id": "cap-one"}
+        ],
+        error_code="source_projection_failed",
+    )
+    journal_path = composition_retry_receipt_log_path(runtime, goal_id)
+    append_composition_retry_receipt(journal_path, receipt)
+
+    def run_status() -> tuple[subprocess.CompletedProcess[str], dict]:
+        result = subprocess.run(
+            [
+                sys.executable,
+                "-m",
+                "loopx.cli",
+                "--registry",
+                str(registry),
+                "--format",
+                "json",
+                "status",
+                "--goal-id",
+                goal_id,
+            ],
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+        return result, json.loads(result.stdout)
+
+    healthy_result, healthy_payload = run_status()
+    assert healthy_result.returncode == 0, healthy_payload
+    assert healthy_payload["ok"] is True
+    assert (
+        healthy_payload["pending_composition_retry_receipts"]["pending_count"]
+        == 1
+    )
+    assert healthy_payload["pending_composition_retry_receipts"]["pending"][0][
+        "receipt_id"
+    ] == receipt["receipt_id"]
+
+    journal_parent = journal_path.parent
+    original_mode = stat.S_IMODE(journal_parent.stat().st_mode)
+    try:
+        journal_parent.chmod(0)
+        denied_result, denied_payload = run_status()
+    finally:
+        journal_parent.chmod(original_mode)
+
+    assert denied_result.returncode == 1, denied_payload
+    assert denied_payload["ok"] is False
+    assert (
+        denied_payload["error"]
+        == "composition retry receipt journal is unreadable"
+    )
+    assert str(runtime) not in denied_payload["error"]
+    assert denied_payload["attention_queue"]["items"][0]["status"] == (
+        "status_collection_failed"
+    )
+
+    recovered_result, recovered_payload = run_status()
+    assert recovered_result.returncode == 0, recovered_payload
+    assert recovered_payload["ok"] is True
+    assert recovered_payload["pending_composition_retry_receipts"]["pending"][
+        0
+    ]["receipt_id"] == receipt["receipt_id"]