From 8e194d26137e4e95163603019ce93abd826bdbe4 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 10 Aug 2026 15:32:53 +0200 Subject: [PATCH 1/2] fix(daemon): recover typed CAS frontier failures Problem: stopped-daemon raw recovery ignored canonical and historical CAS frontier evidence even though the production repair selector accepted both as replay authority. What changed: correlate the daemon probe with the exact raw-artifact coordinate and existing typed replay-authority vocabulary. Add a temporary-SQLite route test that persists the failure through ArchiveStore and drains it through the daemon recovery stage and repair engine. Ref polylogue-dyica and polylogue-dyica.1. Co-authored-by: Codex --- polylogue/daemon/convergence_stages.py | 22 ++++++++- tests/unit/daemon/test_raw_parse_recovery.py | 48 ++++++++++++++++++++ 2 files changed, 69 insertions(+), 1 deletion(-) diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index 6ef02fd410..f8b8cca962 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: @@ -924,10 +929,25 @@ def _raw_parse_recovery_pending_count(db_path: Path, path: Path, *, archive_root 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..6046b05dc2 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,52 @@ 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_source_open_failure_is_failed_and_retryable( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: From 08c363b0de3160234732f060e1791ad0e84dff6b Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 10 Aug 2026 16:14:42 +0200 Subject: [PATCH 2/2] fix(daemon): align CAS recovery eligibility Problem: The stopped-daemon pending probe could retain validation-failed CAS rows and skip previously parsed, unmaterialized CAS rows, unlike the repair selector it triggers. What changed: Mirror failed-validation eligibility and separate unparsed legacy retries from typed CAS retry authority. File-backed regressions cover both routes. Ref polylogue-dyica. Co-Authored-By: Codex --- polylogue/daemon/convergence_stages.py | 15 ++++--- tests/unit/daemon/test_raw_parse_recovery.py | 41 ++++++++++++++++++++ 2 files changed, 51 insertions(+), 5 deletions(-) diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index f8b8cca962..09c5d7b8c3 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/polylogue/daemon/convergence_stages.py @@ -923,12 +923,17 @@ 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 diff --git a/tests/unit/daemon/test_raw_parse_recovery.py b/tests/unit/daemon/test_raw_parse_recovery.py index 6046b05dc2..513c3b0ac6 100644 --- a/tests/unit/daemon/test_raw_parse_recovery.py +++ b/tests/unit/daemon/test_raw_parse_recovery.py @@ -299,6 +299,47 @@ def test_raw_parse_recovery_stage_drains_typed_cas_frontier_failure( 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: