fix(preflight): classify blocked replay evidence

devtools/preflight_ledger.py

@@ -114,15 +114,15 @@ def _source_distribution(root: Path) -> dict[str, object]:
                            c.status AS census_status,
                            CASE WHEN c.raw_id IS NULL THEN 1 ELSE 0 END AS coverage_unknown,
                            CASE WHEN c.status IN ('failed', 'non_session') THEN 1 ELSE 0 END AS terminal,
-                           CASE WHEN (r.parse_error IS NOT NULL AND TRIM(r.parse_error) != '')
+                           CASE WHEN r.parse_error IS NOT NULL
                                   OR LOWER(COALESCE(r.validation_status, '')) = 'failed'
                                 THEN 1 ELSE 0 END AS failure,
                            CASE WHEN c.status = 'complete' AND c.member_count > 0 THEN 1 ELSE 0 END AS census_eligible
                     FROM raw_sessions AS r
                     LEFT JOIN raw_membership_census AS c ON c.raw_id = r.raw_id
                 )
                 SELECT origin, COUNT(*), COALESCE(SUM(blob_size), 0),
-                       COALESCE(SUM(parse_error IS NOT NULL AND TRIM(parse_error) != ''), 0),
+                       COALESCE(SUM(parse_error IS NOT NULL), 0),
                        COALESCE(SUM(validation_status = 'failed'), 0),
                        COALESCE(SUM(revision_authority = 'quarantined'), 0),
                        COALESCE(SUM(CASE WHEN revision_authority = 'quarantined' THEN blob_size ELSE 0 END), 0),
@@ -149,7 +149,7 @@ def _source_distribution(root: Path) -> dict[str, object]:
             totals = conn.execute(
                 """
                 SELECT COUNT(*), COALESCE(SUM(blob_size), 0),
-                       COALESCE(SUM(parse_error IS NOT NULL AND TRIM(parse_error) != ''), 0),
+                       COALESCE(SUM(parse_error IS NOT NULL), 0),
                        COALESCE(SUM(LOWER(COALESCE(validation_status, '')) = 'failed'), 0),
                        COALESCE(SUM(revision_authority = 'quarantined'), 0),
                        COALESCE(SUM(CASE WHEN revision_authority = 'quarantined' THEN blob_size ELSE 0 END), 0),
@@ -365,19 +365,29 @@ def _replay_preflight(root: Path, *, limit: int) -> dict[str, object]:
         )
     candidate_count = _count(payload.get("candidate_count"))
     blocked_count = _count(payload.get("blocked_candidate_count"))
-    state = "fail" if candidate_count else "warn" if blocked_count else "pass"
+    executable_component_count = _count(payload.get("executable_authority_component_count"))
+    # ``candidate_count`` counts raw rows, while ``blocked_candidate_count``
+    # includes authority/resource debt.  The backlog already computes the
+    # executable authority-component population, so use that typed relation
+    # to distinguish executable work from blocked-only work.
+    state = (
+        "fail" if executable_component_count else "warn" if blocked_count else "unknown" if candidate_count else "pass"
+    )
     return _status(
         state=state,
         reason=(
             "executable raw replay candidates remain"
             if state == "fail"
             else "raw replay candidates are authority/resource blocked"
             if state == "warn"
+            else "raw replay candidates lack executable or blocked classification"
+            if state == "unknown"
             else None
         ),
         available=True,
         candidate_count=candidate_count,
         blocked_candidate_count=blocked_count,
+        executable_authority_component_count=executable_component_count,
         authority_quarantined_count=_count(payload.get("authority_quarantined_count")),
         evidence=payload,
     )

polylogue/storage/raw_failure_lifecycle.py

@@ -120,10 +120,7 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra
         if raw_table is None:
             return RawFailureLifecycleSnapshot(False, reason="source.db is missing raw_sessions")
         parse_failures = int(
-            conn.execute(
-                "SELECT COUNT(*) FROM raw_sessions WHERE parse_error IS NOT NULL AND TRIM(parse_error) != ''"
-            ).fetchone()[0]
-            or 0
+            conn.execute("SELECT COUNT(*) FROM raw_sessions WHERE parse_error IS NOT NULL").fetchone()[0] or 0
         )
         validation_failures = int(
             conn.execute("SELECT COUNT(*) FROM raw_sessions WHERE validation_status = 'failed'").fetchone()[0] or 0
@@ -138,7 +135,7 @@ def read_raw_failure_lifecycle(source_db: Path, *, sample_limit: int = 10) -> Ra
             SELECT r.raw_id, r.origin, r.source_path, r.source_index,
                    r.validation_status, r.acquired_at_ms
             FROM raw_sessions AS r
-            WHERE (r.parse_error IS NOT NULL AND TRIM(r.parse_error) != '')
+            WHERE r.parse_error IS NOT NULL
                OR r.validation_status = 'failed'
         )
         """

polylogue/storage/usage.py

@@ -1196,11 +1196,10 @@ def _source_raw_stats(
             SELECT r.origin AS origin,
                    COUNT(*) AS raw_session_count,
                    COALESCE(SUM(CASE
-                       WHEN r.parse_error IS NOT NULL AND TRIM(r.parse_error) != '' THEN 1 ELSE 0
+                       WHEN r.parse_error IS NOT NULL THEN 1 ELSE 0
                    END), 0) AS raw_parse_error_count,
                    COALESCE(SUM(CASE
-                       WHEN (r.parse_error IS NULL OR TRIM(r.parse_error) = '')
-                        AND s.session_id IS NULL THEN 1 ELSE 0
+                       WHEN r.parse_error IS NULL AND s.session_id IS NULL THEN 1 ELSE 0
                    END), 0) AS acquired_not_materialized_count
             FROM {alias_sql}.raw_sessions AS r
             LEFT JOIN sessions AS s ON s.raw_id = r.raw_id
@@ -1248,7 +1247,7 @@ def _acquired_not_materialized_raw_rows(
         LEFT JOIN sessions AS s ON s.raw_id = r.raw_id
         LEFT JOIN {alias_sql}.raw_membership_census AS c ON c.raw_id = r.raw_id
         {_where_origin(origin, table_alias="r")}
-          {"AND" if origin is not None else "WHERE"} (r.parse_error IS NULL OR TRIM(r.parse_error) = '')
+          {"AND" if origin is not None else "WHERE"} r.parse_error IS NULL
           AND s.session_id IS NULL
         ORDER BY r.origin, r.raw_id
         """,

tests/unit/devtools/test_preflight_ledger.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import sqlite3
+from collections.abc import Callable
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import cast
@@ -24,8 +25,24 @@ def _list(value: object) -> list[object]:
     return cast(list[object], value)
 
 
-def _mixed_replay_backlog(*_args: object, **_kwargs: object) -> dict[str, object]:
-    return {"available": True, "candidate_count": 2, "blocked_candidate_count": 5}
+def _replay_backlog(
+    candidate_count: int,
+    blocked_candidate_count: int,
+    executable_authority_component_count: int | None = None,
+) -> Callable[..., dict[str, object]]:
+    def backlog(*_args: object, **_kwargs: object) -> dict[str, object]:
+        return {
+            "available": True,
+            "candidate_count": candidate_count,
+            "blocked_candidate_count": blocked_candidate_count,
+            "executable_authority_component_count": (
+                candidate_count
+                if executable_authority_component_count is None
+                else executable_authority_component_count
+            ),
+        }
+
+    return backlog
 
 
 def _initialize_all_tiers(root: Path) -> None:
@@ -184,16 +201,60 @@ def test_preflight_fails_closed_on_missing_census_relation(tmp_path: Path) -> No
 def test_preflight_fails_when_executable_replay_candidates_coexist_with_blocked(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    monkeypatch.setattr(preflight_ledger, "raw_materialization_replay_backlog", _mixed_replay_backlog)
+    _initialize_all_tiers(tmp_path)
+    monkeypatch.setattr(preflight_ledger, "raw_materialization_replay_backlog", _replay_backlog(2, 5))
 
-    replay = preflight_ledger._replay_preflight(tmp_path, limit=10)
+    report = build_preflight_ledger(tmp_path, limit=10)
+    replay = _mapping(_mapping(report["checks"])["replay_backlog"])
 
     assert replay["state"] == "fail"
     assert replay["candidate_count"] == 2
     assert replay["blocked_candidate_count"] == 5
 
 
-def test_preflight_ignores_blank_parse_errors_in_origin_and_totals(tmp_path: Path) -> None:
+def test_preflight_fails_when_executable_replay_candidates_outnumber_blocked(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _initialize_all_tiers(tmp_path)
+    monkeypatch.setattr(preflight_ledger, "raw_materialization_replay_backlog", _replay_backlog(5, 2))
+
+    report = build_preflight_ledger(tmp_path, limit=10)
+    replay = _mapping(_mapping(report["checks"])["replay_backlog"])
+
+    assert replay["state"] == "fail"
+    assert replay["candidate_count"] == 5
+    assert replay["blocked_candidate_count"] == 2
+
+
+def test_preflight_warns_when_replay_candidates_are_all_resource_blocked(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _initialize_all_tiers(tmp_path)
+    monkeypatch.setattr(preflight_ledger, "raw_materialization_replay_backlog", _replay_backlog(5, 5, 0))
+
+    report = build_preflight_ledger(tmp_path, limit=10)
+    replay = _mapping(_mapping(report["checks"])["replay_backlog"])
+
+    assert replay["state"] == "warn"
+    assert replay["candidate_count"] == 5
+    assert replay["blocked_candidate_count"] == 5
+    assert replay["executable_authority_component_count"] == 0
+
+
+def test_preflight_is_unknown_when_replay_candidates_are_unclassified(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _initialize_all_tiers(tmp_path)
+    monkeypatch.setattr(preflight_ledger, "raw_materialization_replay_backlog", _replay_backlog(1, 0, 0))
+
+    report = build_preflight_ledger(tmp_path, limit=10)
+    replay = _mapping(_mapping(report["checks"])["replay_backlog"])
+
+    assert replay["state"] == "unknown"
+    assert report["ok"] is False
+
+
+def test_preflight_reports_every_non_null_raw_parse_error(tmp_path: Path) -> None:
     _initialize_all_tiers(tmp_path)
     _insert_raws(
         tmp_path,
@@ -224,9 +285,9 @@ def test_preflight_ignores_blank_parse_errors_in_origin_and_totals(tmp_path: Pat
     by_origin = {str(item["origin"]): item for item in (_mapping(value) for value in _list(source["by_origin"]))}
     origin = by_origin["codex-session"]
     eligibility = _mapping(origin["eligibility"])
-    assert totals["parse_failures"] == 1
-    assert origin["parse_failures"] == 1
-    assert eligibility["actionable_count"] == 1
+    assert totals["parse_failures"] == 3
+    assert origin["parse_failures"] == 3
+    assert eligibility["actionable_count"] == 3
 
 
 def test_preflight_blocks_unexplained_raw_failure_lifecycle(tmp_path: Path) -> None:

tests/unit/storage/test_origin_usage_report.py

@@ -453,6 +453,7 @@ def test_origin_usage_report_exposes_source_debt_and_stale_rollups(tmp_path: Pat
     _insert_raw_session(source_conn, raw_id="raw-materialized", native_id="provider-usage-report")
     _insert_raw_session(source_conn, raw_id="raw-missing", native_id="missing")
     _insert_raw_session(source_conn, raw_id="raw-error", native_id="bad", parse_error="bad json")
+    _insert_raw_session(source_conn, raw_id="raw-empty-error", native_id="empty", parse_error="")
     source_conn.commit()
     source_conn.close()
 
@@ -503,8 +504,8 @@ def test_origin_usage_report_exposes_source_debt_and_stale_rollups(tmp_path: Pat
 
     row = report.origins[0]
     assert row.coverage_state == "acquired_not_materialized"
-    assert row.raw_session_count == 3
-    assert row.raw_parse_error_count == 1
+    assert row.raw_session_count == 4
+    assert row.raw_parse_error_count == 2
     assert row.acquired_not_materialized_count == 1
     assert row.sample_acquired_not_materialized_raw_ids == ("raw-missing",)
     assert row.stale_rollup_session_count == 1