diff --git a/devtools/preflight_ledger.py b/devtools/preflight_ledger.py index 06c265dc2d..67dc5ac0bc 100644 --- a/devtools/preflight_ledger.py +++ b/devtools/preflight_ledger.py @@ -114,7 +114,7 @@ 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 @@ -122,7 +122,7 @@ def _source_distribution(root: Path) -> dict[str, object]: 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,7 +365,14 @@ 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=( @@ -373,11 +380,14 @@ def _replay_preflight(root: Path, *, limit: int) -> dict[str, object]: 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, ) diff --git a/polylogue/storage/raw_failure_lifecycle.py b/polylogue/storage/raw_failure_lifecycle.py index 0c94c8820f..01a0034ab7 100644 --- a/polylogue/storage/raw_failure_lifecycle.py +++ b/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' ) """ diff --git a/polylogue/storage/usage.py b/polylogue/storage/usage.py index 244e1ce9fa..a1db0798c6 100644 --- a/polylogue/storage/usage.py +++ b/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 """, diff --git a/tests/unit/devtools/test_preflight_ledger.py b/tests/unit/devtools/test_preflight_ledger.py index c64b2f095f..b07a7244d7 100644 --- a/tests/unit/devtools/test_preflight_ledger.py +++ b/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: diff --git a/tests/unit/storage/test_origin_usage_report.py b/tests/unit/storage/test_origin_usage_report.py index ea2939aaee..ec8a6d5f9f 100644 --- a/tests/unit/storage/test_origin_usage_report.py +++ b/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