From b1ac686263416ab3152ce0629c6959a839a3c1be Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 18:43:41 +0200 Subject: [PATCH 1/4] perf(fts): search the block_id index for a session's partition rows Partition inspection and publish selected a session's identity-ledger rows with substr(block_id, 1, length(key) + 1) = key || ':', which plans as a full scan of messages_fts_identity for every key on every pass. block_id is session_id || ':' || ..., so the half-open range [key || ':', key || ';') selects exactly the same rows through the block_id UNIQUE index (';' is the code point after ':'). Measured on a 359-session archive: 10.5 ms -> 0.1 ms per key. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- polylogue/storage/fts/derivation.py | 22 +++++++++++---- tests/unit/storage/test_fts_derivation.py | 34 +++++++++++++++++++++++ 2 files changed, 50 insertions(+), 6 deletions(-) diff --git a/polylogue/storage/fts/derivation.py b/polylogue/storage/fts/derivation.py index e91a8b53ba..ad3f8d6c89 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/tests/unit/storage/test_fts_derivation.py b/tests/unit/storage/test_fts_derivation.py index 36fdbbdd2e..f5390bf433 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) From 6812abc8d6b01bfedea196fd826e4c455fc952d6 Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 18:43:58 +0200 Subject: [PATCH 2/4] perf(raw-authority): find stale verdict cohorts in two statements find_raw_authority_verdict_cache_work probed every logical_source_key with two point queries, so each daemon convergence check cost O(cohorts) statements. It now reads the cohort rows and the cached fingerprints in one pass each and applies the same staleness rule per key. write_raw_authority_verdict_cache deleted only the cohort's own rows; a raw already cached under a previous logical_source_key collided with the raw_id primary key (54 warmup failures in one rehearsal, each re-running the whole-archive check on the next chunk). Stale rows for the cohort's raw_ids are removed before the insert. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- .../storage/raw_authority_verdict_cache.py | 30 ++++++++++++++----- .../test_raw_authority_verdict_cache.py | 25 ++++++++++++++++ 2 files changed, 47 insertions(+), 8 deletions(-) diff --git a/polylogue/storage/raw_authority_verdict_cache.py b/polylogue/storage/raw_authority_verdict_cache.py index 1a5cce6730..73fce0a721 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/storage/test_raw_authority_verdict_cache.py b/tests/unit/storage/test_raw_authority_verdict_cache.py index 73fd117083..596a9372df 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 From 5b5b5679f738c79881acfe83cedb12a54673554c Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 18:44:00 +0200 Subject: [PATCH 3/4] fix(watcher): ingest a cold backlog when no source file is recent The catch-up loop broke out of its priority groups on the first empty group, so a backlog with no file modified in the last hour was never planned: a fresh daemon over a quiet source tree logged "watching" and ingested nothing. An empty group is skipped; only a stop request ends the loop. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- polylogue/sources/live/watcher.py | 5 +-- .../sources/test_live_catchup_planning.py | 31 +++++++++++++++++++ 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/polylogue/sources/live/watcher.py b/polylogue/sources/live/watcher.py index 6140135ded..4a2b57341d 100644 --- a/polylogue/sources/live/watcher.py +++ b/polylogue/sources/live/watcher.py @@ -590,9 +590,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: diff --git a/tests/unit/sources/test_live_catchup_planning.py b/tests/unit/sources/test_live_catchup_planning.py index cc490e6c43..36986747e6 100644 --- a/tests/unit/sources/test_live_catchup_planning.py +++ b/tests/unit/sources/test_live_catchup_planning.py @@ -341,6 +341,37 @@ async def fake_ingest_files( 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, From 1a0ac2368fc7e6b8f0eb18b0cedb5e78e411ae7d Mon Sep 17 00:00:00 2001 From: Sinity Date: Sat, 5 Sep 2026 18:44:26 +0200 Subject: [PATCH 4/4] perf(daemon): bound a catch-up chunk's convergence to the chunk's own input Per-chunk convergence paid archive-wide costs on every chunk of a catch-up: the exact FTS readiness audit (whole-table aggregates, run twice per chunk: after FTS repair and after insights), the raw-authority verdict warmer (one probe per cohort), the Claude workflow and delegation graphs (rebuilt from every raw artifact), a scan of every hook sidecar journal (1.65 GB on the rehearsal spool: 4.5 s warm, 73 s once the page cache was evicted), and the periodic whole-archive FTS sweep interleaving with chunks. Stage check time was outside the ledger, so a 102 s chunk showed 2.5 s of stages. - ConvergenceStage.whole_archive marks stages whose work is a function of the archive; converge_batch(whole_archive=False) records them SKIPPED. The watcher runs every catch-up chunk but the last in that scope, so the archive-wide stages run once per catch-up. - The readiness audit is its own whole-archive stage (fts_readiness), run after fts and derived; partition-scoped FTS passes and insight rebuilds no longer publish it. The periodic whole-archive FTS sweep waits while a chunked catch-up is active. - Hook paste enrichment reads only the batch's sessions' sidecar journals (-.jsonl). - The convergence ledger charges .check and hook_paste_enrichment. Scratch run, 400 files / 325 MB, 100 chunks, same host: convergence 200.4 s -> see PR body for the after figures. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DNGJ3awfNrsLaMdHgQZvid --- polylogue/daemon/cli.py | 22 +- polylogue/daemon/convergence.py | 30 ++- polylogue/daemon/convergence_stages.py | 76 +++++- polylogue/daemon/fts_convergence.py | 6 +- polylogue/sources/live/batch.py | 38 ++- .../sources/live/hook_paste_enrichment.py | 36 ++- polylogue/sources/live/watcher.py | 30 ++- tests/unit/daemon/test_catch_up_chunk_cost.py | 223 ++++++++++++++++++ tests/unit/daemon/test_convergence_stages.py | 1 + tests/unit/daemon/test_daemon_convergence.py | 64 ++++- .../sources/test_hook_paste_enrichment.py | 66 ++++++ .../sources/test_live_batch_convergence.py | 3 +- .../sources/test_live_catchup_planning.py | 5 + tests/unit/sources/test_live_watcher.py | 2 + 14 files changed, 570 insertions(+), 32 deletions(-) create mode 100644 tests/unit/daemon/test_catch_up_chunk_cost.py diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index 88a5d69eaa..3224fcd965 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 f8f0c7ef13..edc72686cd 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 9adea49b72..fc8c37621d 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 c109905675..a07c632895 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 67161a6ee8..a5d1397e26 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 a7aa3d5c67..1be414a01d 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 4a2b57341d..87d1e2457e 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()] @@ -653,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(): @@ -676,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 @@ -754,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. @@ -1702,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) @@ -1713,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/tests/unit/daemon/test_catch_up_chunk_cost.py b/tests/unit/daemon/test_catch_up_chunk_cost.py new file mode 100644 index 0000000000..a460103046 --- /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 5253fc619e..978504349c 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 62c252d8be..a4b89e7cbb 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 24553781b2..73d3ac611a 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 50437eda93..4e891dea8c 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 36986747e6..c178171d31 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,6 +341,8 @@ 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] diff --git a/tests/unit/sources/test_live_watcher.py b/tests/unit/sources/test_live_watcher.py index 1eb9f58bf8..656ea0c578 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