fix(daemon): recover typed CAS frontier failures

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:

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: