diff --git a/docs/plans/degrade-loudly-allowlist.yaml b/docs/plans/degrade-loudly-allowlist.yaml index 6545775139..980ecaf0d3 100644 --- a/docs/plans/degrade-loudly-allowlist.yaml +++ b/docs/plans/degrade-loudly-allowlist.yaml @@ -300,6 +300,22 @@ entries: - Exception occurrence: 0 reason: 'Returns {"available": False, "error": str(exc)} -- an already-typed signal.' +- path: polylogue/storage/archive_readiness.py + function: ._action_readiness_counts + exceptions: + - Error + occurrence: 0 + reason: 'Sets actions_view_error = str(exc) in the returned counts dict -- an already-typed signal. + Extracted from polylogue/cli/commands/status.py (polylogue-ogn1 layering fix); pre-existing + behavior, unchanged by the move.' +- path: polylogue/storage/archive_readiness.py + function: .archive_readiness_status + exceptions: + - Error + occurrence: 0 + reason: 'Returns {"checked": False, "reason": str(exc), "surfaces": {}} -- an already-typed signal. + Extracted from polylogue/cli/commands/status.py (polylogue-ogn1 layering fix); pre-existing + behavior, unchanged by the move.' - path: polylogue/storage/artifacts/inspection.py function: ._hermes_state_db_schema_version exceptions: diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index 89c8a38792..48216b5006 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/polylogue/cli/commands/maintenance/_rebuild_index.py @@ -16,6 +16,24 @@ from polylogue.paths import archive_root from polylogue.storage.archive_identity import ArchiveLocation +_BUILTIN_DAEMON_URL = "http://127.0.0.1:8766" + + +def _default_daemon_url() -> str: + """Resolve the default daemon URL through the layered config resolver. + + polylogue-ogn1: this option's default previously read + ``POLYLOGUE_DAEMON_URL`` directly via ``os.environ.get``, bypassing the + 5-layer config precedence chain (site TOML -> user TOML -> env -> CLI) + that every other daemon-URL-consuming surface in this repo goes through + (see ``polylogue.cli.commands.status._default_daemon_url``). A site/user + TOML ``daemon.url`` override was silently ignored here even though it was + honoured everywhere else. + """ + from polylogue.config import load_polylogue_config + + return load_polylogue_config().daemon_url or _BUILTIN_DAEMON_URL + def _run_daemon_rebuild( daemon_url: str, @@ -342,8 +360,8 @@ def _rebuild_index_selection_plan( ) @click.option( "--daemon-url", - default=lambda: __import__("os").environ.get("POLYLOGUE_DAEMON_URL", "http://127.0.0.1:8766"), - show_default="POLYLOGUE_DAEMON_URL or http://127.0.0.1:8766", + default=_default_daemon_url, + show_default="resolved via load_polylogue_config().daemon_url (site/user TOML -> POLYLOGUE_DAEMON_URL -> built-in default)", help="Daemon HTTP base URL used with --daemon.", ) def rebuild_index_command( diff --git a/polylogue/cli/commands/status.py b/polylogue/cli/commands/status.py index d683c4f90f..ca2eccd7bb 100644 --- a/polylogue/cli/commands/status.py +++ b/polylogue/cli/commands/status.py @@ -22,8 +22,8 @@ status_snapshot_has_fresh_provenance, ) from polylogue.readiness.claim_guard import derive_claim_guard +from polylogue.storage.archive_readiness import archive_readiness_status as _archive_readiness_status from polylogue.storage.archive_readiness import raw_materialization_ready as _raw_materialization_ready_bool -from polylogue.storage.insights.session.status import session_insight_status_sync from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier @@ -759,338 +759,6 @@ def _sqlite_maintenance_status(root: Path) -> dict[str, Any]: } -def _archive_readiness_status(root: Path) -> dict[str, Any]: - from polylogue.storage.archive_readiness import missing_source_raw_session_evidence - - index_db = root / "index.db" - source_db = root / "source.db" - if not index_db.exists(): - return {"checked": False, "reason": "missing_index_tier", "surfaces": {}} - - missing_source_evidence = missing_source_raw_session_evidence(root) - try: - conn = sqlite3.connect(f"file:{index_db}?mode=ro", uri=True) - try: - if not _table_exists(conn, "sessions"): - return {"checked": False, "reason": "missing_sessions_table", "surfaces": {}} - source_check_available = source_db.exists() - source_conn: sqlite3.Connection | None = None - try: - if source_check_available: - source_conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) - source_check_available = _table_exists(source_conn, "raw_sessions") - counts = _archive_readiness_counts( - conn, - source_conn=source_conn, - source_check_available=source_check_available, - ) - if missing_source_evidence.get("available"): - missing_raw_count = _safe_int(missing_source_evidence.get("missing_raw_session_count")) - missing_raw_samples = _safe_list(missing_source_evidence.get("missing_raw_session_samples")) - lost_source_count = _safe_int(missing_source_evidence.get("lost_source_evidence_count")) - lost_source_samples = _safe_list(missing_source_evidence.get("lost_source_evidence_samples")) - counts.update( - { - "missing_raw_session_count": missing_raw_count, - "missing_raw_session_samples": missing_raw_samples - or _safe_list(counts.get("missing_raw_session_samples")), - "lost_source_evidence_count": lost_source_count, - "lost_source_evidence_samples": lost_source_samples - or _safe_list(counts.get("lost_source_evidence_samples")), - } - ) - finally: - if source_conn is not None: - source_conn.close() - finally: - conn.close() - except sqlite3.Error as exc: - return {"checked": False, "reason": str(exc), "surfaces": {}} - - surfaces = _archive_status_surfaces(counts, source_check_available=source_check_available) - ready_count = sum(1 for info in surfaces.values() if info["ready"] is True) - blocked_count = sum(1 for info in surfaces.values() if info["ready"] is not True) - return { - "checked": True, - "reason": None, - "source_check_available": source_check_available, - "ready_surface_count": ready_count, - "blocked_surface_count": blocked_count, - "total_surface_count": len(surfaces), - "counts": counts, - "surfaces": surfaces, - } - - -def _archive_readiness_counts( - conn: sqlite3.Connection, - *, - source_conn: sqlite3.Connection | None, - source_check_available: bool, -) -> dict[str, Any]: - session_count = _fast_count(conn, "SELECT COUNT(*) FROM sessions") - raw_link_count = ( - _fast_count(conn, "SELECT COUNT(*) FROM sessions WHERE raw_id IS NOT NULL") - if _column_exists(conn, "sessions", "raw_id") - else 0 - ) - missing_raw_session_count = 0 - missing_raw_session_samples: list[dict[str, Any]] = [] - if source_check_available and source_conn is not None and _column_exists(conn, "sessions", "raw_id"): - raw_ids = { - str(row[0]) - for row in source_conn.execute("SELECT raw_id FROM raw_sessions").fetchall() - if row[0] is not None - } - missing_rows = [ - row - for row in conn.execute( - """ - SELECT session_id, origin, native_id, raw_id, message_count, updated_at_ms - FROM sessions - WHERE raw_id IS NOT NULL - ORDER BY updated_at_ms DESC, session_id - """ - ).fetchall() - if str(row[3]) not in raw_ids - ] - missing_raw_session_count = len(missing_rows) - missing_raw_session_samples = [ - { - "session_id": str(row[0]), - "origin": str(row[1]), - "native_id": str(row[2]), - "missing_raw_id": str(row[3]), - "message_count": int(row[4] or 0), - "updated_at_ms": None if row[5] is None else int(row[5]), - "evidence_status": "lost_source_evidence", - "loss_reason": "index_raw_id_missing_from_source_tier", - "recovery_requirement": "restore_exact_raw_artifact_or_keep_blocked", - } - for row in missing_rows[:10] - ] - insight_status = session_insight_status_sync(conn, verify_freshness=True) - return { - "session_count": session_count, - "raw_link_count": raw_link_count, - "missing_raw_session_count": missing_raw_session_count, - "missing_raw_session_samples": missing_raw_session_samples, - "lost_source_evidence_count": missing_raw_session_count, - "lost_source_evidence_samples": missing_raw_session_samples, - "message_count": _fast_count(conn, "SELECT COUNT(*) FROM messages") if _table_exists(conn, "messages") else 0, - "text_block_count": _fast_count(conn, "SELECT COUNT(*) FROM blocks WHERE search_text != ''") - if _table_exists(conn, "blocks") - else 0, - "messages_fts_count": _fast_count(conn, "SELECT COUNT(*) FROM messages_fts") - if _table_exists(conn, "messages_fts") - else 0, - "profile_row_count": insight_status.profile_row_count, - "missing_profile_row_count": insight_status.missing_profile_row_count, - "stale_profile_row_count": insight_status.stale_profile_row_count, - "orphan_profile_row_count": insight_status.orphan_profile_row_count, - "work_event_row_count": insight_status.work_event_inference_count, - "expected_work_event_row_count": insight_status.expected_work_event_inference_count, - "stale_work_event_row_count": insight_status.stale_work_event_inference_count, - "orphan_work_event_row_count": insight_status.orphan_work_event_inference_count, - "phase_row_count": insight_status.phase_inference_count, - "expected_phase_row_count": insight_status.expected_phase_inference_count, - "stale_phase_row_count": insight_status.stale_phase_inference_count, - "orphan_phase_row_count": insight_status.orphan_phase_inference_count, - "thread_count": insight_status.thread_count, - "root_thread_count": insight_status.root_threads, - "stale_thread_count": insight_status.stale_thread_count, - "orphan_thread_count": insight_status.orphan_thread_count, - **_action_readiness_counts(conn), - "missing_session_profile_materialization": insight_status.missing_session_profile_materialization_count, - "missing_work_events_materialization": insight_status.missing_work_event_materialization_count, - "missing_phases_materialization": insight_status.missing_phase_materialization_count, - "missing_thread_materialization": insight_status.missing_thread_materialization_count, - "missing_latency_materialization": insight_status.missing_latency_materialization_count, - } - - -def _action_readiness_counts(conn: sqlite3.Connection) -> dict[str, Any]: - """Return exact, non-vacuous evidence for the derived ``actions`` view.""" - tool_use_block_count = ( - _fast_count(conn, "SELECT COUNT(*) FROM blocks WHERE block_type = 'tool_use'") - if _table_exists(conn, "blocks") - else 0 - ) - actions_view_present = _view_exists(conn, "actions") - action_count = 0 - actions_view_error: str | None = None - if actions_view_present: - try: - action_count = _fast_count(conn, "SELECT COUNT(*) FROM actions") - except sqlite3.Error as exc: - actions_view_error = str(exc) - return { - "action_count": action_count, - "tool_use_block_count": tool_use_block_count, - "actions_view_present": actions_view_present, - "actions_view_error": actions_view_error, - } - - -def _archive_status_surfaces(counts: dict[str, Any], *, source_check_available: bool) -> dict[str, dict[str, Any]]: - def surface(*, ready: bool | None, blockers: list[str], evidence: dict[str, Any]) -> dict[str, Any]: - return {"ready": ready, "blockers": blockers, "evidence": evidence} - - def count(key: str, default: int = 0) -> int: - return int(counts.get(key, default)) - - def present_blockers(*keys: str) -> list[str]: - return [key for key in keys if count(key) != 0] - - def mismatch_blocker(actual_key: str, expected_key: str, blocker: str) -> list[str]: - expected = count(expected_key, count(actual_key)) - return [blocker] if count(actual_key) != expected else [] - - raw_blockers: list[str] = [] - raw_ready: bool | None - if not source_check_available: - raw_ready = None - raw_blockers.append("source_tier_unavailable") - elif count("missing_raw_session_count"): - raw_ready = False - raw_blockers.append("missing_source_raw_sessions") - else: - raw_ready = True - - search_blockers = ["messages_fts_row_mismatch"] if count("text_block_count") != count("messages_fts_count") else [] - profile_blockers: list[str] = [] - if count("missing_profile_row_count"): - profile_blockers.append("missing_profile_rows") - profile_blockers.extend( - present_blockers( - "missing_session_profile_materialization", - "stale_profile_row_count", - "orphan_profile_row_count", - ) - ) - - def materialized(name: str) -> tuple[bool, list[str]]: - key = f"missing_{name}_materialization" - missing = count(key) - return (missing == 0, [] if missing == 0 else [key]) - - work_blockers = present_blockers( - "missing_work_events_materialization", - "stale_work_event_row_count", - "orphan_work_event_row_count", - ) - work_blockers.extend( - mismatch_blocker("work_event_row_count", "expected_work_event_row_count", "work_event_row_mismatch") - ) - phase_blockers = present_blockers( - "missing_phases_materialization", - "stale_phase_row_count", - "orphan_phase_row_count", - ) - phase_blockers.extend(mismatch_blocker("phase_row_count", "expected_phase_row_count", "phase_row_mismatch")) - thread_blockers = present_blockers( - "missing_thread_materialization", - "stale_thread_count", - "orphan_thread_count", - ) - thread_blockers.extend(mismatch_blocker("thread_count", "root_thread_count", "thread_root_mismatch")) - latency_ready, latency_blockers = materialized("latency") - tool_usage_blockers: list[str] = [] - if not bool(counts.get("actions_view_present", False)): - tool_usage_blockers.append("actions_view_missing") - elif counts.get("actions_view_error"): - tool_usage_blockers.append("actions_view_unreadable") - elif count("tool_use_block_count") != count("action_count"): - tool_usage_blockers.append("actions_tool_use_count_mismatch") - - return { - "archive_sessions": surface( - ready=True, - blockers=[], - evidence={"session_count": count("session_count"), "message_count": count("message_count")}, - ), - "raw_artifacts": surface( - ready=raw_ready, - blockers=raw_blockers, - evidence={ - "source_check_available": source_check_available, - "raw_link_count": count("raw_link_count"), - "missing_raw_session_count": count("missing_raw_session_count"), - "missing_raw_session_samples": list(counts.get("missing_raw_session_samples") or []), - "lost_source_evidence_count": count("lost_source_evidence_count"), - "lost_source_evidence_samples": list(counts.get("lost_source_evidence_samples") or []), - }, - ), - "search": surface( - ready=not search_blockers, - blockers=search_blockers, - evidence={ - "text_block_count": count("text_block_count"), - "messages_fts_count": count("messages_fts_count"), - }, - ), - "session_profiles": surface( - ready=not profile_blockers, - blockers=profile_blockers, - evidence={ - "profile_row_count": count("profile_row_count"), - "missing_profile_row_count": count("missing_profile_row_count"), - "missing_materialization_count": count("missing_session_profile_materialization"), - "stale_profile_row_count": count("stale_profile_row_count"), - "orphan_profile_row_count": count("orphan_profile_row_count"), - }, - ), - "timeline_work_events": surface( - ready=not work_blockers, - blockers=work_blockers, - evidence={ - "work_event_row_count": count("work_event_row_count"), - "expected_work_event_row_count": count("expected_work_event_row_count", count("work_event_row_count")), - "missing_materialization_count": count("missing_work_events_materialization"), - "stale_work_event_row_count": count("stale_work_event_row_count"), - "orphan_work_event_row_count": count("orphan_work_event_row_count"), - }, - ), - "timeline_phases": surface( - ready=not phase_blockers, - blockers=phase_blockers, - evidence={ - "phase_row_count": count("phase_row_count"), - "expected_phase_row_count": count("expected_phase_row_count", count("phase_row_count")), - "missing_materialization_count": count("missing_phases_materialization"), - "stale_phase_row_count": count("stale_phase_row_count"), - "orphan_phase_row_count": count("orphan_phase_row_count"), - }, - ), - "threads": surface( - ready=not thread_blockers, - blockers=thread_blockers, - evidence={ - "thread_count": count("thread_count"), - "root_thread_count": count("root_thread_count", count("thread_count")), - "missing_materialization_count": count("missing_thread_materialization"), - "stale_thread_count": count("stale_thread_count"), - "orphan_thread_count": count("orphan_thread_count"), - }, - ), - "tool_usage": surface( - ready=not tool_usage_blockers, - blockers=tool_usage_blockers, - evidence={ - "action_count": count("action_count"), - "tool_use_block_count": count("tool_use_block_count"), - "actions_view_present": bool(counts.get("actions_view_present", False)), - "actions_view_error": counts.get("actions_view_error"), - }, - ), - "latency_profiles": surface( - ready=latency_ready, - blockers=latency_blockers, - evidence={"missing_materialization_count": counts["missing_latency_materialization"]}, - ), - } - - def _direct_archive_counts(conn: Any) -> dict[str, int]: if _table_exists(conn, "sessions"): messages = ( diff --git a/polylogue/daemon/http.py b/polylogue/daemon/http.py index 71b2ba95a3..9ddff9d8c9 100644 --- a/polylogue/daemon/http.py +++ b/polylogue/daemon/http.py @@ -1243,6 +1243,16 @@ def __bool__(self) -> bool: return self.allowed +# polylogue-ogn1: the write bridge's default run_sync/hold timeout (30s, +# DaemonWriteThreadBridge.__init__) is sized for ordinary request-scoped +# writes. A bounded rebuild-index pass is allowed to run far longer -- the +# CLI's own --daemon HTTP client already tolerates up to 600s +# (_rebuild_index.py's _run_daemon_rebuild, urlopen(..., timeout=600)) -- so +# the HTTP route asks the bridge to wait that same 600s instead of the 30s +# default, which would otherwise kill a still-running rebuild pass early. +_REBUILD_INDEX_WRITE_TIMEOUT_S = 600.0 + + class DaemonAPIHandler(BaseHTTPRequestHandler): """HTTP handler for the daemon API server. @@ -5228,7 +5238,14 @@ def _handle_maintenance_run(self) -> None: @daemon_safe_handler def _handle_rebuild_index(self) -> None: - """POST /api/maintenance/rebuild-index — one coordinator-owned replay pass.""" + """POST /api/maintenance/rebuild-index — one coordinator-owned replay pass. + + polylogue-ogn1: waits up to ``_REBUILD_INDEX_WRITE_TIMEOUT_S`` through + the write bridge, matching the CLI's own ``--daemon`` HTTP client + timeout (``_run_daemon_rebuild``'s ``urlopen(..., timeout=600)``) + rather than the bridge's much shorter default request timeout (30s), + which would otherwise kill a still-running rebuild pass early. + """ content_length = int(self.headers.get("Content-Length", 0)) body_raw = self.rfile.read(content_length) if content_length > 0 else b"{}" try: @@ -5303,15 +5320,22 @@ def _handle_rebuild_index(self) -> None: bridge = getattr(self.server, "write_bridge", None) if bridge is None: - # Direct handler unit tests predate the server-owned bridge; real - # daemon servers always install it and therefore use run_sync. - receipt = rebuild_index_from_source_sync(request) - else: - receipt = cast(DaemonWriteThreadBridge, bridge).run_sync( - "http.maintenance.rebuild-index", - rebuild_index_from_source_sync, - request, - ) + # polylogue-ogn1: a real DaemonAPIHTTPServer always installs + # write_bridge in __init__ (either the caller's coordinator or an + # owned standalone one) -- this branch is never reachable there. + # Fail closed instead of running the rebuild directly outside the + # sole-writer coordinator: a route that can execute an authority- + # promoting archive write without ever holding the writer gate is + # a bypass of this daemon's single-writer invariant, not a safe + # fallback, even if nothing exercises it in production today. + self._send_error(HTTPStatus.SERVICE_UNAVAILABLE, "write_coordinator_unavailable") + return + receipt = cast(DaemonWriteThreadBridge, bridge).run_sync_with_timeout( + "http.maintenance.rebuild-index", + _REBUILD_INDEX_WRITE_TIMEOUT_S, + rebuild_index_from_source_sync, + request, + ) self._send_json(HTTPStatus.OK, receipt.to_dict()) @daemon_safe_handler diff --git a/polylogue/daemon/write_coordinator.py b/polylogue/daemon/write_coordinator.py index d66cd98216..155b3b81a8 100644 --- a/polylogue/daemon/write_coordinator.py +++ b/polylogue/daemon/write_coordinator.py @@ -492,11 +492,36 @@ def run_sync(self, actor: str, function: Callable[P, T], /, *args: P.args, **kwa Unlike :meth:`hold`, this is for a complete bounded request operation: the coordinator owns the worker thread until the function has really returned, so a timed-out HTTP caller never admits a second writer. + + Waits at most this bridge's constructor ``timeout`` (default 30s) for + completion. Use :meth:`run_sync_with_timeout` for an operation whose + own contract needs a longer bound (see polylogue-ogn1). + """ + return self.run_sync_with_timeout(actor, self._timeout, function, *args, **kwargs) + + def run_sync_with_timeout( + self, + actor: str, + timeout: float, + function: Callable[P, T], + /, + *args: P.args, + **kwargs: P.kwargs, + ) -> T: + """Like :meth:`run_sync`, waiting up to ``timeout`` seconds instead of the bridge default. + + polylogue-ogn1: the bridge's constructor ``timeout`` (30s) is sized for + ordinary request-scoped writes (reset, ingest, maintenance run). A + bounded index rebuild pass can legitimately run far longer -- the + CLI/HTTP contract already allows up to 600s (``_run_daemon_rebuild``'s + ``urlopen(..., timeout=600)``) -- so that call site needs its own, + longer wait here rather than being silently killed by the bridge's + default gate at 30s while the rebuild is still replaying. """ future = asyncio.run_coroutine_threadsafe( self._coordinator.run_sync(actor, function, *args, **kwargs), self._loop ) - return future.result(timeout=self._timeout) + return future.result(timeout=timeout) def daemon_write_telemetry_payload() -> dict[str, object]: diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 3c270b2866..69da13baf6 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -247,10 +247,20 @@ def count_source_raw_sessions(root: Path) -> int: def missing_index_raw_ids(root: Path) -> list[str]: + """Return source raw_ids that have not yet reached ``index.sessions``. + + polylogue-ogn1: a missing/lost ``index.db`` (fresh archive, or one just + reset via ``ops reset --index``) means every source row is missing from + the index by definition -- return the full source set instead of an + empty list, so ``--only-missing`` actually rebuilds something on a + fresh/lost index rather than silently doing nothing. + """ source_db = root / "source.db" - index_db = ArchiveLocation.resolve(root).active_index_path - if not source_db.exists() or not index_db.exists(): + if not source_db.exists(): return [] + index_db = ArchiveLocation.resolve(root).active_index_path + if not index_db.exists(): + return all_index_rebuild_raw_ids(root) with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: conn.execute("ATTACH DATABASE ? AS idx", (str(index_db),)) rows = conn.execute( @@ -304,9 +314,9 @@ def select_rebuild_raw_ids(request: RebuildIndexRequest) -> tuple[int, list[str] async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildIndexReceipt: """Replay one source snapshot into an owned generation and optionally promote it.""" - from polylogue.cli.commands.status import _archive_readiness_status from polylogue.maintenance.archive_verification import verify_archive from polylogue.maintenance.replay import rebuild_index_from_source as replay_source + from polylogue.storage.archive_readiness import archive_readiness_status from polylogue.storage.index_generation import IndexGenerationStore, RebuildLease, source_revision_snapshot from polylogue.storage.repair import repair_session_insights @@ -522,7 +532,7 @@ async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildInde f"bulk-build FTS/trigram parity failed for generation {generation.generation_id}: {failing}" ) terminal_started_at = time.perf_counter() - readiness = _archive_readiness_status(generation_root) + readiness = archive_readiness_status(generation_root) logger.info( "rebuild_terminal_stage_complete", generation_id=generation.generation_id, diff --git a/polylogue/storage/archive_readiness.py b/polylogue/storage/archive_readiness.py index 0a0948ef64..9c9620488a 100644 --- a/polylogue/storage/archive_readiness.py +++ b/polylogue/storage/archive_readiness.py @@ -16,6 +16,7 @@ ) from polylogue.archive.revision_authority import BYTE_AUTHORITY_CENSUS_DETAIL from polylogue.logging import get_logger +from polylogue.storage.insights.session.status import session_insight_status_sync from polylogue.storage.raw_authority import raw_authority_detail_query_handle logger = get_logger(__name__) @@ -837,9 +838,406 @@ def _raw_gap_parsed_non_session_artifact( ) +# --------------------------------------------------------------------------- +# Archive readiness surfaces (polylogue-ogn1) +# +# Extracted from ``polylogue/cli/commands/status.py``: the substrate module +# ``polylogue/maintenance/rebuild_index.py`` was importing a private +# CLI-surface helper (``_archive_readiness_status``) to check whether a freshly +# rebuilt generation is exact-ready before promotion. That is the inverse of +# this repo's documented layering rule ("surfaces may not import substrate +# internals directly", ``docs/plans/layering.yaml``) — here the substrate was +# reaching *up* into a CLI leaf adapter. This block gives both the CLI +# (`status.py`, human-facing readiness reporting) and the substrate +# (`rebuild_index.py`, promotion gating) a single shared home for the +# computation; the CLI now delegates to ``archive_readiness_status`` below +# instead of owning the only copy. The handful of tiny SQLite-introspection +# one-liners below (``_fast_count``/``_safe_int``/``_table_exists``/etc.) are +# intentionally duplicated from ``status.py``'s own private copies rather than +# migrated wholesale: those are used throughout the rest of ``status.py`` for +# unrelated status surfaces outside this cluster's scope, and a bulk +# utility-relocation refactor was not part of the layering fix being made. +# --------------------------------------------------------------------------- + + +def _fast_count(conn: sqlite3.Connection, sql: str, params: tuple[object, ...] = ()) -> int: + row = conn.execute(sql, params).fetchone() + return int(row[0] or 0) if row is not None else 0 + + +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) if value is not None else default + except (TypeError, ValueError): + return default + + +def _safe_list(value: Any) -> list[Any]: + return value if isinstance(value, list) else [] + + +def _schema_object_exists(conn: sqlite3.Connection, name: str, *, types: tuple[str, ...]) -> bool: + placeholders = ", ".join("?" for _ in types) + row = conn.execute( + f"SELECT 1 FROM sqlite_master WHERE type IN ({placeholders}) AND name = ? LIMIT 1", + (*types, name), + ).fetchone() + return row is not None + + +def _table_exists(conn: sqlite3.Connection, table_name: str) -> bool: + return _schema_object_exists(conn, table_name, types=("table",)) + + +def _view_exists(conn: sqlite3.Connection, view_name: str) -> bool: + return _schema_object_exists(conn, view_name, types=("view",)) + + +def _column_exists(conn: sqlite3.Connection, table_name: str, column_name: str) -> bool: + return any(str(row[1]) == column_name for row in conn.execute(f"PRAGMA table_info({table_name})").fetchall()) + + +def _action_readiness_counts(conn: sqlite3.Connection) -> dict[str, Any]: + """Return exact, non-vacuous evidence for the derived ``actions`` view.""" + tool_use_block_count = ( + _fast_count(conn, "SELECT COUNT(*) FROM blocks WHERE block_type = 'tool_use'") + if _table_exists(conn, "blocks") + else 0 + ) + actions_view_present = _view_exists(conn, "actions") + action_count = 0 + actions_view_error: str | None = None + if actions_view_present: + try: + action_count = _fast_count(conn, "SELECT COUNT(*) FROM actions") + except sqlite3.Error as exc: + actions_view_error = str(exc) + return { + "action_count": action_count, + "tool_use_block_count": tool_use_block_count, + "actions_view_present": actions_view_present, + "actions_view_error": actions_view_error, + } + + +def _archive_readiness_counts( + conn: sqlite3.Connection, + *, + source_conn: sqlite3.Connection | None, + source_check_available: bool, +) -> dict[str, Any]: + session_count = _fast_count(conn, "SELECT COUNT(*) FROM sessions") + raw_link_count = ( + _fast_count(conn, "SELECT COUNT(*) FROM sessions WHERE raw_id IS NOT NULL") + if _column_exists(conn, "sessions", "raw_id") + else 0 + ) + missing_raw_session_count = 0 + missing_raw_session_samples: list[dict[str, Any]] = [] + if source_check_available and source_conn is not None and _column_exists(conn, "sessions", "raw_id"): + raw_ids = { + str(row[0]) + for row in source_conn.execute("SELECT raw_id FROM raw_sessions").fetchall() + if row[0] is not None + } + missing_rows = [ + row + for row in conn.execute( + """ + SELECT session_id, origin, native_id, raw_id, message_count, updated_at_ms + FROM sessions + WHERE raw_id IS NOT NULL + ORDER BY updated_at_ms DESC, session_id + """ + ).fetchall() + if str(row[3]) not in raw_ids + ] + missing_raw_session_count = len(missing_rows) + missing_raw_session_samples = [ + { + "session_id": str(row[0]), + "origin": str(row[1]), + "native_id": str(row[2]), + "missing_raw_id": str(row[3]), + "message_count": int(row[4] or 0), + "updated_at_ms": None if row[5] is None else int(row[5]), + "evidence_status": "lost_source_evidence", + "loss_reason": "index_raw_id_missing_from_source_tier", + "recovery_requirement": "restore_exact_raw_artifact_or_keep_blocked", + } + for row in missing_rows[:10] + ] + insight_status = session_insight_status_sync(conn, verify_freshness=True) + return { + "session_count": session_count, + "raw_link_count": raw_link_count, + "missing_raw_session_count": missing_raw_session_count, + "missing_raw_session_samples": missing_raw_session_samples, + "lost_source_evidence_count": missing_raw_session_count, + "lost_source_evidence_samples": missing_raw_session_samples, + "message_count": _fast_count(conn, "SELECT COUNT(*) FROM messages") if _table_exists(conn, "messages") else 0, + "text_block_count": _fast_count(conn, "SELECT COUNT(*) FROM blocks WHERE search_text != ''") + if _table_exists(conn, "blocks") + else 0, + "messages_fts_count": _fast_count(conn, "SELECT COUNT(*) FROM messages_fts") + if _table_exists(conn, "messages_fts") + else 0, + "profile_row_count": insight_status.profile_row_count, + "missing_profile_row_count": insight_status.missing_profile_row_count, + "stale_profile_row_count": insight_status.stale_profile_row_count, + "orphan_profile_row_count": insight_status.orphan_profile_row_count, + "work_event_row_count": insight_status.work_event_inference_count, + "expected_work_event_row_count": insight_status.expected_work_event_inference_count, + "stale_work_event_row_count": insight_status.stale_work_event_inference_count, + "orphan_work_event_row_count": insight_status.orphan_work_event_inference_count, + "phase_row_count": insight_status.phase_inference_count, + "expected_phase_row_count": insight_status.expected_phase_inference_count, + "stale_phase_row_count": insight_status.stale_phase_inference_count, + "orphan_phase_row_count": insight_status.orphan_phase_inference_count, + "thread_count": insight_status.thread_count, + "root_thread_count": insight_status.root_threads, + "stale_thread_count": insight_status.stale_thread_count, + "orphan_thread_count": insight_status.orphan_thread_count, + **_action_readiness_counts(conn), + "missing_session_profile_materialization": insight_status.missing_session_profile_materialization_count, + "missing_work_events_materialization": insight_status.missing_work_event_materialization_count, + "missing_phases_materialization": insight_status.missing_phase_materialization_count, + "missing_thread_materialization": insight_status.missing_thread_materialization_count, + "missing_latency_materialization": insight_status.missing_latency_materialization_count, + } + + +def _archive_status_surfaces(counts: dict[str, Any], *, source_check_available: bool) -> dict[str, dict[str, Any]]: + def surface(*, ready: bool | None, blockers: list[str], evidence: dict[str, Any]) -> dict[str, Any]: + return {"ready": ready, "blockers": blockers, "evidence": evidence} + + def count(key: str, default: int = 0) -> int: + return int(counts.get(key, default)) + + def present_blockers(*keys: str) -> list[str]: + return [key for key in keys if count(key) != 0] + + def mismatch_blocker(actual_key: str, expected_key: str, blocker: str) -> list[str]: + expected = count(expected_key, count(actual_key)) + return [blocker] if count(actual_key) != expected else [] + + raw_blockers: list[str] = [] + raw_ready: bool | None + if not source_check_available: + raw_ready = None + raw_blockers.append("source_tier_unavailable") + elif count("missing_raw_session_count"): + raw_ready = False + raw_blockers.append("missing_source_raw_sessions") + else: + raw_ready = True + + search_blockers = ["messages_fts_row_mismatch"] if count("text_block_count") != count("messages_fts_count") else [] + profile_blockers: list[str] = [] + if count("missing_profile_row_count"): + profile_blockers.append("missing_profile_rows") + profile_blockers.extend( + present_blockers( + "missing_session_profile_materialization", + "stale_profile_row_count", + "orphan_profile_row_count", + ) + ) + + def materialized(name: str) -> tuple[bool, list[str]]: + key = f"missing_{name}_materialization" + missing = count(key) + return (missing == 0, [] if missing == 0 else [key]) + + work_blockers = present_blockers( + "missing_work_events_materialization", + "stale_work_event_row_count", + "orphan_work_event_row_count", + ) + work_blockers.extend( + mismatch_blocker("work_event_row_count", "expected_work_event_row_count", "work_event_row_mismatch") + ) + phase_blockers = present_blockers( + "missing_phases_materialization", + "stale_phase_row_count", + "orphan_phase_row_count", + ) + phase_blockers.extend(mismatch_blocker("phase_row_count", "expected_phase_row_count", "phase_row_mismatch")) + thread_blockers = present_blockers( + "missing_thread_materialization", + "stale_thread_count", + "orphan_thread_count", + ) + thread_blockers.extend(mismatch_blocker("thread_count", "root_thread_count", "thread_root_mismatch")) + latency_ready, latency_blockers = materialized("latency") + tool_usage_blockers: list[str] = [] + if not bool(counts.get("actions_view_present", False)): + tool_usage_blockers.append("actions_view_missing") + elif counts.get("actions_view_error"): + tool_usage_blockers.append("actions_view_unreadable") + elif count("tool_use_block_count") != count("action_count"): + tool_usage_blockers.append("actions_tool_use_count_mismatch") + + return { + "archive_sessions": surface( + ready=True, + blockers=[], + evidence={"session_count": count("session_count"), "message_count": count("message_count")}, + ), + "raw_artifacts": surface( + ready=raw_ready, + blockers=raw_blockers, + evidence={ + "source_check_available": source_check_available, + "raw_link_count": count("raw_link_count"), + "missing_raw_session_count": count("missing_raw_session_count"), + "missing_raw_session_samples": list(counts.get("missing_raw_session_samples") or []), + "lost_source_evidence_count": count("lost_source_evidence_count"), + "lost_source_evidence_samples": list(counts.get("lost_source_evidence_samples") or []), + }, + ), + "search": surface( + ready=not search_blockers, + blockers=search_blockers, + evidence={ + "text_block_count": count("text_block_count"), + "messages_fts_count": count("messages_fts_count"), + }, + ), + "session_profiles": surface( + ready=not profile_blockers, + blockers=profile_blockers, + evidence={ + "profile_row_count": count("profile_row_count"), + "missing_profile_row_count": count("missing_profile_row_count"), + "missing_materialization_count": count("missing_session_profile_materialization"), + "stale_profile_row_count": count("stale_profile_row_count"), + "orphan_profile_row_count": count("orphan_profile_row_count"), + }, + ), + "timeline_work_events": surface( + ready=not work_blockers, + blockers=work_blockers, + evidence={ + "work_event_row_count": count("work_event_row_count"), + "expected_work_event_row_count": count("expected_work_event_row_count", count("work_event_row_count")), + "missing_materialization_count": count("missing_work_events_materialization"), + "stale_work_event_row_count": count("stale_work_event_row_count"), + "orphan_work_event_row_count": count("orphan_work_event_row_count"), + }, + ), + "timeline_phases": surface( + ready=not phase_blockers, + blockers=phase_blockers, + evidence={ + "phase_row_count": count("phase_row_count"), + "expected_phase_row_count": count("expected_phase_row_count", count("phase_row_count")), + "missing_materialization_count": count("missing_phases_materialization"), + "stale_phase_row_count": count("stale_phase_row_count"), + "orphan_phase_row_count": count("orphan_phase_row_count"), + }, + ), + "threads": surface( + ready=not thread_blockers, + blockers=thread_blockers, + evidence={ + "thread_count": count("thread_count"), + "root_thread_count": count("root_thread_count", count("thread_count")), + "missing_materialization_count": count("missing_thread_materialization"), + "stale_thread_count": count("stale_thread_count"), + "orphan_thread_count": count("orphan_thread_count"), + }, + ), + "tool_usage": surface( + ready=not tool_usage_blockers, + blockers=tool_usage_blockers, + evidence={ + "action_count": count("action_count"), + "tool_use_block_count": count("tool_use_block_count"), + "actions_view_present": bool(counts.get("actions_view_present", False)), + "actions_view_error": counts.get("actions_view_error"), + }, + ), + "latency_profiles": surface( + ready=latency_ready, + blockers=latency_blockers, + evidence={"missing_materialization_count": counts["missing_latency_materialization"]}, + ), + } + + +def archive_readiness_status(root: Path) -> dict[str, Any]: + """Return the exact-readiness surface report for one archive root. + + Shared by the CLI's ``status``/``rebuild-index --plan`` reporting and the + substrate's ``rebuild_index_from_source`` promotion gate: a freshly + rebuilt generation is only promoted once every surface here reports + ``ready``. + """ + index_db = root / "index.db" + source_db = root / "source.db" + if not index_db.exists(): + return {"checked": False, "reason": "missing_index_tier", "surfaces": {}} + + missing_source_evidence = missing_source_raw_session_evidence(root) + try: + conn = sqlite3.connect(f"file:{index_db}?mode=ro", uri=True) + try: + if not _table_exists(conn, "sessions"): + return {"checked": False, "reason": "missing_sessions_table", "surfaces": {}} + source_check_available = source_db.exists() + source_conn: sqlite3.Connection | None = None + try: + if source_check_available: + source_conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) + source_check_available = _table_exists(source_conn, "raw_sessions") + counts = _archive_readiness_counts( + conn, + source_conn=source_conn, + source_check_available=source_check_available, + ) + if missing_source_evidence.get("available"): + missing_raw_count = _safe_int(missing_source_evidence.get("missing_raw_session_count")) + missing_raw_samples = _safe_list(missing_source_evidence.get("missing_raw_session_samples")) + lost_source_count = _safe_int(missing_source_evidence.get("lost_source_evidence_count")) + lost_source_samples = _safe_list(missing_source_evidence.get("lost_source_evidence_samples")) + counts.update( + { + "missing_raw_session_count": missing_raw_count, + "missing_raw_session_samples": missing_raw_samples + or _safe_list(counts.get("missing_raw_session_samples")), + "lost_source_evidence_count": lost_source_count, + "lost_source_evidence_samples": lost_source_samples + or _safe_list(counts.get("lost_source_evidence_samples")), + } + ) + finally: + if source_conn is not None: + source_conn.close() + finally: + conn.close() + except sqlite3.Error as exc: + return {"checked": False, "reason": str(exc), "surfaces": {}} + + surfaces = _archive_status_surfaces(counts, source_check_available=source_check_available) + ready_count = sum(1 for info in surfaces.values() if info["ready"] is True) + blocked_count = sum(1 for info in surfaces.values() if info["ready"] is not True) + return { + "checked": True, + "reason": None, + "source_check_available": source_check_available, + "ready_surface_count": ready_count, + "blocked_surface_count": blocked_count, + "total_surface_count": len(surfaces), + "counts": counts, + "surfaces": surfaces, + } + + __all__ = [ "ACTIVE_REBUILD_STALE_AFTER_S", "active_rebuild_index_attempts", + "archive_readiness_status", "missing_source_raw_session_evidence", "raw_materialization_readiness_snapshot", "raw_materialization_ready", diff --git a/tests/unit/cli/commands/test_status.py b/tests/unit/cli/commands/test_status.py index 093102b3f9..959c789988 100644 --- a/tests/unit/cli/commands/test_status.py +++ b/tests/unit/cli/commands/test_status.py @@ -12,15 +12,12 @@ _ARCHIVE_FACADE_ROUTES, _ARCHIVE_TIER_ENUM, _BUILTIN_DAEMON_URL, - _action_readiness_counts, _archive_cli_route_status, _archive_facade_route_status, _archive_one_tier_status, _archive_primary_tier_count, - _archive_readiness_counts, _archive_route_count_summary, _archive_runtime_path_status, - _archive_status_surfaces, _archive_table_counts, _archive_tier_files, _archive_tier_status, @@ -34,6 +31,19 @@ _table_exists, _view_exists, ) + +# polylogue-ogn1: these three moved from polylogue.cli.commands.status to the +# substrate module polylogue.storage.archive_readiness (the layering fix for +# CodeRabbit finding #8 -- polylogue/maintenance/rebuild_index.py, a +# substrate module, was importing the CLI's private _archive_readiness_status +# to gate promotion, inverting this repo's surfaces-may-not-import-substrate +# rule). status.py now delegates to the same shared implementation instead of +# owning a second copy. +from polylogue.storage.archive_readiness import ( + _action_readiness_counts, + _archive_readiness_counts, + _archive_status_surfaces, +) from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier diff --git a/tests/unit/daemon/test_http_write_coordination.py b/tests/unit/daemon/test_http_write_coordination.py index 24b26edf95..958a1d7b2a 100644 --- a/tests/unit/daemon/test_http_write_coordination.py +++ b/tests/unit/daemon/test_http_write_coordination.py @@ -4,6 +4,7 @@ import contextlib from collections.abc import Awaitable, Callable, Iterator +from http import HTTPStatus from types import SimpleNamespace from unittest.mock import AsyncMock, patch @@ -96,18 +97,67 @@ def dispatch_delete(*_args: object) -> bool: ] -def test_rebuild_index_route_uses_the_bridge_run_sync_writer_path() -> None: - timeline: list[str] = [] - handler = _handler(["api", "maintenance", "rebuild-index"], timeline) +class _RecordingRebuildBridge(_RecordingBridge): + """Adds ``run_sync_with_timeout`` so real ``_handle_rebuild_index`` can run. - def body() -> None: - bridge = handler.server.write_bridge - bridge.run_sync("http.maintenance.rebuild-index", lambda: timeline.append("body")) + polylogue-ogn1: rebuild-index uses ``run_sync_with_timeout`` (not + ``run_sync``) so a long rebuild pass isn't killed by the bridge's much + shorter default request timeout -- see ``DaemonAPIHandler._handle_rebuild_index`` + and ``DaemonWriteThreadBridge.run_sync_with_timeout``. + """ - handler._handle_rebuild_index = body # type: ignore[method-assign] - handler._do_post_impl() + def run_sync_with_timeout( + self, actor: str, timeout: float, function: Callable[..., object], *args: object + ) -> object: + self.timeline.append(f"run_sync_with_timeout:{actor}:{timeout}") + return function(*args) + + +def test_rebuild_index_route_uses_the_bridge_run_sync_with_timeout_writer_path(monkeypatch, tmp_path) -> None: # type: ignore[no-untyped-def] + """Drive the request through the real production dispatch, not a stand-in. - assert timeline == ["run_sync:http.maintenance.rebuild-index", "body"] + The previous version of this test replaced ``_handle_rebuild_index`` + wholesale with a body that itself called ``bridge.run_sync`` -- it only + proved the test's own stand-in called ``run_sync``, never that the real + production handler does anything of the kind (polylogue-ogn1 finding + #10). This exercises the real ``_do_post_impl`` route dispatch and the + real ``_handle_rebuild_index`` implementation end to end, with only the + typed rebuild service itself stubbed out. + """ + import json + from io import BytesIO + + from polylogue.maintenance.rebuild_index import RebuildIndexReceipt + + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(tmp_path)) + timeline: list[str] = [] + handler = _handler(["api", "maintenance", "rebuild-index"], timeline) + handler.server.write_bridge = _RecordingRebuildBridge(timeline) # type: ignore[assignment] + body = json.dumps({"promote": False, "raw_ids": ["raw-1"]}).encode("utf-8") + handler.headers = {"Content-Length": str(len(body))} # type: ignore[assignment] + handler.rfile = BytesIO(body) + + receipt = RebuildIndexReceipt( + archive_root=str(tmp_path), + raw_session_count=1, + selected_raw_count=1, + skipped_by_blob_limit_count=0, + status="replayed", + materialized=True, + materialization={}, + generation={"generation_id": "candidate-1", "active": False}, + readiness={"checked": True, "blocked_surface_count": 0}, + replay={"classified_full_count": 1, "replayed_logical_source_count": 1, "quarantined_raw_count": 0}, + ) + with patch("polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync", return_value=receipt) as rebuild: + with patch.object(handler, "_send_json") as send_json: + handler._do_post_impl() + + assert timeline == ["run_sync_with_timeout:http.maintenance.rebuild-index:600.0"] + request = rebuild.call_args.args[0] + assert request.raw_ids == ("raw-1",) + assert request.promote is False + assert send_json.call_args.args == (HTTPStatus.OK, receipt.to_dict()) @pytest.mark.parametrize("signal", ["traces", "metrics", "logs"]) diff --git a/tests/unit/daemon/test_maintenance_endpoints.py b/tests/unit/daemon/test_maintenance_endpoints.py index 3943ee8b34..f0682ac989 100644 --- a/tests/unit/daemon/test_maintenance_endpoints.py +++ b/tests/unit/daemon/test_maintenance_endpoints.py @@ -176,7 +176,10 @@ def test_rebuild_index_runs_the_typed_service_inside_the_route_executor(self, tm handler.server.write_bridge = type( "Bridge", (), - {"run_sync": lambda _self, _actor, function, *args: function(*args)}, + { + "run_sync": lambda _self, _actor, function, *args: function(*args), + "run_sync_with_timeout": lambda _self, _actor, _timeout, function, *args: function(*args), + }, )() with patch( "polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync", return_value=receipt @@ -188,6 +191,25 @@ def test_rebuild_index_runs_the_typed_service_inside_the_route_executor(self, tm assert request.promote is False assert send.call_args.args == (HTTPStatus.OK, receipt.to_dict()) + def test_rebuild_index_fails_closed_without_a_write_bridge(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def] + """polylogue-ogn1: a missing write_bridge must reject, never execute directly. + + A real ``DaemonAPIHTTPServer`` always installs ``write_bridge`` in its + constructor, so this can only happen for a bare stand-in server (as + used here). Regardless, the handler must not fall back to running the + rebuild outside the sole-writer coordinator -- that would bypass this + daemon's single-writer invariant for a destructive, authority- + promoting operation. + """ + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(tmp_path)) + handler = _make_handler("/api/maintenance/rebuild-index", body={"promote": False, "raw_ids": ["raw-1"]}) + assert getattr(handler.server, "write_bridge", None) is None + with patch("polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync") as rebuild: + with patch.object(handler, "_send_error") as send_error: + handler._handle_rebuild_index() + rebuild.assert_not_called() + send_error.assert_called_once_with(HTTPStatus.SERVICE_UNAVAILABLE, "write_coordinator_unavailable") + class TestMaintenanceRegistryEndpoints: """GET /api/maintenance/status/ and /api/maintenance/operations (#1197).""" diff --git a/tests/unit/daemon/test_write_coordinator.py b/tests/unit/daemon/test_write_coordinator.py index 01b028235b..9fbff1b2c5 100644 --- a/tests/unit/daemon/test_write_coordinator.py +++ b/tests/unit/daemon/test_write_coordinator.py @@ -648,3 +648,87 @@ async def _unexpected_operation() -> None: async def _return_ready() -> str: return "ready" + + +def test_thread_bridge_run_sync_uses_the_bridge_default_timeout() -> None: + """polylogue-ogn1 (#2/#5): the bare ``run_sync`` waits at most the bridge's own timeout. + + A blocking function that outlives the bridge's constructor timeout must + raise ``TimeoutError`` through ``run_sync`` -- proving the default path + is genuinely bounded, not merely documented as such. + """ + loop = asyncio.new_event_loop() + loop_ready = threading.Event() + coordinator_holder: list[DaemonWriteCoordinator] = [] + + def run_loop() -> None: + asyncio.set_event_loop(loop) + coordinator_holder.append(DaemonWriteCoordinator()) + loop_ready.set() + loop.run_forever() + + loop_thread = threading.Thread(target=run_loop, daemon=True) + loop_thread.start() + try: + assert loop_ready.wait(timeout=1.0) + coordinator = coordinator_holder[0] + bridge = DaemonWriteThreadBridge(coordinator, loop, timeout=0.05) + + def slow_write() -> str: + import time + + time.sleep(0.2) + return "too-late" + + with pytest.raises(TimeoutError): + bridge.run_sync("http.slow", slow_write) + + # Let the still-running background write actually finish before + # tearing down the loop, so the coordinator's task unwinds cleanly + # instead of being destroyed mid-flight (cosmetic only -- the + # TimeoutError above is the real assertion). + shutdown = asyncio.run_coroutine_threadsafe(coordinator.shutdown(timeout=1.0), loop) + assert shutdown.result(timeout=1.0) + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join(timeout=1.0) + + +def test_thread_bridge_run_sync_with_timeout_overrides_the_bridge_default() -> None: + """polylogue-ogn1 (#2/#5): a per-call override lets a long operation finish. + + ``run_sync_with_timeout`` must wait up to its own ``timeout`` argument + instead of the bridge's (shorter) constructor default -- this is the fix + for rebuild-index's HTTP route, which needs up to 600s while the bridge's + ordinary request timeout stays a much shorter 30s. + """ + loop = asyncio.new_event_loop() + loop_ready = threading.Event() + coordinator_holder: list[DaemonWriteCoordinator] = [] + + def run_loop() -> None: + asyncio.set_event_loop(loop) + coordinator_holder.append(DaemonWriteCoordinator()) + loop_ready.set() + loop.run_forever() + + loop_thread = threading.Thread(target=run_loop, daemon=True) + loop_thread.start() + try: + assert loop_ready.wait(timeout=1.0) + coordinator = coordinator_holder[0] + # The bridge's own default timeout is far shorter than the override + # below -- if the override were ignored, this would raise TimeoutError. + bridge = DaemonWriteThreadBridge(coordinator, loop, timeout=0.05) + + def slow_write() -> str: + import time + + time.sleep(0.2) + return "done" + + result = bridge.run_sync_with_timeout("http.maintenance.rebuild-index", 2.0, slow_write) + assert result == "done" + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join(timeout=1.0) diff --git a/tests/unit/maintenance/test_rebuild_index_selection.py b/tests/unit/maintenance/test_rebuild_index_selection.py new file mode 100644 index 0000000000..e9e71ab3c9 --- /dev/null +++ b/tests/unit/maintenance/test_rebuild_index_selection.py @@ -0,0 +1,104 @@ +"""Source-row selection helpers in ``maintenance/rebuild_index.py`` (polylogue-ogn1). + +Covers the CodeRabbit findings on PR #3076's rebuild-index coordination that +were still genuinely present against current source: + +- ``missing_index_raw_ids`` must treat every source row as missing when + ``index.db`` does not exist yet (fresh archive, or one just reset via + ``ops reset --index``), not silently return an empty selection -- a fresh + index has nothing indexed, so ``--only-missing`` must select the full + source set, matching ``all_index_rebuild_raw_ids``. +- ``validate_rebuild_index_request`` (the shared service's own validation, + not merely the CLI's) rejects ``max_blob_mb`` without ``raw_ids``/ + ``only_missing``, and rejects a partial selection asking to promote. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from polylogue.maintenance.rebuild_index import ( + RebuildIndexRequest, + all_index_rebuild_raw_ids, + missing_index_raw_ids, + validate_rebuild_index_request, +) +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + + +def _seed_source_raw_session(source_db: Path, *, raw_id: str, native_id: str, acquired_at_ms: int) -> None: + with sqlite3.connect(source_db) as conn: + conn.execute("PRAGMA foreign_keys = ON") + conn.execute( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, blob_hash, blob_size, acquired_at_ms + ) + VALUES (?, 'codex-session', ?, ?, zeroblob(32), 0, ?) + """, + (raw_id, native_id, f"/tmp/{native_id}.jsonl", acquired_at_ms), + ) + + +class TestMissingIndexRawIds: + def test_returns_every_raw_id_when_index_tier_does_not_exist_yet(self, tmp_path: Path) -> None: + """polylogue-ogn1 finding #6: a fresh/lost index has nothing indexed. + + Before this fix, ``missing_index_raw_ids`` short-circuited to ``[]`` + whenever ``index.db`` was absent -- which made ``--only-missing`` + rebuild nothing on a fresh archive or right after + ``ops reset --index``, exactly the case it exists to handle. + """ + source_db = tmp_path / "source.db" + initialize_archive_database(source_db, ArchiveTier.SOURCE) + _seed_source_raw_session(source_db, raw_id="raw-1", native_id="n1", acquired_at_ms=1000) + _seed_source_raw_session(source_db, raw_id="raw-2", native_id="n2", acquired_at_ms=2000) + + assert not (tmp_path / "index.db").exists() + assert missing_index_raw_ids(tmp_path) == all_index_rebuild_raw_ids(tmp_path) == ["raw-1", "raw-2"] + + def test_returns_empty_when_source_tier_does_not_exist(self, tmp_path: Path) -> None: + assert missing_index_raw_ids(tmp_path) == [] + + def test_excludes_raw_ids_already_materialized_in_an_existing_index(self, tmp_path: Path) -> None: + source_db = tmp_path / "source.db" + index_db = tmp_path / "index.db" + initialize_archive_database(source_db, ArchiveTier.SOURCE) + initialize_archive_database(index_db, ArchiveTier.INDEX) + _seed_source_raw_session(source_db, raw_id="raw-1", native_id="n1", acquired_at_ms=1000) + _seed_source_raw_session(source_db, raw_id="raw-2", native_id="n2", acquired_at_ms=2000) + with sqlite3.connect(index_db) as conn: + conn.execute("PRAGMA foreign_keys = ON") + conn.execute( + """ + INSERT INTO sessions ( + native_id, origin, raw_id, title, content_hash, created_at_ms, updated_at_ms + ) + VALUES ('n1', 'codex-session', 'raw-1', 'Session n1', zeroblob(32), 1000, 1000) + """ + ) + + # index.db exists and already has raw-1 materialized -- only raw-2 is missing. + assert missing_index_raw_ids(tmp_path) == ["raw-2"] + + +class TestValidateRebuildIndexRequestSharedService: + """polylogue-ogn1 findings #1/#7: enforced by the shared service, not just the CLI.""" + + def test_rejects_max_blob_mb_without_raw_ids_or_only_missing(self, tmp_path: Path) -> None: + request = RebuildIndexRequest(archive_root=tmp_path, max_blob_mb=10.0) + with pytest.raises(ValueError, match="--max-blob-mb requires --only-missing or --raw-id"): + validate_rebuild_index_request(request) + + def test_accepts_max_blob_mb_with_only_missing(self, tmp_path: Path) -> None: + request = RebuildIndexRequest(archive_root=tmp_path, only_missing=True, max_blob_mb=10.0, promote=False) + validate_rebuild_index_request(request) # must not raise + + def test_rejects_partial_selection_that_still_asks_to_promote(self, tmp_path: Path) -> None: + request = RebuildIndexRequest(archive_root=tmp_path, raw_ids=("raw-1",), promote=True) + with pytest.raises(ValueError, match="require --no-promote"): + validate_rebuild_index_request(request)