diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index 6ef02fd410..09c5d7b8c3 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/polylogue/daemon/convergence_stages.py @@ -22,6 +22,10 @@ from polylogue.config import load_polylogue_config from polylogue.core.enums import Provider +from polylogue.core.raw_failure_evidence import ( + RAW_FAILURE_DEFERRED_SUPPORT_STATUS, + RAW_FAILURE_REPLAY_AUTHORITY_EVIDENCE_KINDS, +) from polylogue.daemon.convergence import ConvergenceStage, StageExecuteReturn, StageExecutionResult from polylogue.daemon.convergence_standing_queries import make_standing_query_stage from polylogue.logging import get_logger @@ -884,6 +888,7 @@ def _raw_parse_recovery_pending_count(db_path: Path, path: Path, *, archive_root return 0 index_db = ArchiveLocation.resolve(durable_root).active_index_path normalized_root = str(path).rstrip("/") + replay_authority_placeholders = ", ".join("?" for _ in RAW_FAILURE_REPLAY_AUTHORITY_EVIDENCE_KINDS) try: conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=5.0) except sqlite3.Error: @@ -918,16 +923,36 @@ def _raw_parse_recovery_pending_count(db_path: Path, path: Path, *, archive_root FROM raw_sessions AS r {materialized_join} WHERE (r.source_path = ? OR r.source_path LIKE ?) - AND r.parsed_at_ms IS NULL + AND COALESCE(r.validation_status, '') != 'failed' AND ( - r.parse_error IS NULL - OR r.parse_error = 'OperationalError: database is locked' - OR r.parse_error LIKE 'decode:%No such file or directory:%' - OR r.parse_error LIKE 'membership_replay_conflict:%' + ( + r.parsed_at_ms IS NULL + AND ( + r.parse_error IS NULL + OR r.parse_error = 'OperationalError: database is locked' + OR r.parse_error LIKE 'decode:%No such file or directory:%' + OR r.parse_error LIKE 'membership_replay_conflict:%' + ) + ) + OR EXISTS ( + SELECT 1 + FROM raw_artifacts AS failure_evidence + WHERE failure_evidence.raw_id IS r.raw_id + AND failure_evidence.origin IS r.origin + AND failure_evidence.source_path IS r.source_path + AND failure_evidence.source_index IS r.source_index + AND failure_evidence.artifact_kind IN ({replay_authority_placeholders}) + AND failure_evidence.support_status = ? + ) ) {materialized_where} """, - (normalized_root, f"{normalized_root}/%"), + ( + normalized_root, + f"{normalized_root}/%", + *sorted(RAW_FAILURE_REPLAY_AUTHORITY_EVIDENCE_KINDS), + RAW_FAILURE_DEFERRED_SUPPORT_STATUS, + ), ).fetchone() return int(row[0] or 0) if row is not None else 0 except sqlite3.Error: diff --git a/tests/unit/daemon/test_raw_parse_recovery.py b/tests/unit/daemon/test_raw_parse_recovery.py index 604168b01b..513c3b0ac6 100644 --- a/tests/unit/daemon/test_raw_parse_recovery.py +++ b/tests/unit/daemon/test_raw_parse_recovery.py @@ -29,6 +29,8 @@ import pytest from polylogue.core.enums import Provider +from polylogue.core.errors import RawCASFrontierError +from polylogue.core.raw_failure_evidence import RawFailureEvidenceKind from polylogue.daemon.convergence import DaemonConverger, StageState from polylogue.daemon.convergence_stages import make_raw_parse_recovery_stage from polylogue.sources.live.cursor import CursorStore @@ -251,6 +253,93 @@ def test_raw_parse_recovery_retries_authorized_parse_failure(tmp_path: Path) -> assert stage.check(path) is True +@pytest.mark.parametrize( + "evidence_kind", + [ + RawFailureEvidenceKind.DEFERRED_CAS_FRONTIER, + RawFailureEvidenceKind.DEFERRED_CODEX_CAS_FRONTIER, + ], +) +def test_raw_parse_recovery_stage_drains_typed_cas_frontier_failure( + tmp_path: Path, evidence_kind: RawFailureEvidenceKind +) -> None: + """A stopped daemon requeues canonical and historical CAS retry authority.""" + initialize_active_archive_root(tmp_path) + path = tmp_path / "cas-frontier.json" + raw_id = _write_stuck_raw(tmp_path, source_path=str(path)) + + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + if evidence_kind is RawFailureEvidenceKind.DEFERRED_CAS_FRONTIER: + archive.mark_raw_parse_failed( + raw_id, + provider=Provider.CHATGPT, + error=RawCASFrontierError("frontier changed while daemon stopped"), + ) + else: + archive.record_raw_failure_evidence( + raw_id, + provider=Provider.CHATGPT, + source_path=str(path), + source_index=0, + acquired_at_ms=1, + kind=evidence_kind, + ) + archive.mark_raw_parse_failed( + raw_id, + provider=Provider.CHATGPT, + error=ValueError("historical CAS frontier failure"), + preserve_existing_failure_evidence=True, + ) + + stage = make_raw_parse_recovery_stage(tmp_path / "index.db") + + assert stage.check(path) is True + assert stage.execute(path) is True + assert stage.check(path) is False + assert _sessions_for_raw(tmp_path, raw_id) == [("conv-stuck", raw_id)] + + +def test_raw_parse_recovery_skips_validation_failed_cas_frontier_failure(tmp_path: Path) -> None: + """A failed validation cannot keep CAS recovery debt pending forever.""" + initialize_active_archive_root(tmp_path) + path = tmp_path / "validation-failed-cas-frontier.json" + raw_id = _write_stuck_raw(tmp_path, source_path=str(path)) + + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + archive.mark_raw_parse_failed( + raw_id, + provider=Provider.CHATGPT, + error=RawCASFrontierError("frontier changed after validation failed"), + ) + with sqlite3.connect(tmp_path / "source.db") as conn: + conn.execute("UPDATE raw_sessions SET validation_status = 'failed' WHERE raw_id = ?", (raw_id,)) + conn.commit() + + assert make_raw_parse_recovery_stage(tmp_path / "index.db").check(path) is False + + +def test_raw_parse_recovery_drains_previously_parsed_cas_frontier_failure(tmp_path: Path) -> None: + """CAS authority replays an unmaterialized raw even when parsing had completed.""" + initialize_active_archive_root(tmp_path) + path = tmp_path / "previously-parsed-cas-frontier.json" + raw_id = _write_stuck_raw(tmp_path, source_path=str(path)) + + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + archive.mark_raw_parse_succeeded(raw_id, provider=Provider.CHATGPT) + archive.mark_raw_parse_failed( + raw_id, + provider=Provider.CHATGPT, + error=RawCASFrontierError("frontier changed after parsing completed"), + ) + + stage = make_raw_parse_recovery_stage(tmp_path / "index.db") + + assert stage.check(path) is True + assert stage.execute(path) is True + assert stage.check(path) is False + assert _sessions_for_raw(tmp_path, raw_id) == [("conv-stuck", raw_id)] + + def test_raw_parse_recovery_source_open_failure_is_failed_and_retryable( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: