diff --git a/loopx/control_plane/post_writeback_composition_retry.py b/loopx/control_plane/post_writeback_composition_retry.py index e10f9f5de2..f2fdffab4a 100644 --- a/loopx/control_plane/post_writeback_composition_retry.py +++ b/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 diff --git a/tests/control_plane/test_post_writeback_composition_retry.py b/tests/control_plane/test_post_writeback_composition_retry.py index 7abf9694ce..4456e9fcad 100644 --- a/tests/control_plane/test_post_writeback_composition_retry.py +++ b/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: diff --git a/tests/control_plane/test_status_receipt_retry_projection.py b/tests/control_plane/test_status_receipt_retry_projection.py index 1c686067bf..955ee4c190 100644 --- a/tests/control_plane/test_status_receipt_retry_projection.py +++ b/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"]