diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index 88a5d69ea..3224fcd96 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -947,13 +947,21 @@ async def _periodic_convergence_check( sources: tuple[WatchSource, ...], *, catch_up_complete: asyncio.Event | None = None, + catch_up_active: Callable[[], bool] | None = None, ) -> None: - """Periodically retry recorded derived convergence debt.""" + """Periodically retry recorded derived convergence debt. + + The archive-wide exact FTS audit is whole-archive work; while the watcher + is inside a chunked catch-up it is skipped (the catch-up's last chunk + publishes readiness) instead of rescanning the growing archive between + chunks. + """ db = _active_index_db_path() await _await_catch_up_gate(catch_up_complete, loop_name="convergence debt retry") while True: await _retry_convergence_debt_once(db) - await _run_periodic_fts_convergence_once(db) + if catch_up_active is None or not catch_up_active(): + await _run_periodic_fts_convergence_once(db) await asyncio.sleep(_CONVERGENCE_DEBT_RETRY_INTERVAL_SECONDS) @@ -3079,6 +3087,9 @@ async def _run_daemon_services_under_active_writer_lease( # generation before publishing either HTTP socket. Otherwise an # immediate similarity request can recreate legacy WAL sidecars after # the lifecycle checkpoint and turn a clean restart into a failure. + # Filled once the watcher exists; maintenance loops consult it for + # catch-up activity without holding the watcher before creation. + watcher_holder: list[LiveWatcher] = [] if not watcher_blocked: await _run_startup_embedding_lifecycle(write_coordinator, archive_root_path) if lifecycle_events_enabled: @@ -3163,7 +3174,11 @@ async def _run_daemon_services_under_active_writer_lease( catch_up_complete_gate = asyncio.Event() if enable_watch else None periodic_loops = [ _periodic_raw_materialization_convergence(catch_up_complete=catch_up_complete_gate), - _periodic_convergence_check(sources, catch_up_complete=catch_up_complete_gate), + _periodic_convergence_check( + sources, + catch_up_complete=catch_up_complete_gate, + catch_up_active=lambda: bool(watcher_holder and watcher_holder[0].catch_up_active), + ), _periodic_wal_checkpoint(), _periodic_fts_merge(), _periodic_heartbeat(), @@ -3223,6 +3238,7 @@ async def _run_daemon_services_under_active_writer_lease( catch_up_event_emitter=emit_catch_up_cycle, write_coordinator=write_coordinator, ) + watcher_holder.append(watcher) watcher_catch_up_complete = getattr(watcher, "catch_up_complete", None) if catch_up_complete_gate is not None and watcher_catch_up_complete is not None: maintenance_tasks.append( diff --git a/polylogue/daemon/convergence.py b/polylogue/daemon/convergence.py index f8f0c7ef1..edc72686c 100644 --- a/polylogue/daemon/convergence.py +++ b/polylogue/daemon/convergence.py @@ -82,6 +82,12 @@ class ConvergenceStage: barrier_check_sessions: Callable[[Sequence[str]], set[str]] | None = None # Optional bounded, secret-safe operator status payload. status: Callable[[], Mapping[str, object]] | None = None + # The stage's work is a function of the whole archive, not of the batch's + # subjects (a graph rebuilt from every raw artifact, an exact archive-wide + # readiness audit). ``converge_batch(whole_archive=False)`` skips such + # stages so a catch-up chunk's cost stays bounded by its own input; the + # catch-up's final chunk runs them once for the whole backlog. + whole_archive: bool = False @dataclass(slots=True) @@ -345,8 +351,19 @@ def _evict_converged_sessions(self, session_ids: Iterable[str]) -> None: if state is not None and state.converged: del self._session_states[session_id] - def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState], dict[str, float]]: - """Converge a changed source batch with per-subject stage barriers.""" + def converge_batch( + self, files: Iterable[Path], *, whole_archive: bool = True + ) -> tuple[dict[Path, FileState], dict[str, float]]: + """Converge a changed source batch with per-subject stage barriers. + + ``whole_archive=False`` bounds the pass to the batch's own subjects: + stages declared ``whole_archive`` are recorded ``SKIPPED`` (converged, + no debt) because their staleness is re-derived from archive content by + the next whole-archive pass, never from this batch's outcome. + + Stage ``check``/``check_many`` time is charged to ``.check`` in + the returned ledger so the batch's convergence time is fully attributed. + """ paths = tuple(dict.fromkeys(files)) if not paths: return {}, {} @@ -366,10 +383,15 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState], active_paths = tuple(path for path in paths if path not in blocked_paths) if not active_paths: continue + if stage.whole_archive and not whole_archive: + for path in active_paths: + self._file_states[path].stages[stage_name] = StageState.SKIPPED + continue if stage.check_many is None or stage.execute_many is None: for path in active_paths: state = self._file_states[path] + t_check = time.perf_counter() try: needs_work = stage.check(path) except Exception: @@ -382,6 +404,8 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState], state.stages[stage_name] = StageState.FAILED state.error_count += 1 continue + finally: + _record_stage_times(batch_stage_times, f"{stage_name}.check", time.perf_counter() - t_check, {}) if not needs_work: state.stages[stage_name] = StageState.DONE @@ -419,6 +443,7 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState], scope="stage", ) else: + t_check = time.perf_counter() try: batch_needs_work = set(stage.check_many(active_paths)).intersection(active_paths) except Exception: @@ -428,6 +453,7 @@ def converge_batch(self, files: Iterable[Path]) -> tuple[dict[Path, FileState], state.stages[stage_name] = StageState.FAILED state.error_count += 1 else: + _record_stage_times(batch_stage_times, f"{stage_name}.check", time.perf_counter() - t_check, {}) for path in active_paths: if path not in batch_needs_work: self._file_states[path].stages[stage_name] = StageState.DONE diff --git a/polylogue/daemon/convergence_stages.py b/polylogue/daemon/convergence_stages.py index 9adea49b7..fc8c37621 100644 --- a/polylogue/daemon/convergence_stages.py +++ b/polylogue/daemon/convergence_stages.py @@ -426,6 +426,7 @@ def execute_many(paths: Sequence[Path]) -> StageExecuteReturn: execute=execute, check_many=check_many, execute_many=execute_many, + whole_archive=True, ) @@ -480,6 +481,7 @@ def execute_many(paths: Sequence[Path]) -> StageExecuteReturn: execute=execute, check_many=check_many, execute_many=execute_many, + whole_archive=True, ) @@ -539,7 +541,6 @@ def execute(path: Path) -> StageExecuteReturn: session_ids=session_ids, page_size=_DAEMON_INSIGHT_REBUILD_PAGE_SIZE, ) - _record_fts_freshness_after_insights(conn) conn.commit() logger.info( "insights: refreshed sessions=%d profiles=%d work_events=%d phases=%d threads=%d", @@ -614,7 +615,6 @@ def execute_many(paths: Sequence[Path]) -> StageExecuteReturn: session_ids=session_ids, page_size=_DAEMON_INSIGHT_REBUILD_PAGE_SIZE, ) - _record_fts_freshness_after_insights(conn) conn.commit() logger.info( "insights: batch refreshed paths=%d sessions=%d profiles=%d work_events=%d phases=%d threads=%d", @@ -681,7 +681,6 @@ def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: session_ids=ids, page_size=_DAEMON_INSIGHT_REBUILD_PAGE_SIZE, ) - _record_fts_freshness_after_insights(conn) conn.commit() remaining = _stale_session_profile_ids(conn, ids) logger.info( @@ -1067,6 +1066,70 @@ def execute_many(_paths: Sequence[Path]) -> StageExecuteReturn: check_many=check_many, execute_many=execute_many, false_means_pending=True, + whole_archive=True, + ) + + +def make_fts_readiness_stage(db_path: Path) -> ConvergenceStage: + """Publish the exact archive-wide FTS readiness audit once per pass. + + The audit aggregates every ``blocks`` and ``messages_fts`` row, so it is + whole-archive work: partition-scoped FTS repair and per-session insight + rebuilds leave it to this stage, which runs after both, once per + whole-archive convergence pass. + """ + + def archive_db() -> Path: + return _active_archive_index_path(db_path) or db_path + + def publish() -> StageExecuteReturn: + database = archive_db() + if not database.exists(): + return True + try: + conn = _open_archive_insight_write_connection(database) + try: + _record_fts_freshness_after_insights(conn) + conn.commit() + finally: + conn.close() + except Exception as exc: + if _is_transient_sqlite_lock(exc): + logger.info("fts readiness: audit deferred because sqlite is busy: %s", exc) + return False + logger.warning("fts readiness: audit failed", exc_info=True) + raise + return True + + def check(path: Path) -> bool: + return archive_db().exists() + + def execute(path: Path) -> StageExecuteReturn: + return publish() + + def check_many(paths: Sequence[Path]) -> set[Path]: + return set(paths) if paths and archive_db().exists() else set() + + def execute_many(paths: Sequence[Path]) -> StageExecuteReturn: + return publish() if paths else True + + def check_sessions(session_ids: Sequence[str]) -> set[str]: + return set(session_ids) if session_ids and archive_db().exists() else set() + + def execute_sessions(session_ids: Sequence[str]) -> StageExecuteReturn: + return publish() if session_ids else True + + return ConvergenceStage( + name="fts_readiness", + description="Publish the exact archive-wide FTS readiness audit", + check=check, + execute=execute, + check_many=check_many, + execute_many=execute_many, + check_sessions=check_sessions, + execute_sessions=execute_sessions, + false_means_pending=True, + whole_archive=True, ) @@ -1107,6 +1170,7 @@ def make_default_convergence_stages( make_claude_workflow_stage(db_path), make_delegation_work_evidence_stage(db_path), make_derived_stage(db_path), + make_fts_readiness_stage(db_path), make_standing_query_stage(db_path, evaluator=ArchiveCanonicalPlanEvaluator(db_path)), ) ) @@ -2449,9 +2513,8 @@ def _archive_insights_execute_ids( finally: if marker_conn is not None: marker_conn.close() - # The rebuild commits its own rows. Publish and commit the final exact FTS - # state in the same production stage before reporting success. - _record_fts_freshness_after_insights(conn) + # The rebuild commits its own rows; the exact archive-wide FTS audit is + # published once per whole-archive pass by ``make_fts_readiness_stage``. conn.commit() remaining = _archive_stale_session_profile_ids(conn, list(session_ids)) logger.info( @@ -2474,6 +2537,7 @@ def _archive_insights_execute_ids( "make_delegation_work_evidence_stage", "make_default_convergence_stages", "make_embed_stage", + "make_fts_readiness_stage", "make_fts_stage", "make_derived_stage", "make_raw_authority_verdict_cache_stage", diff --git a/polylogue/daemon/fts_convergence.py b/polylogue/daemon/fts_convergence.py index c10990567..a07c63289 100644 --- a/polylogue/daemon/fts_convergence.py +++ b/polylogue/daemon/fts_convergence.py @@ -80,7 +80,11 @@ def run_once_sync( ) with open_daemon_connection(self._db_path, timeout=30.0) as conn: result = FtsDerivationAdapter().converge(conn, keys=partition_keys) - if result.outcome is FtsOutcome.DONE: + # The readiness projection is an exact archive-wide audit; + # a partition-scoped pass cannot claim it and must not pay + # for it (``make_fts_readiness_stage`` publishes it once per + # whole-archive convergence pass). + if result.outcome is FtsOutcome.DONE and partition_keys is None: self._publish_readiness_projection(conn) state = { FtsOutcome.DONE: FtsOwnerState.READY_EXACT, diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 67161a6ee..a5d1397e2 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -759,8 +759,14 @@ async def ingest_files( skipped_file_count: int = 0, emit_event: bool = True, max_pass_seconds: float | None = None, + whole_archive_convergence: bool = True, ) -> LiveBatchMetrics: - """Ingest files in batch, run post-ingest convergence, and return metrics.""" + """Ingest files in batch, run post-ingest convergence, and return metrics. + + ``whole_archive_convergence=False`` bounds post-ingest convergence to + this batch's own subjects (a catch-up chunk); the caller runs one + whole-archive pass at the end of its catch-up. + """ authorization = self.require_cursor_authority(paths) refused_paths = self._refused_paths self._refused_paths = frozenset() @@ -906,6 +912,8 @@ async def flush_append_plans() -> None: "watcher.live_ingest.append_convergence", self._converge_paths, [plan.path for plan in append_result.succeeded], + whole_archive=whole_archive_convergence, + session_ids=tuple(append_result.session_ids_by_path.values()), ) convergence_time_s += elapsed release_process_memory() @@ -1156,6 +1164,8 @@ async def flush_append_plans() -> None: "watcher.live_ingest.full_convergence", self._converge_paths, full_result.succeeded, + whole_archive=whole_archive_convergence, + session_ids=full_result.changed_session_ids, ) convergence_time_s += elapsed release_process_memory() @@ -1733,7 +1743,11 @@ def _record_convergence_outcome(self, path: Path, debts: Iterable[ConvergenceDeb record_convergence_outcome(self._cursor, path, debts, archive_root=archive_root) def _converge_paths( - self, paths: Iterable[Path] + self, + paths: Iterable[Path], + *, + whole_archive: bool = True, + session_ids: Iterable[str] = (), ) -> tuple[set[Path], float, dict[str, float], list[ConvergenceDebt]]: unique_paths = tuple(sorted(dict.fromkeys(paths))) if not unique_paths: @@ -1745,23 +1759,33 @@ def _converge_paths( try: converge_batch = getattr(self._converger, "converge_batch", None) if callable(converge_batch): - states, timings = converge_batch(unique_paths) + # The keyword is passed only when it narrows the pass, so + # convergers without the parameter keep their whole-archive + # default. + states, timings = ( + converge_batch(unique_paths) if whole_archive else converge_batch(unique_paths, whole_archive=False) + ) batch_completed = { path for path in unique_paths if path in states and bool(getattr(states[path], "converged", False)) } debt_items = convergence_debt_from_states(unique_paths, states) - # #1654: after convergence, check for new hook events - # that carry paste evidence and update matching messages. + batch_stage_timings = {stage_name: float(elapsed) for stage_name, elapsed in timings.items()} + # #1654: after convergence, check for new hook events that + # carry paste evidence and update matching messages. Scoped to + # this batch's sessions so the scan is bounded by the batch, + # not by the archive's whole hook history. + t_paste = time.perf_counter() try: from polylogue.sources.live.hook_paste_enrichment import enrich_paste_from_hooks - enrich_paste_from_hooks(self._cursor._db_path) + enrich_paste_from_hooks(self._cursor._db_path, session_ids=tuple(dict.fromkeys(session_ids))) except Exception: logger.debug("hook_paste: enrichment failed (non-fatal)", exc_info=True) + batch_stage_timings["hook_paste_enrichment"] = time.perf_counter() - t_paste return ( batch_completed, time.perf_counter() - started, - {stage_name: float(elapsed) for stage_name, elapsed in timings.items()}, + batch_stage_timings, debt_items, ) diff --git a/polylogue/sources/live/hook_paste_enrichment.py b/polylogue/sources/live/hook_paste_enrichment.py index a7aa3d5c6..1be414a01 100644 --- a/polylogue/sources/live/hook_paste_enrichment.py +++ b/polylogue/sources/live/hook_paste_enrichment.py @@ -15,6 +15,7 @@ import json import sqlite3 +from collections.abc import Iterable from hashlib import sha256 from pathlib import Path @@ -32,12 +33,32 @@ _TIMESTAMP_TOLERANCE_MS = 3000 -def _iter_hook_paste_events(hooks_dir: Path) -> list[dict[str, object]]: +def _sidecar_paths(hooks_dir: Path, session_ids: Iterable[str] | None) -> list[Path]: + """Sidecar journals to scan: every one, or only the given sessions'. + + ``polylogue-hook`` journals each session to ``-.jsonl`` + in the sidecar dir, so a session's paste events live in the one file named + by its native id. Scoping to the batch's sessions keeps the scan bounded by + the batch instead of by the archive's whole hook history. + """ + if not hooks_dir.exists(): + return [] + if session_ids is None: + return sorted(hooks_dir.glob("*.jsonl")) + paths: dict[Path, None] = {} + for session_id in session_ids: + native_id = str(session_id).split(":", 1)[-1] + if not native_id: + continue + for candidate in sorted(hooks_dir.glob(f"*-{native_id}.jsonl")): + paths[candidate] = None + return list(paths) + + +def _iter_hook_paste_events(hooks_dir: Path, session_ids: Iterable[str] | None = None) -> list[dict[str, object]]: """Scan hook sidecar JSONL files, return UserPromptSubmit events with paste.""" events: list[dict[str, object]] = [] - if not hooks_dir.exists(): - return events - for jsonl_path in hooks_dir.glob("*.jsonl"): + for jsonl_path in _sidecar_paths(hooks_dir, session_ids): try: with open(jsonl_path, encoding="utf-8") as fh: for line in fh: @@ -163,7 +184,7 @@ def _enrich_archive_paste_from_hooks(index_db: Path, events: list[dict[str, obje return updated -def enrich_paste_from_hooks(db_path: Path) -> int: +def enrich_paste_from_hooks(db_path: Path, *, session_ids: Iterable[str] | None = None) -> int: """Scan hook sidecar files and update has_paste on matching messages. ``db_path`` is the caller's ops.db path (``archive_root / "ops.db"``), so @@ -172,9 +193,12 @@ def enrich_paste_from_hooks(db_path: Path) -> int: archive it was actually called for, never a different one that happens to be the process-wide default. + ``session_ids`` (archive ``origin:native_id`` ids) bounds the scan to those + sessions' sidecar journals; ``None`` scans every journal in the directory. + Returns the number of messages updated. """ - events = _iter_hook_paste_events(db_path.parent / "hooks") + events = _iter_hook_paste_events(db_path.parent / "hooks", session_ids) if not events: return 0 diff --git a/polylogue/sources/live/watcher.py b/polylogue/sources/live/watcher.py index 6140135de..87d1e2457 100644 --- a/polylogue/sources/live/watcher.py +++ b/polylogue/sources/live/watcher.py @@ -333,6 +333,7 @@ def __init__( self._ingest_lock = asyncio.Lock() self._stop = asyncio.Event() self._catch_up_complete = asyncio.Event() + self._catch_up_active = False self._archived_cursor_conns: tuple[sqlite3.Connection, sqlite3.Connection] | None = None # Set once per reconciliation scope: True when the index tier has no # materialized sessions at all despite source.db holding successfully @@ -370,6 +371,15 @@ async def _run_writer_sync( def catch_up_complete(self) -> asyncio.Event: return self._catch_up_complete + @property + def catch_up_active(self) -> bool: + """Whether a chunked catch-up ingest loop is running right now. + + Whole-archive maintenance passes wait for it to finish rather than + repeating archive-wide work between chunks. + """ + return self._catch_up_active + def _existing_source_roots(self) -> list[Path]: """Return configured roots that exist at the instant of a scan.""" return [source.root for source in self._sources if source.exists()] @@ -590,9 +600,10 @@ async def _catch_up(self, roots: list[Path]) -> None: len(hot_candidates), ) for group in (hot_candidates, cold_candidates): - if self._stop.is_set() or not group: + if self._stop.is_set(): break - await self._catch_up_candidates(group) + if group: + await self._catch_up_candidates(group) await self._drain_hook_spools() async def _catch_up_candidates(self, candidates: tuple[CandidateSourceFile, ...]) -> None: @@ -652,6 +663,7 @@ async def prepare_catch_up() -> None: plan.skipped_file_count, len(chunks), ) + self._catch_up_active = True try: for index, chunk in enumerate(chunks, start=1): if self._stop.is_set(): @@ -675,11 +687,16 @@ async def ingest_chunk( chunk_paths: list[Path] = chunk_paths, ) -> None: nonlocal attempted, ingested, failed - metrics = await self._ingest_files( - chunk_paths, - queued_file_count=len(plan.candidates) if chunk_index == 1 else len(chunk_paths), - skipped_file_count=plan.skipped_file_count if chunk_index == 1 else 0, - ) + # Whole-archive convergence stages run once, on the last + # chunk, so each earlier chunk pays only for its own + # subjects. + ingest_kwargs: dict[str, Any] = { + "queued_file_count": len(plan.candidates) if chunk_index == 1 else len(chunk_paths), + "skipped_file_count": plan.skipped_file_count if chunk_index == 1 else 0, + } + if chunk_index != len(chunks): + ingest_kwargs["whole_archive_convergence"] = False + metrics = await self._ingest_files(chunk_paths, **ingest_kwargs) if metrics is not None: _log_ingest_metrics(f"live.watcher: catch-up chunk {chunk_index}/{len(chunks)}", metrics) # Keep the catch-up coordinator compatible with older @@ -753,6 +770,8 @@ async def ingest_chunk( operation_id, "failure", plan, attempted, ingested, failed, stage_timings_s, cycle_started ) raise + finally: + self._catch_up_active = False def _hook_sources(self) -> tuple[WatchSource, ...]: """Return the declared hook topology, preserving configured order. @@ -1701,6 +1720,7 @@ async def _ingest_files( *, queued_file_count: int | None = None, skipped_file_count: int = 0, + whole_archive_convergence: bool = True, ) -> LiveBatchMetrics: """Ingest files through the reusable daemon live batch processor.""" self._batch_processor.require_cursor_authority(paths) @@ -1712,6 +1732,7 @@ async def ingest() -> LiveBatchMetrics: queued_file_count=queued_file_count, skipped_file_count=skipped_file_count, max_pass_seconds=_LIVE_INGEST_MAX_PASS_SECONDS, + whole_archive_convergence=whole_archive_convergence, ) run = getattr(self._write_coordinator, "run", None) diff --git a/polylogue/storage/fts/derivation.py b/polylogue/storage/fts/derivation.py index e91a8b53b..ad3f8d6c8 100644 --- a/polylogue/storage/fts/derivation.py +++ b/polylogue/storage/fts/derivation.py @@ -142,6 +142,16 @@ def _schema_compatible(conn: sqlite3.Connection) -> bool: return row is not None and int(row[0]) == len(expected) +def _session_block_id_range(key: str) -> tuple[str, str]: + """Half-open ``block_id`` range covering exactly one session's blocks. + + ``block_id`` is ``session_id || ':' || ...``; ``';'`` is the code point + after ``':'``, so ``[key || ':', key || ';')`` selects the session's rows + through the ``block_id`` UNIQUE index instead of a ``substr`` table scan. + """ + return f"{key}:", f"{key};" + + def _digest(rows: Sequence[FtsInputRow]) -> str: payload = [ [ @@ -299,10 +309,10 @@ def inspect(self, conn: sqlite3.Connection, key: str) -> FtsPartitionInspection: SELECT COUNT(*) FROM messages_fts_identity AS i JOIN messages_fts_docsize AS d ON d.id = i.rowid LEFT JOIN blocks AS b ON b.block_id = i.block_id - WHERE substr(i.block_id, 1, length(?) + 1) = ? || ':' + WHERE i.block_id >= ? AND i.block_id < ? AND (b.block_id IS NULL OR b.session_id != ? OR b.search_text = '') """, - (key, key, key), + (*_session_block_id_range(key), key), ).fetchone()[0] ) wrong_rows = int( @@ -329,10 +339,10 @@ def inspect(self, conn: sqlite3.Connection, key: str) -> FtsPartitionInspection: duplicate_sql = ( "SELECT COALESCE(SUM(n - 1), 0) FROM (" "SELECT block_id, COUNT(*) AS n FROM messages_fts_identity " - "WHERE substr(block_id, 1, length(?) + 1) = ? || ':' " + "WHERE block_id >= ? AND block_id < ? " "GROUP BY block_id HAVING n > 1)" ) - duplicate_params = (key, key) + duplicate_params = _session_block_id_range(key) duplicate_rows = int(conn.execute(duplicate_sql, duplicate_params).fetchone()[0]) status = FtsKeyStatus.VALID detail: str | None = None @@ -397,9 +407,9 @@ def publish(self, conn: sqlite3.Connection, computed: FtsPartitionInput) -> bool for row in conn.execute( """ SELECT i.rowid FROM messages_fts_identity AS i - WHERE substr(i.block_id, 1, length(?) + 1) = ? || ':' + WHERE i.block_id >= ? AND i.block_id < ? """, - (computed.key, computed.key), + _session_block_id_range(computed.key), ) ) if rowids: diff --git a/polylogue/storage/raw_authority_verdict_cache.py b/polylogue/storage/raw_authority_verdict_cache.py index 1a5cce673..73fce0a72 100644 --- a/polylogue/storage/raw_authority_verdict_cache.py +++ b/polylogue/storage/raw_authority_verdict_cache.py @@ -98,21 +98,35 @@ def find_raw_authority_verdict_cache_work( ``max_cohorts`` bounds returned work. Append fragments share the same cache now that the projection reads their persisted byte-authority links. """ - rows = conn.execute( + # One pass over each table instead of two point queries per cohort: the + # staleness rule is the one ``_read_cached_raw_authority_verdicts_from_connection`` + # applies per key (missing rows, member-count drift, fingerprint drift). + current_rows: dict[str, list[tuple[str, str, str, str, str]]] = {} + for row in conn.execute( """ - SELECT logical_source_key + SELECT logical_source_key, raw_id, revision_kind, lower(hex(blob_hash)), revision_authority, + COALESCE(predecessor_raw_id, '') FROM raw_sessions WHERE logical_source_key IS NOT NULL AND logical_source_key != '' - GROUP BY logical_source_key ORDER BY logical_source_key """ - ).fetchall() + ): + current_rows.setdefault(str(row[0]), []).append( + (str(row[1]), str(row[2]), str(row[3]), str(row[4]), str(row[5])) + ) + cached_rows: dict[str, list[bytes]] = {} + for logical_source_key, cached_fingerprint in conn.execute( + "SELECT logical_source_key, cohort_fingerprint FROM raw_authority_verdicts" + ): + cached_rows.setdefault(str(logical_source_key), []).append(bytes(cached_fingerprint)) pending: list[str] = [] - for (logical_source_key,) in rows: - key = str(logical_source_key) - if _read_cached_raw_authority_verdicts_from_connection(conn, key) is not None: - continue + for key, rows in current_rows.items(): + cached = cached_rows.get(key) + if cached is not None and len(cached) == len(rows): + fingerprint = _cohort_fingerprint(rows) + if all(cached_fingerprint == fingerprint for cached_fingerprint in cached): + continue if max_cohorts is None or len(pending) < max_cohorts: pending.append(key) return RawAuthorityVerdictCacheWork(tuple(pending)) diff --git a/tests/unit/daemon/test_catch_up_chunk_cost.py b/tests/unit/daemon/test_catch_up_chunk_cost.py new file mode 100644 index 000000000..a46010304 --- /dev/null +++ b/tests/unit/daemon/test_catch_up_chunk_cost.py @@ -0,0 +1,223 @@ +"""A catch-up chunk's convergence cost is bounded by the chunk, not the archive. + +The live batch processor runs the daemon's real convergence stages after each +catch-up chunk. Archive-wide work (the exact FTS readiness audit, the raw +authority verdict warmer, graph rebuilds from every raw artifact, a scan of +every hook sidecar journal) must not be paid per chunk: it runs once, on the +catch-up's final chunk. + +Anti-vacuity: the statement count is measured inside ``_converge_paths`` only. +Reverting any ``whole_archive`` deferral (raw authority: two statements per +cohort; Claude workflow: one per artifact row) makes the large archive's chunk +issue more statements than the small archive's, reverting the FTS readiness +deferral makes the snapshot spy fire, and reverting sidecar scoping makes the +scan touch every session's journal. +""" + +from __future__ import annotations + +import asyncio +import json +import sqlite3 +from pathlib import Path +from types import SimpleNamespace +from typing import Any, cast + +import pytest + +from polylogue.daemon import convergence_stages +from polylogue.daemon.convergence import DaemonConverger +from polylogue.daemon.convergence_stages import make_default_convergence_stages +from polylogue.sources.live import hook_paste_enrichment +from polylogue.sources.live.batch import LiveBatchProcessor +from polylogue.sources.live.cursor import CursorStore +from polylogue.sources.live.watcher import WatchSource + +_MESSAGES_PER_SESSION = 8 + + +def _session_records(uuid: str) -> list[dict[str, object]]: + records: list[dict[str, object]] = [] + for index in range(_MESSAGES_PER_SESSION): + is_user = index % 2 == 0 + records.append( + { + "parentUuid": None if index == 0 else f"{uuid}-msg-{index - 1:04d}", + "sessionId": uuid, + "type": "user" if is_user else "assistant", + "message": { + "role": "user" if is_user else "assistant", + "content": f"Synthetic message {index} of {uuid}. Searchable prose about convergence cost.", + }, + "uuid": f"{uuid}-msg-{index:04d}", + "timestamp": f"2026-05-05T00:{index // 60:02d}:{index % 60:02d}.000Z", + "cwd": "/realm/project/polylogue", + "version": "1.0.6", + "isSidechain": False, + "userType": "external", + } + ) + return records + + +def _write_session(corpus_root: Path, hooks_dir: Path, ordinal: int) -> Path: + uuid = f"deadbeef-0000-0000-0000-{ordinal:012x}" + path = corpus_root / "test-project" / f"{uuid}.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("".join(json.dumps(record) + "\n" for record in _session_records(uuid)), encoding="utf-8") + (hooks_dir / f"claude-code-{uuid}.jsonl").write_text( + json.dumps( + { + "event_type": "UserPromptSubmit", + "timestamp": "2026-05-05T00:00:00Z", + "payload": {"session_id": uuid, "prompt": "Inspect [Pasted text #1]"}, + } + ) + + "\n", + encoding="utf-8", + ) + return path + + +class _Polylogue: + def __init__(self, archive_root: Path, db_path: Path) -> None: + self.archive_root = archive_root + self.backend = SimpleNamespace(db_path=db_path) + + +class _ChunkProbe: + """Statements issued and sidecars read while one chunk converges.""" + + def __init__(self, monkeypatch: pytest.MonkeyPatch) -> None: + self.statements = 0 + self.sidecars_read = 0 + self.snapshot_calls = 0 + self.active = False + real_connect = sqlite3.connect + real_sidecar_paths = hook_paste_enrichment._sidecar_paths + real_snapshot = convergence_stages._record_fts_freshness_after_insights + + def counting_connect(*args: Any, **kwargs: Any) -> sqlite3.Connection: + conn = cast(sqlite3.Connection, real_connect(*args, **kwargs)) + conn.set_trace_callback(self._count_statement) + return conn + + def counting_sidecar_paths(hooks_dir: Path, session_ids: Any) -> list[Path]: + paths = real_sidecar_paths(hooks_dir, session_ids) + if self.active: + self.sidecars_read += len(paths) + return paths + + def counting_snapshot(conn: sqlite3.Connection) -> bool: + if self.active: + self.snapshot_calls += 1 + return real_snapshot(conn) + + monkeypatch.setattr(sqlite3, "connect", counting_connect) + monkeypatch.setattr(hook_paste_enrichment, "_sidecar_paths", counting_sidecar_paths) + monkeypatch.setattr(convergence_stages, "_record_fts_freshness_after_insights", counting_snapshot) + + def _count_statement(self, sql: str) -> None: + # SQLite reports its own virtual-table maintenance (FTS5 segment + # merges) as ``--``-prefixed statements; only product statements + # measure the convergence pass. + if self.active and not sql.lstrip().startswith("--"): + self.statements += 1 + + +def _build( + tmp_path: Path, + *, + monkeypatch: pytest.MonkeyPatch, + seeded_sessions: int, +) -> tuple[LiveBatchProcessor, Path, Path]: + archive_root = tmp_path / f"archive-{seeded_sessions}" + archive_root.mkdir() + hooks_dir = archive_root / "hooks" + hooks_dir.mkdir() + corpus_root = tmp_path / f"corpus-{seeded_sessions}" + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(archive_root)) + monkeypatch.setenv("POLYLOGUE_CONFIG", str(archive_root / "polylogue.toml")) + db_path = archive_root / "index.db" + converger = DaemonConverger(stages=make_default_convergence_stages(db_path)) + processor = LiveBatchProcessor( + cast(Any, _Polylogue(archive_root, db_path)), + (WatchSource(name="claude-code", root=corpus_root),), + cursor=CursorStore(db_path), + parser_fingerprint="chunk-cost-v1", + converger=converger, + ) + seeded = [_write_session(corpus_root, hooks_dir, ordinal) for ordinal in range(seeded_sessions)] + metrics = asyncio.run(processor.ingest_files(seeded, emit_event=False)) + assert metrics.succeeded_file_count == seeded_sessions + return processor, corpus_root, hooks_dir + + +def _converge_chunk( + processor: LiveBatchProcessor, + probe: _ChunkProbe, + paths: list[Path], + *, + whole_archive: bool, +) -> Any: + real_converge = processor._converge_paths + + def probed_converge(*args: Any, **kwargs: Any) -> Any: + probe.active = True + try: + return real_converge(*args, **kwargs) + finally: + probe.active = False + + processor._converge_paths = probed_converge # type: ignore[method-assign] + try: + return asyncio.run(processor.ingest_files(paths, emit_event=False, whole_archive_convergence=whole_archive)) + finally: + processor._converge_paths = real_converge # type: ignore[method-assign] + + +@pytest.mark.parametrize("chunk_files", [2]) +def test_chunk_convergence_cost_does_not_grow_with_archive_size( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, chunk_files: int +) -> None: + small_sessions, large_sessions = 2, 14 + probe = _ChunkProbe(monkeypatch) + results: dict[int, tuple[int, int, int, dict[str, float]]] = {} + for seeded_sessions in (small_sessions, large_sessions): + processor, corpus_root, hooks_dir = _build(tmp_path, monkeypatch=monkeypatch, seeded_sessions=seeded_sessions) + chunk = [_write_session(corpus_root, hooks_dir, seeded_sessions + offset) for offset in range(chunk_files)] + probe.statements = probe.sidecars_read = probe.snapshot_calls = 0 + metrics = _converge_chunk(processor, probe, chunk, whole_archive=False) + assert metrics.succeeded_file_count == chunk_files + results[seeded_sessions] = ( + probe.statements, + probe.sidecars_read, + probe.snapshot_calls, + dict(metrics.stage_timings_s), + ) + + small_statements, small_sidecars, small_snapshots, small_stages = results[small_sessions] + large_statements, large_sidecars, large_snapshots, large_stages = results[large_sessions] + deferred = {"raw_authority_verdict_cache", "claude_workflow", "delegation_work_evidence", "fts_readiness"} + + assert small_sidecars == large_sidecars == chunk_files + assert small_snapshots == large_snapshots == 0 + assert not deferred & set(small_stages) and not deferred & set(large_stages) + assert "hook_paste_enrichment" in large_stages + # A chunk's statements are a function of its own files; the slack covers + # per-chunk variance and stays far below one deferred stage's growth + # (two statements per cohort, one per artifact row). + assert large_statements <= small_statements + 4, (small_statements, large_statements) + + +def test_final_catch_up_chunk_runs_the_whole_archive_stages(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + probe = _ChunkProbe(monkeypatch) + processor, corpus_root, hooks_dir = _build(tmp_path, monkeypatch=monkeypatch, seeded_sessions=2) + chunk = [_write_session(corpus_root, hooks_dir, 2)] + + probe.statements = probe.sidecars_read = probe.snapshot_calls = 0 + metrics = _converge_chunk(processor, probe, chunk, whole_archive=True) + + assert metrics.succeeded_file_count == 1 + assert probe.snapshot_calls == 1 + assert {"fts_readiness", "raw_authority_verdict_cache"} <= set(metrics.stage_timings_s) diff --git a/tests/unit/daemon/test_convergence_stages.py b/tests/unit/daemon/test_convergence_stages.py index 5253fc619..978504349 100644 --- a/tests/unit/daemon/test_convergence_stages.py +++ b/tests/unit/daemon/test_convergence_stages.py @@ -1171,6 +1171,7 @@ def test_default_convergence_stages_always_register_embed_stage( "claude_workflow", "delegation_work_evidence", "derived", + "fts_readiness", "standing-queries", ] diff --git a/tests/unit/daemon/test_daemon_convergence.py b/tests/unit/daemon/test_daemon_convergence.py index 62c252d8b..a4b89e7cb 100644 --- a/tests/unit/daemon/test_daemon_convergence.py +++ b/tests/unit/daemon/test_daemon_convergence.py @@ -70,7 +70,7 @@ def execute_many(candidates: Sequence[Path]) -> bool: assert checked == [tuple(paths)] assert [set(candidates) for candidates in executed] == [set(paths)] assert set(states) == set(paths) - assert set(stage_times) == {"derived"} + assert set(stage_times) == {"derived", "derived.check"} assert all(state.converged for state in states.values()) assert converger._file_states == {} @@ -247,3 +247,65 @@ def check_sessions(session_ids: Sequence[str]) -> set[str]: assert states["conv-b"].stages["embed"] is StageState.PENDING assert states["conv-b"].last_error == "session stage embed returned False" assert set(converger._session_states) == {"conv-b"} + + +def test_converge_batch_chunk_scope_skips_whole_archive_stages(tmp_path: Path) -> None: + """Anti-vacuity: dropping the ``whole_archive`` skip runs the archive-wide stage per chunk.""" + paths = [tmp_path / "a.jsonl", tmp_path / "b.jsonl"] + for path in paths: + path.write_text("{}\n", encoding="utf-8") + archive_wide_checks: list[tuple[Path, ...]] = [] + archive_wide_runs: list[tuple[Path, ...]] = [] + scoped_runs: list[tuple[Path, ...]] = [] + + def archive_wide_check_many(candidates: Sequence[Path]) -> set[Path]: + archive_wide_checks.append(tuple(candidates)) + return set(candidates) + + def archive_wide_execute_many(candidates: Sequence[Path]) -> bool: + archive_wide_runs.append(tuple(candidates)) + return True + + def scoped_execute_many(candidates: Sequence[Path]) -> bool: + scoped_runs.append(tuple(candidates)) + return True + + converger = DaemonConverger( + [ + ConvergenceStage( + name="graph", + description="rebuilt from every raw artifact", + check=lambda _path: True, + execute=lambda _path: True, + check_many=archive_wide_check_many, + execute_many=archive_wide_execute_many, + whole_archive=True, + ), + ConvergenceStage( + name="derived", + description="bounded by the batch's subjects", + check=lambda _path: True, + execute=lambda _path: True, + check_many=lambda candidates: set(candidates), + execute_many=scoped_execute_many, + ), + ] + ) + + chunk_states, chunk_timings = converger.converge_batch(paths, whole_archive=False) + + assert archive_wide_checks == [] + assert archive_wide_runs == [] + assert scoped_runs == [tuple(paths)] + assert all(state.stages["graph"] is StageState.SKIPPED for state in chunk_states.values()) + assert all(state.stages["derived"] is StageState.DONE for state in chunk_states.values()) + assert all(state.converged for state in chunk_states.values()) + assert "graph" not in chunk_timings + assert "derived.check" in chunk_timings + + final_states, final_timings = converger.converge_batch(paths) + + assert archive_wide_checks == [tuple(paths)] + assert archive_wide_runs == [tuple(paths)] + assert all(state.stages["graph"] is StageState.DONE for state in final_states.values()) + assert {"graph", "graph.check", "derived", "derived.check"} <= set(final_timings) diff --git a/tests/unit/sources/test_hook_paste_enrichment.py b/tests/unit/sources/test_hook_paste_enrichment.py index 24553781b..73d3ac611 100644 --- a/tests/unit/sources/test_hook_paste_enrichment.py +++ b/tests/unit/sources/test_hook_paste_enrichment.py @@ -109,3 +109,69 @@ def test_hook_paste_enrichment_never_reads_a_sibling_archives_hooks_dir( updated = hook_paste_enrichment.enrich_paste_from_hooks(scratch_ops_db) assert updated == 0 + + +def _seed_paste_candidate(index_db: Path, native_id: str, hook_time_ms: int) -> None: + with sqlite3.connect(index_db) as conn: + conn.execute( + """ + INSERT INTO sessions ( + native_id, origin, content_hash, created_at_ms, updated_at_ms + ) VALUES (?, 'codex-session', ?, ?, ?) + """, + (native_id, native_id.encode().ljust(32, b"s")[:32], hook_time_ms, hook_time_ms), + ) + conn.execute( + """ + INSERT INTO messages ( + session_id, native_id, position, role, content_hash, occurred_at_ms + ) VALUES (?, 'm1', 0, 'user', ?, ?) + """, + (f"codex-session:{native_id}", native_id.encode().ljust(32, b"m")[:32], hook_time_ms + 100), + ) + + +def _write_sidecar(hooks_dir: Path, native_id: str) -> Path: + path = hooks_dir / f"codex-{native_id}.jsonl" + path.write_text( + json.dumps( + { + "event_type": "UserPromptSubmit", + "timestamp": "2026-05-07T12:00:00Z", + "payload": {"session_id": native_id, "prompt": "Inspect [Pasted text #1]"}, + } + ) + + "\n", + encoding="utf-8", + ) + return path + + +def test_hook_paste_enrichment_reads_only_the_batch_sessions_sidecars(tmp_path: Path) -> None: + """Anti-vacuity: a scan of every sidecar journal would enrich the untouched session too. + + A session's hook journal is ``-.jsonl``; the batch + passes its archive session ids and only those journals are read, so the + scan is bounded by the batch instead of the archive's whole hook history. + """ + index_db = tmp_path / "index.db" + initialize_archive_database(index_db, ArchiveTier.INDEX) + hook_time_ms = int(datetime(2026, 5, 7, 12, 0, tzinfo=UTC).timestamp() * 1000) + _seed_paste_candidate(index_db, "batch-native", hook_time_ms) + _seed_paste_candidate(index_db, "other-native", hook_time_ms) + hooks_dir = tmp_path / "hooks" + hooks_dir.mkdir() + batch_sidecar = _write_sidecar(hooks_dir, "batch-native") + _write_sidecar(hooks_dir, "other-native") + + assert hook_paste_enrichment._sidecar_paths(hooks_dir, ("codex-session:batch-native",)) == [batch_sidecar] + assert len(hook_paste_enrichment._sidecar_paths(hooks_dir, None)) == 2 + + updated = hook_paste_enrichment.enrich_paste_from_hooks( + tmp_path / "ops.db", session_ids=("codex-session:batch-native",) + ) + + assert updated == 1 + with sqlite3.connect(index_db) as conn: + rows = conn.execute("SELECT session_id, has_paste FROM messages ORDER BY session_id").fetchall() + assert rows == [("codex-session:batch-native", 1), ("codex-session:other-native", 0)] diff --git a/tests/unit/sources/test_live_batch_convergence.py b/tests/unit/sources/test_live_batch_convergence.py index 50437eda9..4e891dea8 100644 --- a/tests/unit/sources/test_live_batch_convergence.py +++ b/tests/unit/sources/test_live_batch_convergence.py @@ -80,7 +80,8 @@ def test_live_batch_converges_known_paths_by_source_path(tmp_path: Path) -> None assert completed == {source} assert elapsed >= 0.0 - assert timings == {"batch": 1.0} + assert timings["batch"] == 1.0 + assert timings["hook_paste_enrichment"] >= 0.0 assert debts == [] assert converger.session_calls == [] assert converger.batch_calls == [(source,)] diff --git a/tests/unit/sources/test_live_catchup_planning.py b/tests/unit/sources/test_live_catchup_planning.py index cc490e6c4..c178171d3 100644 --- a/tests/unit/sources/test_live_catchup_planning.py +++ b/tests/unit/sources/test_live_catchup_planning.py @@ -319,6 +319,7 @@ def test_catch_up_ingests_needed_files_in_bounded_chunks( monkeypatch.setattr(live_watcher, "_CATCH_UP_MAX_BATCH_BYTES", 100) calls: list[tuple[list[Path], int | None, int]] = [] + whole_archive_flags: list[bool] = [] retry_scan_calls: list[int] = [] async def fake_ingest_files( @@ -326,8 +327,10 @@ async def fake_ingest_files( *, queued_file_count: int | None = None, skipped_file_count: int = 0, + whole_archive_convergence: bool = True, ) -> None: calls.append((paths, queued_file_count, skipped_file_count)) + whole_archive_flags.append(whole_archive_convergence) watcher._ingest_files = fake_ingest_files # type: ignore[assignment,method-assign] watcher._schedule_failed_retry_scan = lambda: retry_scan_calls.append(len(calls)) # type: ignore[method-assign] @@ -338,9 +341,42 @@ async def fake_ingest_files( assert calls[0][1:] == (5, 0) assert calls[1][1:] == (2, 0) assert calls[2][1:] == (1, 0) + # Whole-archive convergence stages run once, on the last chunk. + assert whole_archive_flags == [False, False, True] assert retry_scan_calls == [3] +def test_catch_up_ingests_a_cold_backlog_without_a_recent_source( + tmp_path: Path, + frozen_clock: FrozenClock, +) -> None: + """Anti-vacuity: breaking out of the priority loop on an empty hot group skips the whole backlog.""" + root = tmp_path / "src" + root.mkdir() + historical = [root / f"historical-{index}.jsonl" for index in range(2)] + now = 1_800_000_000.0 + stale = now - live_watcher._CATCH_UP_HOT_FILE_AGE_S - 1 + for path in historical: + path.write_text('{"role":"user","content":"old"}\n') + os.utime(path, (stale, stale)) + watcher = LiveWatcher( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=None)), + (WatchSource(name="test", root=root),), + ) + frozen_clock.set_time(now) + + calls: list[list[Path]] = [] + + async def fake_ingest_files(paths: list[Path], **_kwargs: object) -> None: + calls.append(paths) + + watcher._ingest_files = fake_ingest_files # type: ignore[assignment,method-assign] + + asyncio.run(watcher._catch_up([root])) + + assert calls == [historical] + + def test_catch_up_ingests_recent_source_before_historical_backlog( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/sources/test_live_watcher.py b/tests/unit/sources/test_live_watcher.py index 1eb9f58bf..656ea0c57 100644 --- a/tests/unit/sources/test_live_watcher.py +++ b/tests/unit/sources/test_live_watcher.py @@ -1326,11 +1326,13 @@ async def stop_after_ingest( *, queued_file_count: int | None = None, skipped_file_count: int = 0, + whole_archive_convergence: bool = True, ) -> LiveBatchMetrics: metrics = await original_ingest( paths, queued_file_count=queued_file_count, skipped_file_count=skipped_file_count, + whole_archive_convergence=whole_archive_convergence, ) watcher.stop() return metrics diff --git a/tests/unit/storage/test_fts_derivation.py b/tests/unit/storage/test_fts_derivation.py index 36fdbbdd2..f5390bf43 100644 --- a/tests/unit/storage/test_fts_derivation.py +++ b/tests/unit/storage/test_fts_derivation.py @@ -190,3 +190,37 @@ def test_trigger_loss_is_incompatible_and_never_runtime_repaired(test_conn: sqli assert test_conn.execute("SELECT 1 FROM sqlite_master WHERE name = 'messages_fts_ad'").fetchone() is None restore_fts_triggers_sync(test_conn) + + +def test_partition_inspection_and_publish_search_the_block_id_index(test_conn: sqlite3.Connection) -> None: + """Anti-vacuity: a ``substr(block_id, ...)`` prefix predicate plans as a table scan. + + Every statement the partition inspection and publish issue against + ``messages_fts_identity`` must be served by the ``block_id`` index, so a + partition's cost is bounded by its own rows rather than the archive's. + The sibling key ``s1`` / ``s10`` proves the range is exact at the ``:`` + boundary. + """ + adapter = FtsDerivationAdapter() + short_key, _short_rowid = _seed_session(test_conn, "s1") + long_key, long_rowid = _seed_session(test_conn, "s10") + test_conn.execute("DELETE FROM messages_fts_identity WHERE rowid = ?", (long_rowid,)) + test_conn.commit() + + statements: list[str] = [] + test_conn.set_trace_callback(statements.append) + try: + short_inspection = adapter.inspect(test_conn, short_key) + assert adapter.publish(test_conn, adapter.input_for(test_conn, short_key)) + finally: + test_conn.set_trace_callback(None) + + assert short_inspection.status is FtsKeyStatus.VALID + assert adapter.inspect(test_conn, long_key).status is FtsKeyStatus.STALE + identity_statements = [ + sql for sql in statements if "messages_fts_identity" in sql and sql.lstrip().upper().startswith("SELECT") + ] + assert identity_statements, "inspection and publish must consult the identity ledger" + for sql in identity_statements: + plan = " | ".join(str(row[3]) for row in test_conn.execute(f"EXPLAIN QUERY PLAN {sql}").fetchall()) + assert "SCAN i" not in plan and "SCAN messages_fts_identity" not in plan, (sql, plan) diff --git a/tests/unit/storage/test_raw_authority_verdict_cache.py b/tests/unit/storage/test_raw_authority_verdict_cache.py index 73fd11708..596a9372d 100644 --- a/tests/unit/storage/test_raw_authority_verdict_cache.py +++ b/tests/unit/storage/test_raw_authority_verdict_cache.py @@ -323,3 +323,28 @@ def test_rekeyed_raw_replaces_its_stale_cache_row(tmp_path: Path) -> None: conn = archive._ensure_source_conn() rows = conn.execute("SELECT logical_source_key FROM raw_authority_verdicts WHERE raw_id = 'moved'").fetchall() assert [row[0] for row in rows] == ["codex:s1"] + + +def test_find_work_classifies_every_cohort_in_two_statements(tmp_path: Path) -> None: + """Anti-vacuity: a per-cohort probe issues two statements per cohort, red at three cohorts.""" + initialize_active_archive_root(tmp_path) + with ArchiveStore.open_existing(tmp_path, read_only=False) as archive: + _bind_full(archive, raw_id="fresh-1", payload=b"fresh\n", logical_source_key="codex:fresh") + _bind_full(archive, raw_id="stale-1", payload=b"stale\n", logical_source_key="codex:stale") + _bind_full(archive, raw_id="cold-1", payload=b"cold\n", logical_source_key="codex:cold") + get_or_compute_raw_authority_verdicts(archive, "codex:fresh", now_ms=1000) + get_or_compute_raw_authority_verdicts(archive, "codex:stale", now_ms=1000) + _bind_full(archive, raw_id="stale-2", payload=b"stale\nmore\n", logical_source_key="codex:stale") + + conn = archive._ensure_source_conn() + statements: list[str] = [] + conn.set_trace_callback(statements.append) + try: + work = find_raw_authority_verdict_cache_work(conn) + bounded = find_raw_authority_verdict_cache_work(conn, max_cohorts=1) + finally: + conn.set_trace_callback(None) + + assert work == RawAuthorityVerdictCacheWork(("codex:cold", "codex:stale")) + assert bounded == RawAuthorityVerdictCacheWork(("codex:cold",)) + assert len(statements) == 4