From 0bcd192377bcdb32add7bff0b6aaebd9e2a4686f Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 13:10:21 +0200 Subject: [PATCH 1/7] fix(storage): exclude superseded revisions from rebuild resume debt Problem: IndexGenerationStore.next_raw_page (the raw-replay rebuild's paged keyset cursor, used by both the offline `rebuild-index` CLI and the daemon's own bulk-rebuild loop) scheduled every row in `raw_sessions` unconditionally, including raws whose `raw_session_memberships.decision` is already durably `superseded_equivalent`/`superseded_prefix`. Those raws never gain their own `index.sessions` row (only their cohort's accepted head does), so every rebuild pass re-scheduled and re-parsed them for nothing -- a live archive carried 6,946 such rows. Solution: next_raw_page's SQL now excludes a raw only when EVERY persisted membership row for it is superseded, reusing the closed `MembershipDecision` vocabulary `classify_membership_revisions` already writes back to durable source.db (no rebuild-local classifier). A raw with no membership row (never censused) or with at least one non-superseded row remains eligible, so a genuinely accepted-but- unindexed or still-pending raw is never dropped. The filter lives inside the same query as the keyset cursor, so pagination correctness is unaffected. Verification: devtools test tests/unit/storage/test_index_generation.py -k NextRawPage -> 4 passed. Confirmed red-first: reverting the index_generation.py change while keeping the new tests fails 2 of 4 (test_fully_superseded_raw_is_excluded_from_the_page, test_exclusion_survives_the_keyset_cursor_across_pages). Ref polylogue-b5l.1 Co-Authored-By: Claude --- polylogue/storage/index_generation.py | 49 +++++++- tests/unit/storage/test_index_generation.py | 129 ++++++++++++++++++++ 2 files changed, 173 insertions(+), 5 deletions(-) diff --git a/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index 56031f5f2f..1744efea53 100644 --- a/polylogue/storage/index_generation.py +++ b/polylogue/storage/index_generation.py @@ -17,6 +17,7 @@ from pathlib import Path from types import TracebackType +from polylogue.archive.session_revision_membership import MembershipDecision from polylogue.storage.archive_identity import ArchiveLocation from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier @@ -25,6 +26,18 @@ _LOCK_PID_PATTERN = re.compile(r"pid=(\d+)") +#: ``raw_session_memberships.decision`` values that mean "this raw is +#: resolved, durable history" rather than resume debt -- reused directly from +#: ``classify_membership_revisions``'s own closed vocabulary +#: (``polylogue.archive.session_revision_membership.MembershipDecision``) +#: instead of a rebuild-local classifier (polylogue-b5l.1 design note: reuse +#: the existing revision authority vocabulary, never fork a parallel one). +_SUPERSEDED_DECISIONS: tuple[str, ...] = ( + MembershipDecision.SUPERSEDED_EQUIVALENT.value, + MembershipDecision.SUPERSEDED_PREFIX.value, +) +_SUPERSEDED_DECISION_PLACEHOLDERS = ",".join("?" for _ in _SUPERSEDED_DECISIONS) + #: Superseded generations kept after a promotion. One is enough to roll back #: to the previous index; each costs roughly the size of the index itself #: (~35 GB on the reference archive), so keeping more is expensive storage, @@ -495,28 +508,54 @@ def next_raw_page( (``ArchiveStore.expand_raw_membership_selection``) already walks the full ``raw_sessions``/``raw_session_memberships`` graph regardless of which page triggered it -- neither depends on processing order. + + polylogue-b5l.1: a raw whose EVERY persisted + ``raw_session_memberships`` row already carries a durable + ``superseded_equivalent``/``superseded_prefix`` decision (the + classification ``classify_membership_revisions`` itself writes back, + durable in ``source.db`` independent of any index generation) is + legitimate resolved history, not resume debt: it will never gain an + ``index.sessions`` row of its own (only its cohort's accepted head + does), so scheduling it wastes a full page slot re-parsing content a + prior pass already resolved -- every single rebuild pass would + otherwise re-touch it. A raw with no membership row at all (never + censused) or with at least one non-superseded row (``applied``/ + ``ambiguous``/``deferred``/still pending) is left eligible -- this + only ever narrows the schedule, it never risks dropping a genuinely + unresolved or newly-accepted raw. """ if limit <= 0: raise ValueError("rebuild raw page limit must be positive") source_db = self.archive_root / "source.db" + not_resume_debt_clause = f"""( + NOT EXISTS (SELECT 1 FROM raw_session_memberships m WHERE m.raw_id = raw_sessions.raw_id) + OR EXISTS ( + SELECT 1 FROM raw_session_memberships m + WHERE m.raw_id = raw_sessions.raw_id + AND (m.decision IS NULL OR m.decision NOT IN ({_SUPERSEDED_DECISION_PLACEHOLDERS})) + ) + )""" if transaction.last_blob_hash_hex is None or transaction.last_raw_id is None: - query = """ + query = f""" SELECT raw_id, blob_hash, blob_size FROM raw_sessions + WHERE {not_resume_debt_clause} ORDER BY blob_hash, raw_id LIMIT ? """ - params: tuple[object, ...] = (limit + 1,) + params: tuple[object, ...] = (*_SUPERSEDED_DECISIONS, limit + 1) else: last_blob_hash = bytes.fromhex(transaction.last_blob_hash_hex) - query = """ + query = f""" SELECT raw_id, blob_hash, blob_size FROM raw_sessions - WHERE blob_hash > ? - OR (blob_hash = ? AND raw_id > ?) + WHERE (blob_hash > ? + OR (blob_hash = ? AND raw_id > ?)) + AND {not_resume_debt_clause} ORDER BY blob_hash, raw_id LIMIT ? """ params = ( last_blob_hash, last_blob_hash, transaction.last_raw_id, + *_SUPERSEDED_DECISIONS, limit + 1, ) with closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True)) as conn: diff --git a/tests/unit/storage/test_index_generation.py b/tests/unit/storage/test_index_generation.py index 51a9a440fe..4ef24a7555 100644 --- a/tests/unit/storage/test_index_generation.py +++ b/tests/unit/storage/test_index_generation.py @@ -505,3 +505,132 @@ def test_pruning_never_removes_a_never_promoted_rebuild_candidate(tmp_path: Path # The genuine rollback target -- the previously-active generation -- is what # the retained slot is for, not the inactive candidate. assert Path(first.index_path).exists() + + +def _seed_membership( + source_db: Path, + *, + raw_id: str, + logical_source_key: str, + decision: str | None, +) -> None: + with sqlite3.connect(source_db) as conn: + conn.execute("PRAGMA foreign_keys = ON") + conn.execute( + """ + INSERT INTO raw_session_memberships ( + raw_id, logical_source_key, provider_session_id, source_revision, + normalized_content_hash, message_count, revision_authority, decision, decided_at_ms + ) VALUES (?, ?, ?, ?, zeroblob(32), 1, 'byte_proven', ?, ?) + """, + (raw_id, logical_source_key, raw_id, raw_id, decision, 1 if decision is not None else None), + ) + + +def _seed_raw(conn: sqlite3.Connection, *, raw_id: str, blob_hash: bytes, acquired_at_ms: int) -> None: + conn.execute( + """INSERT INTO raw_sessions (raw_id, origin, native_id, source_path, source_index, blob_hash, + blob_size, acquired_at_ms, validation_status) + VALUES (?, 'codex-session', ?, ?, 0, ?, 1, ?, 'passed')""", + (raw_id, raw_id, f"/{raw_id}.jsonl", blob_hash, acquired_at_ms), + ) + + +class TestNextRawPageExcludesSupersededResumeDebt: + """polylogue-b5l.1 AC3: a raw whose every persisted membership decision is + ``superseded_equivalent``/``superseded_prefix`` is resolved history, not + resume debt -- it never gains its own ``index.sessions`` row (only its + cohort's accepted head does), so scheduling it every pass wastes a full + page slot re-parsing content a prior pass already resolved. A genuinely + accepted-but-unindexed raw, or one never censused at all, must still be + selected. + """ + + def test_fully_superseded_raw_is_excluded_from_the_page(self, tmp_path: Path) -> None: + _archive(tmp_path) + with sqlite3.connect(tmp_path / "source.db") as conn: + _seed_raw(conn, raw_id="raw-superseded", blob_hash=b"\x01" * 32, acquired_at_ms=10) + _seed_raw(conn, raw_id="raw-accepted", blob_hash=b"\x02" * 32, acquired_at_ms=20) + _seed_membership( + tmp_path / "source.db", + raw_id="raw-superseded", + logical_source_key="cohort-1", + decision="superseded_prefix", + ) + _seed_membership( + tmp_path / "source.db", + raw_id="raw-accepted", + logical_source_key="cohort-2", + decision="applied", + ) + + store = IndexGenerationStore.for_archive_root(tmp_path) + transaction = store.create_transaction(source_snapshot="source-v1") + page = store.next_raw_page(transaction, limit=10) + + raw_ids = [row[0] for row in page.rows] + assert raw_ids == ["raw-accepted"] + + def test_never_censused_raw_remains_eligible(self, tmp_path: Path) -> None: + """A raw with no membership row at all (never classified) must still + be scheduled -- excluding it would silently drop genuinely novel, + never-processed content.""" + _archive(tmp_path) + with sqlite3.connect(tmp_path / "source.db") as conn: + _seed_raw(conn, raw_id="raw-novel", blob_hash=b"\x03" * 32, acquired_at_ms=10) + + store = IndexGenerationStore.for_archive_root(tmp_path) + transaction = store.create_transaction(source_snapshot="source-v1") + page = store.next_raw_page(transaction, limit=10) + + assert [row[0] for row in page.rows] == ["raw-novel"] + + def test_raw_superseded_in_one_cohort_but_pending_in_another_remains_eligible(self, tmp_path: Path) -> None: + """A multi-membership raw (e.g. a bundle member) is only resume-debt + -free when EVERY known membership row is superseded; a mixed shape + (superseded in one cohort, still ambiguous/pending in another) must + remain eligible.""" + _archive(tmp_path) + with sqlite3.connect(tmp_path / "source.db") as conn: + _seed_raw(conn, raw_id="raw-mixed", blob_hash=b"\x04" * 32, acquired_at_ms=10) + _seed_membership( + tmp_path / "source.db", raw_id="raw-mixed", logical_source_key="cohort-a", decision="superseded_prefix" + ) + _seed_membership(tmp_path / "source.db", raw_id="raw-mixed", logical_source_key="cohort-b", decision=None) + + store = IndexGenerationStore.for_archive_root(tmp_path) + transaction = store.create_transaction(source_snapshot="source-v1") + page = store.next_raw_page(transaction, limit=10) + + assert [row[0] for row in page.rows] == ["raw-mixed"] + + def test_exclusion_survives_the_keyset_cursor_across_pages(self, tmp_path: Path) -> None: + """The superseded-exclusion filter is applied inside the same SQL + query as the keyset cursor, so a superseded raw sitting between two + eligible pages must never surface on a later page either.""" + _archive(tmp_path) + with sqlite3.connect(tmp_path / "source.db") as conn: + _seed_raw(conn, raw_id="raw-a", blob_hash=b"\x01" * 32, acquired_at_ms=10) + _seed_raw(conn, raw_id="raw-superseded", blob_hash=b"\x02" * 32, acquired_at_ms=20) + _seed_raw(conn, raw_id="raw-b", blob_hash=b"\x03" * 32, acquired_at_ms=30) + _seed_membership( + tmp_path / "source.db", + raw_id="raw-superseded", + logical_source_key="cohort-1", + decision="superseded_equivalent", + ) + + store = IndexGenerationStore.for_archive_root(tmp_path) + transaction = store.create_transaction(source_snapshot="source-v1") + first_page = store.next_raw_page(transaction, limit=1) + assert [row[0] for row in first_page.rows] == ["raw-a"] + + transaction = store.checkpoint_transaction( + transaction, + status="paused", + last_blob_hash_hex=first_page.rows[0][1], + last_raw_id=first_page.rows[0][0], + processed_raw_count=1, + ) + second_page = store.next_raw_page(transaction, limit=1) + assert [row[0] for row in second_page.rows] == ["raw-b"] From 7fcdb440d99aa523057c0618dd2558f73ec4cf1a Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 13:12:55 +0200 Subject: [PATCH 2/7] test(maintenance): prove RebuildLease holds through the raw-replay pass Problem: polylogue-b5l.1 AC1 requires the raw-replay rebuild (rebuild_index_from_source, the ops reset --index && polylogued run path) to hold the archive-root writer lease from its first write through final parity/activation -- not merely check it once at entry, which is exactly the narrow point-in-time-check shape the 2026-07-10 competing-daemon incident exhibited. PR #2872 proved this property for the clone-forward fast-forward path only; the raw-replay path had no equivalent regression test even though `with RebuildLease(root):` already wraps the whole pass body. Solution: a new regression test monkeypatches polylogue.storage.repair.repair_session_insights -- a terminal stage that runs after replay has committed rows and before FTS parity/readiness/promotion -- to attempt a concurrent ActiveWriterLease.acquire() from inside the pass, and asserts it fails. This proves the lease is held deep inside the pass, not just at the top, and that it is released again once the pass returns. Verification: devtools test tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py -> 1 passed. Ref polylogue-b5l.1 Co-Authored-By: Claude --- .../test_rebuild_index_lease_lifecycle.py | 111 ++++++++++++++++++ 1 file changed, 111 insertions(+) create mode 100644 tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py diff --git a/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py b/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py new file mode 100644 index 0000000000..cd4f4fee62 --- /dev/null +++ b/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py @@ -0,0 +1,111 @@ +"""polylogue-b5l.1 AC1: the raw-replay rebuild must hold ``RebuildLease`` for +its ENTIRE lifecycle, not merely a point-in-time check at entry. + +The 2026-07-10 competing-daemon incident was exactly a narrow point-in-time +check (``_require_service_stopped``'s systemctl probe) that missed a +transient-unit window between the check and the write. PR #2872 proved +``RebuildLease``/``ActiveWriterLease`` mutual exclusion for the clone-forward +fast-forward path (``devtools/archive_schema_fast_forward.py``); this test +proves the SAME property for the raw-replay path +(``rebuild_index_from_source`` / ``ops reset --index && polylogued run``), +at a point deep inside the pass -- after replay has already committed rows to +the owned inactive generation, immediately before terminal FTS-parity / +readiness / promotion -- not just at the top of the function. + +Anti-vacuity: the mutation that makes this test fail is narrowing +``with RebuildLease(root):`` in ``_rebuild_index_from_source_owned`` to wrap +only the initial checks (e.g. moving replay/terminal-stage work outside the +``with`` block) -- exactly the "checked once, not held" shape the 2026-07-10 +incident exhibited. With the lease held for the whole pass, a concurrent +``ActiveWriterLease.acquire()`` attempted from inside +``repair_session_insights`` (a terminal stage that runs AFTER replay and +BEFORE promotion) must fail; if the lease were released early, that same +attempt would silently succeed. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from polylogue.core.enums import Provider +from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync +from polylogue.storage.index_generation import ActiveWriterLease, RebuildLeaseUnavailableError +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root + + +def _codex_session(native_id: str) -> bytes: + rows: list[dict[str, object]] = [ + {"type": "session_meta", "payload": {"id": native_id, "timestamp": "2026-07-16T10:00:00Z"}}, + { + "type": "response_item", + "payload": { + "type": "message", + "id": f"{native_id}-m0", + "role": "user", + "content": [{"type": "input_text", "text": f"hello {native_id}"}], + }, + }, + ] + return b"".join(json.dumps(row, sort_keys=True).encode() + b"\n" for row in rows) + + +def _seed_one_codex_session(root: Path) -> None: + initialize_active_archive_root(root) + with ArchiveStore.open_existing(root, read_only=False) as archive: + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session("sess-lease-lifecycle"), + source_path="lease-lifecycle-test/0.jsonl", + acquired_at_ms=1, + ) + + +def test_rebuild_lease_blocks_a_concurrent_writer_deep_inside_the_pass( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "archive" + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) + _seed_one_codex_session(root) + + probe_result: dict[str, object] = {"attempted": False, "blocked": False} + + import polylogue.storage.repair as repair_module + + real_repair_session_insights = repair_module.repair_session_insights + + def probing_repair_session_insights(*args: object, **kwargs: object) -> object: + # This terminal stage runs strictly AFTER replay has already + # committed rows into the owned inactive generation, and strictly + # BEFORE FTS parity / readiness / promotion -- exactly the window + # the 2026-07-10 incident's narrow point-in-time check missed. + probe_result["attempted"] = True + writer = ActiveWriterLease(root) + try: + writer.acquire() + except RebuildLeaseUnavailableError: + probe_result["blocked"] = True + else: + writer.close() + return real_repair_session_insights(*args, **kwargs) + + monkeypatch.setattr(repair_module, "repair_session_insights", probing_repair_session_insights) + + receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + + assert receipt.status == "replayed" + assert probe_result["attempted"] is True, "the probe never ran; the test setup itself is broken" + assert probe_result["blocked"] is True, ( + "a concurrent ActiveWriterLease acquisition succeeded mid-pass -- RebuildLease was not held " + "for the pass's entire lifecycle" + ) + + # After the pass returns, the lease must be released -- a later, + # legitimate writer must not be blocked forever by a lease this rebuild + # forgot to release. + writer = ActiveWriterLease(root) + writer.acquire() + writer.close() From 66cc163d581fdf08bacf61261451f3753f725c94 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 13:16:43 +0200 Subject: [PATCH 3/7] feat(storage): add a read-only rebuild-lease status probe Problem: polylogue-b5l.1 AC5 requires the raw-replay rebuild to expose lease ownership for status/recovery surfaces, and to make stale-lock recovery explicit. There was no way to inspect the lease (held/holder/liveness) without either blocking or, worse, risking a reclaim as a side effect of merely looking. Solution: RebuildLeaseStatus + rebuild_lease_status(archive_root) in polylogue/storage/index_generation.py. It attempts a non-blocking flock: success means nothing holds the lease (released immediately after the probe); failure means it is genuinely held, and the lock file's recorded pid/host are reported alongside a liveness check against that pid, with `stale=True` when the recorded holder is provably dead (the same condition RebuildLease.__enter__ already reclaims on next acquisition). Never disturbs a real holder and never blocks. Verification: devtools test tests/unit/storage/test_index_generation.py -k RebuildLeaseStatus -> 6 passed. Ref polylogue-b5l.1 Co-Authored-By: Claude --- polylogue/storage/index_generation.py | 81 +++++++++++++++++++++ tests/unit/storage/test_index_generation.py | 79 ++++++++++++++++++++ 2 files changed, 160 insertions(+) diff --git a/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index 1744efea53..208bb71193 100644 --- a/polylogue/storage/index_generation.py +++ b/polylogue/storage/index_generation.py @@ -25,6 +25,7 @@ logger = logging.getLogger(__name__) _LOCK_PID_PATTERN = re.compile(r"pid=(\d+)") +_LOCK_HOST_PATTERN = re.compile(r"host=(\S+)") #: ``raw_session_memberships.decision`` values that mean "this raw is #: resolved, durable history" rather than resume debt -- reused directly from @@ -172,6 +173,16 @@ def _lock_holder_pid(path: Path) -> int | None: return int(match.group(1)) +def _lock_holder_host(path: Path) -> str | None: + """Best-effort recorded hostname from an existing lock file; ``None`` if absent/unreadable.""" + try: + text = path.read_text(encoding="utf-8") + except OSError: + return None + match = _LOCK_HOST_PATTERN.search(text) + return match.group(1) if match is not None else None + + def _pid_is_alive(pid: int) -> bool: """Whether ``pid`` still names a live process, best-effort via ``kill(pid, 0)``.""" if pid <= 0: @@ -286,6 +297,74 @@ def close(self) -> None: self._fd = None +@dataclass(frozen=True, slots=True) +class RebuildLeaseStatus: + """Read-only snapshot of the archive-root rebuild lease, for status surfaces. + + polylogue-b5l.1 AC5: an operator/agent must be able to see who owns the + lease, whether the recorded holder is actually still alive, and whether + the lock looks reclaimable, without disturbing a real holder and without + duplicating ``RebuildLease``/``ActiveWriterLease`` as the sole exclusion + mechanism. + """ + + held: bool + holder_pid: int | None + holder_host: str | None + #: ``None`` when ``held`` is False (nothing to check liveness against) or + #: when no pid could be parsed from the lock file at all. + holder_alive: bool | None + #: True when the lease is held but its recorded pid is provably dead -- + #: exactly the ``RebuildLease.__enter__`` reclaim condition + #: (``_open_lock_fd``), surfaced here for operator visibility before a + #: fresh acquisition would silently reclaim it. + stale: bool + + def to_dict(self) -> dict[str, object]: + return { + "held": self.held, + "holder_pid": self.holder_pid, + "holder_host": self.holder_host, + "holder_alive": self.holder_alive, + "stale": self.stale, + } + + +def rebuild_lease_status(archive_root: Path) -> RebuildLeaseStatus: + """Probe the rebuild lease without blocking or disturbing a genuine holder. + + Attempts a non-blocking exclusive ``flock``: if it succeeds, nothing + currently holds the lease and the probe releases it immediately; if it + fails with ``EAGAIN``/``EACCES`` (``BlockingIOError``), the lease is + genuinely held and the lock file's recorded pid/host are reported + best-effort for diagnosis (the file content may be stale or unreadable). + """ + path = archive_root / ".index-rebuild.lock" + if not path.exists(): + return RebuildLeaseStatus(held=False, holder_pid=None, holder_host=None, holder_alive=None, stale=False) + holder_pid = _lock_holder_pid(path) + holder_host = _lock_holder_host(path) + fd = os.open(path, os.O_RDWR) + try: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + alive = _pid_is_alive(holder_pid) if holder_pid is not None else None + return RebuildLeaseStatus( + held=True, + holder_pid=holder_pid, + holder_host=holder_host, + holder_alive=alive, + stale=holder_pid is not None and alive is False, + ) + fcntl.flock(fd, fcntl.LOCK_UN) + return RebuildLeaseStatus( + held=False, holder_pid=holder_pid, holder_host=holder_host, holder_alive=None, stale=False + ) + finally: + os.close(fd) + + class IndexGenerationStore: """Create, checkpoint, and atomically promote inactive generations. @@ -805,8 +884,10 @@ def _fsync_directory(path: Path) -> None: "IndexGeneration", "IndexRebuildTransaction", "IndexGenerationStore", + "RebuildLeaseStatus", "RebuildRawPage", "RebuildLease", "RebuildLeaseUnavailableError", + "rebuild_lease_status", "source_revision_snapshot", ] diff --git a/tests/unit/storage/test_index_generation.py b/tests/unit/storage/test_index_generation.py index 4ef24a7555..2eeed65d33 100644 --- a/tests/unit/storage/test_index_generation.py +++ b/tests/unit/storage/test_index_generation.py @@ -17,6 +17,7 @@ IndexGenerationStore, RebuildLease, RebuildLeaseUnavailableError, + rebuild_lease_status, source_revision_snapshot, ) @@ -634,3 +635,81 @@ def test_exclusion_survives_the_keyset_cursor_across_pages(self, tmp_path: Path) ) second_page = store.next_raw_page(transaction, limit=1) assert [row[0] for row in second_page.rows] == ["raw-b"] + + +class TestRebuildLeaseStatus: + """polylogue-b5l.1 AC5: a read-only lease probe for status surfaces -- + must never block, never disturb a genuine holder, and must distinguish + "not held" / "held by a live process" / "held but recorded pid is dead + (reclaimable)".""" + + def test_reports_not_held_when_no_lock_file_exists(self, tmp_path: Path) -> None: + status = rebuild_lease_status(tmp_path) + assert status.held is False + assert status.holder_pid is None + assert status.stale is False + + def test_reports_not_held_after_a_lease_is_released(self, tmp_path: Path) -> None: + with RebuildLease(tmp_path): + pass + status = rebuild_lease_status(tmp_path) + assert status.held is False + # The lock file's recorded pid/host from the released lease is still + # readable (best-effort diagnosis), but "held" reflects reality now. + assert status.holder_pid == os.getpid() + + def test_reports_held_by_this_process_while_a_lease_is_open(self, tmp_path: Path) -> None: + with RebuildLease(tmp_path): + status = rebuild_lease_status(tmp_path) + assert status.held is True + assert status.holder_pid == os.getpid() + assert status.holder_alive is True + assert status.stale is False + + def test_reports_held_by_a_separate_live_process(self, tmp_path: Path) -> None: + ready = multiprocessing.Event() + release = multiprocessing.Event() + process = multiprocessing.Process(target=_hold_lease, args=(str(tmp_path), ready, release)) + process.start() + assert ready.wait(5) + try: + status = rebuild_lease_status(tmp_path) + assert status.held is True + assert status.holder_pid == process.pid + assert status.holder_alive is True + assert status.stale is False + finally: + release.set() + process.join(5) + assert process.exitcode == 0 + + def test_reports_stale_when_recorded_holder_pid_is_dead(self, tmp_path: Path) -> None: + lock_path = tmp_path / ".index-rebuild.lock" + lock_path.parent.mkdir(parents=True, exist_ok=True) + holder_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) + fcntl.flock(holder_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + os.write(holder_fd, f"pid={_DEFINITELY_DEAD_PID} host=nowhere\n".encode()) + os.fsync(holder_fd) + try: + status = rebuild_lease_status(tmp_path) + assert status.held is True + assert status.holder_pid == _DEFINITELY_DEAD_PID + assert status.holder_host == "nowhere" + assert status.holder_alive is False + assert status.stale is True + finally: + fcntl.flock(holder_fd, fcntl.LOCK_UN) + os.close(holder_fd) + + def test_probe_never_blocks_or_disturbs_a_genuine_holder(self, tmp_path: Path) -> None: + """Calling the probe repeatedly while a real lease is held must never + raise, never remove the lock file, and never itself release the + real holder's lock.""" + with RebuildLease(tmp_path): + for _ in range(3): + status = rebuild_lease_status(tmp_path) + assert status.held is True + # The real holder must still hold it after repeated probing. + with pytest.raises(RebuildLeaseUnavailableError): + with RebuildLease(tmp_path): + pass From be4e67428c702b472b8bde058333b08068da9e86 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 13:21:51 +0200 Subject: [PATCH 4/7] feat(maintenance): add consolidated rebuild_status for AC5 status surface Problem: polylogue-b5l.1 AC5 requires the raw-replay rebuild to report owner/build/archive/schema/generation/heartbeat/cursor/delta/recovery from one place, with explicit stale-lock recovery guidance. Previously an operator had to hand-cross-reference .index-rebuild.lock, .index-active-pointer, and a transaction JSON file under .index-rebuild-transactions/ separately, with no delta/recovery synthesis at all. Solution: rebuild_status(archive_root, operation_id=None, include_daemon_bulk_rebuild=True) in polylogue/maintenance/rebuild_index.py assembles: the read-only lease probe (RebuildLeaseStatus from the prior commit), the active generation's metadata, the active index's schema (PRAGMA user_version), the resumable transaction (defaulting to the daemon's well-known DAEMON_BULK_REBUILD_OPERATION_ID when no operation_id is given -- the ops reset --index && polylogued run case never has an explicit one to pass), a source-snapshot delta comparison, and a recovery message list covering a stale (dead-pid) lease, a failed transaction, and a drifted source snapshot. Entirely read-only: never acquires RebuildLease, never mutates a transaction or generation. Verification: devtools test tests/unit/maintenance/test_rebuild_status.py -> 6 passed. Ref polylogue-b5l.1 Co-Authored-By: Claude --- polylogue/maintenance/rebuild_index.py | 111 ++++++++++ tests/unit/maintenance/test_rebuild_status.py | 192 ++++++++++++++++++ 2 files changed, 303 insertions(+) create mode 100644 tests/unit/maintenance/test_rebuild_status.py diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py index 8c54e568eb..22b6400b45 100644 --- a/polylogue/maintenance/rebuild_index.py +++ b/polylogue/maintenance/rebuild_index.py @@ -1079,6 +1079,116 @@ def rebuild_index_from_source_sync(request: RebuildIndexRequest) -> RebuildIndex return asyncio.run(rebuild_index_from_source(request)) +def rebuild_status( + archive_root: Path, + *, + operation_id: str | None = None, + include_daemon_bulk_rebuild: bool = True, +) -> dict[str, object]: + """Consolidated raw-replay rebuild status for operator/agent surfaces. + + polylogue-b5l.1 AC5: one read gives lease ownership, the active + generation, the resumable transaction's cursor/delta, and explicit + stale-lock/failed-transaction recovery guidance -- instead of an operator + hand-cross-referencing ``.index-rebuild.lock``, ``.index-active-pointer``, + and a transaction JSON file under ``.index-rebuild-transactions/``. + + ``operation_id`` selects which persisted transaction to report. When + omitted and ``include_daemon_bulk_rebuild`` is True (the default), this + falls back to the daemon's own well-known bulk-rebuild operation id + (``DAEMON_BULK_REBUILD_OPERATION_ID``) -- the common case for + ``ops reset --index && polylogued run``, where the daemon never has an + explicit operation id to hand the caller. Read-only throughout: never + acquires ``RebuildLease``, never mutates any transaction or generation. + """ + from polylogue.daemon.bulk_rebuild import DAEMON_BULK_REBUILD_OPERATION_ID + from polylogue.storage.index_generation import ( + IndexGenerationStore, + rebuild_lease_status, + source_revision_snapshot, + ) + + location = ArchiveLocation.resolve(archive_root) + lease = rebuild_lease_status(archive_root) + store = IndexGenerationStore(location) + + active_generation: dict[str, object] | None = None + try: + active_target = store.active_pointer.resolve(strict=True) + except OSError: + active_target = None + if active_target is not None: + for metadata_path in store.generations_root.glob("gen-*/generation.json"): + try: + generation = store.load(metadata_path.parent.name) + if generation.state == "active" and Path(generation.index_path).resolve(strict=True) == active_target: + active_generation = cast(dict[str, object], asdict(generation)) + break + except (OSError, ValueError, TypeError): + continue + + schema_version: int | None = None + try: + with contextlib.closing(sqlite3.connect(f"file:{store.active_pointer}?mode=ro", uri=True, timeout=5.0)) as conn: + row = conn.execute("PRAGMA user_version").fetchone() + schema_version = int(row[0]) if row is not None else None + except sqlite3.Error: + schema_version = None + + resolved_operation_id = operation_id + if resolved_operation_id is None and include_daemon_bulk_rebuild: + resolved_operation_id = DAEMON_BULK_REBUILD_OPERATION_ID + + transaction_payload: dict[str, object] | None = None + delta: dict[str, object] | None = None + if resolved_operation_id is not None: + try: + transaction = store.load_transaction(resolved_operation_id) + except FileNotFoundError: + transaction = None + except (OSError, ValueError, TypeError, KeyError): + transaction = None + if transaction is not None: + transaction_payload = cast(dict[str, object], asdict(transaction)) + current_snapshot = source_revision_snapshot(archive_root) if (archive_root / "source.db").exists() else None + delta = { + "source_snapshot_matches": ( + current_snapshot is not None and current_snapshot == transaction.source_snapshot + ), + "current_source_snapshot": current_snapshot, + "transaction_source_snapshot": transaction.source_snapshot, + } + + recovery: list[str] = [] + if lease.stale: + recovery.append( + f"lease lock file records dead pid={lease.holder_pid} host={lease.holder_host!r}; " + "the next RebuildLease acquisition reclaims it automatically -- no manual action required " + "unless a fresh attempt still refuses" + ) + if transaction_payload is not None and transaction_payload.get("status") == "failed": + recovery.append( + f"transaction {resolved_operation_id!r} is failed: {transaction_payload.get('error')!r}; " + "resume with the same --operation-id to retry the same candidate, or discard it to start fresh" + ) + if delta is not None and delta.get("source_snapshot_matches") is False: + recovery.append( + f"transaction {resolved_operation_id!r} source snapshot no longer matches current source.db; " + "the next pass against this operation id will refuse as stale -- start a new operation" + ) + + return { + "archive_root": str(archive_root), + "lease": lease.to_dict(), + "generation": active_generation, + "schema_version": schema_version, + "operation_id": resolved_operation_id, + "transaction": transaction_payload, + "delta": delta, + "recovery": recovery, + } + + __all__ = [ "RebuildIndexReceipt", "RebuildIndexRequest", @@ -1088,6 +1198,7 @@ def rebuild_index_from_source_sync(request: RebuildIndexRequest) -> RebuildIndex "missing_index_raw_ids", "rebuild_index_from_source", "rebuild_index_from_source_sync", + "rebuild_status", "select_rebuild_raw_ids", "validate_rebuild_index_request", ] diff --git a/tests/unit/maintenance/test_rebuild_status.py b/tests/unit/maintenance/test_rebuild_status.py new file mode 100644 index 0000000000..b64d229e3d --- /dev/null +++ b/tests/unit/maintenance/test_rebuild_status.py @@ -0,0 +1,192 @@ +"""``rebuild_status`` (polylogue-b5l.1 AC5): one consolidated read for lease +ownership, the active generation, the resumable transaction's cursor/delta, +and explicit stale-lock/failed-transaction recovery guidance. + +Anti-vacuity: the mutation that makes +``test_reports_stale_lease_recovery_guidance`` fail is removing the +``lease.stale`` branch's recovery message (or ``rebuild_lease_status``'s own +dead-pid detection this depends on); the mutation that makes +``test_reports_source_snapshot_delta_when_source_has_drifted`` fail is +dropping the ``source_revision_snapshot`` comparison and always reporting +``source_snapshot_matches=True``. +""" + +from __future__ import annotations + +import fcntl +import json +import os +import sqlite3 +from pathlib import Path + +import pytest + +from polylogue.core.enums import Provider +from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync, rebuild_status +from polylogue.storage.index_generation import IndexGenerationStore, source_revision_snapshot +from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root, initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier + +_DEFINITELY_DEAD_PID = 2**31 - 1 + + +def _init_empty_source(root: Path) -> None: + root.mkdir(parents=True, exist_ok=True) + initialize_archive_database(root / "source.db", ArchiveTier.SOURCE) + + +def _codex_session(native_id: str) -> bytes: + rows: list[dict[str, object]] = [ + {"type": "session_meta", "payload": {"id": native_id, "timestamp": "2026-07-16T10:00:00Z"}}, + { + "type": "response_item", + "payload": { + "type": "message", + "id": f"{native_id}-m0", + "role": "user", + "content": [{"type": "input_text", "text": f"hello {native_id}"}], + }, + }, + ] + return b"".join(json.dumps(row, sort_keys=True).encode() + b"\n" for row in rows) + + +def _seed_one_real_codex_session(root: Path) -> None: + initialize_active_archive_root(root) + with ArchiveStore.open_existing(root, read_only=False) as archive: + archive.write_raw_payload( + provider=Provider.CODEX, + payload=_codex_session("sess-status-probe"), + source_path="status-probe-test/0.jsonl", + acquired_at_ms=1, + ) + + +def test_reports_no_lease_and_no_transaction_on_a_fresh_archive(tmp_path: Path) -> None: + root = tmp_path / "archive" + _init_empty_source(root) + + status = rebuild_status(root, operation_id="does-not-exist", include_daemon_bulk_rebuild=False) + + assert status["archive_root"] == str(root) + assert status["lease"] == { + "held": False, + "holder_pid": None, + "holder_host": None, + "holder_alive": None, + "stale": False, + } + assert status["generation"] is None + assert status["transaction"] is None + assert status["delta"] is None + assert status["recovery"] == [] + + +def test_reports_stale_lease_recovery_guidance(tmp_path: Path) -> None: + root = tmp_path / "archive" + _init_empty_source(root) + lock_path = root / ".index-rebuild.lock" + holder_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600) + fcntl.flock(holder_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + os.write(holder_fd, f"pid={_DEFINITELY_DEAD_PID} host=nowhere\n".encode()) + os.fsync(holder_fd) + try: + status = rebuild_status(root, operation_id="none", include_daemon_bulk_rebuild=False) + assert status["lease"]["held"] is True + assert status["lease"]["stale"] is True + recovery = status["recovery"] + assert isinstance(recovery, list) + assert any("dead pid" in message for message in recovery) + finally: + fcntl.flock(holder_fd, fcntl.LOCK_UN) + os.close(holder_fd) + + +def test_reports_active_generation_and_schema_version_after_a_rebuild( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "archive" + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) + _seed_one_real_codex_session(root) + receipt = rebuild_index_from_source_sync(RebuildIndexRequest(archive_root=root)) + assert receipt.status == "replayed" + + status = rebuild_status(root, operation_id="none", include_daemon_bulk_rebuild=False) + + generation = status["generation"] + assert isinstance(generation, dict) + assert generation["state"] == "active" + assert status["schema_version"] is not None + + +def test_reports_transaction_cursor_and_no_delta_when_source_unchanged(tmp_path: Path) -> None: + root = tmp_path / "archive" + _init_empty_source(root) + with sqlite3.connect(root / "source.db") as conn: + conn.execute( + """INSERT INTO raw_sessions (raw_id, origin, native_id, source_path, source_index, blob_hash, + blob_size, acquired_at_ms, validation_status) + VALUES ('raw-a', 'codex-session', 'raw-a', '/raw-a', 0, randomblob(32), 1, 1, 'passed')""" + ) + store = IndexGenerationStore.for_archive_root(root) + transaction = store.create_transaction( + source_snapshot=source_revision_snapshot(root), operation_id="status-probe-op" + ) + transaction = store.checkpoint_transaction( + transaction, status="paused", last_raw_id="raw-a", last_blob_hash_hex="00" * 32, processed_raw_count=1 + ) + + status = rebuild_status(root, operation_id="status-probe-op") + + txn_payload = status["transaction"] + assert isinstance(txn_payload, dict) + assert txn_payload["operation_id"] == "status-probe-op" + assert txn_payload["processed_raw_count"] == 1 + delta = status["delta"] + assert isinstance(delta, dict) + assert delta["source_snapshot_matches"] is True + assert status["recovery"] == [] + + +def test_reports_source_snapshot_delta_when_source_has_drifted(tmp_path: Path) -> None: + root = tmp_path / "archive" + _init_empty_source(root) + with sqlite3.connect(root / "source.db") as conn: + conn.execute( + """INSERT INTO raw_sessions (raw_id, origin, native_id, source_path, source_index, blob_hash, + blob_size, acquired_at_ms, validation_status) + VALUES ('raw-a', 'codex-session', 'raw-a', '/raw-a', 0, randomblob(32), 1, 1, 'passed')""" + ) + store = IndexGenerationStore.for_archive_root(root) + transaction = store.create_transaction(source_snapshot="stale-snapshot", operation_id="drift-op") + assert transaction.status == "running" + + status = rebuild_status(root, operation_id="drift-op") + + delta = status["delta"] + assert isinstance(delta, dict) + assert delta["source_snapshot_matches"] is False + recovery = status["recovery"] + assert isinstance(recovery, list) + assert any("source snapshot no longer matches" in message for message in recovery) + + +def test_falls_back_to_the_daemon_well_known_operation_id_by_default(tmp_path: Path) -> None: + """Omitting ``operation_id`` must resolve the daemon's own well-known + bulk-rebuild transaction -- the common case for + ``ops reset --index && polylogued run``, which never has an operation id + to hand this status surface explicitly.""" + from polylogue.daemon.bulk_rebuild import ( + DAEMON_BULK_REBUILD_OPERATION_ID, + resolve_or_start_daemon_bulk_rebuild_transaction, + ) + + root = tmp_path / "archive" + _init_empty_source(root) + resolve_or_start_daemon_bulk_rebuild_transaction(root) + + status = rebuild_status(root) + + assert status["operation_id"] == DAEMON_BULK_REBUILD_OPERATION_ID + assert status["transaction"] is not None From e8ed99c78dd51fdb18c7d2301283f1c7b7940103 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 13:29:52 +0200 Subject: [PATCH 5/7] feat(cli): add maintenance rebuild-index-status command Problem: polylogue-b5l.1 AC5 requires an operator/agent-facing surface for the consolidated rebuild_status payload (lease, generation, schema, cursor/delta, recovery), not just a library function nothing calls. Solution: `polylogue ops maintenance rebuild-index-status` renders polylogue.maintenance.rebuild_index.rebuild_status in plain or JSON form. Read-only; --operation-id selects which transaction to report, defaulting to the daemon's well-known bulk-rebuild operation id. Regenerated docs/plans/topology-target.yaml (devtools render topology-projection) for the new module. Verification: manual smoke test via `python3 -c "... from polylogue.cli import main; main()"` with POLYLOGUE_ARCHIVE_ROOT set to a scratch dir, both plain and --output-format json, confirms correct output. devtools render all --check -> exit 0 (grepped for "out of sync": none). devtools test tests/unit/cli/ -k maintenance -> 159 passed, 1 failed (test_verify_archive_cli_json_reports_every_registered_check, in tests/unit/cli/test_maintenance_verify_archive_cli.py -- untouched by this diff, unrelated to rebuild-index status; last touched by PR #3529 which predates and is unrelated to this change). Ref polylogue-b5l.1 Co-Authored-By: Claude --- docs/plans/topology-target.yaml | 10 +- .../cli/commands/maintenance/__init__.py | 6 ++ .../maintenance/_rebuild_index_status.py | 101 ++++++++++++++++++ 3 files changed, 114 insertions(+), 3 deletions(-) create mode 100644 polylogue/cli/commands/maintenance/_rebuild_index_status.py diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index 6fecaeb00a..a5ef163e9d 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -931,7 +931,7 @@ files: target: polylogue/cli/commands/judge.py owner: stable - path: polylogue/cli/commands/maintenance/__init__.py - loc: 191 + loc: 197 target: polylogue/cli/commands/maintenance/__init__.py owner: stable - path: polylogue/cli/commands/maintenance/_archive_plan.py @@ -986,6 +986,10 @@ files: loc: 550 target: polylogue/cli/commands/maintenance/_rebuild_index.py owner: stable + - path: polylogue/cli/commands/maintenance/_rebuild_index_status.py + loc: 101 + target: polylogue/cli/commands/maintenance/_rebuild_index_status.py + owner: stable - path: polylogue/cli/commands/maintenance/_run.py loc: 218 target: polylogue/cli/commands/maintenance/_run.py @@ -2256,7 +2260,7 @@ files: target: polylogue/maintenance/raw_membership_writeback_apply.py owner: stable - path: polylogue/maintenance/rebuild_index.py - loc: 1093 + loc: 1204 target: polylogue/maintenance/rebuild_index.py owner: stable - path: polylogue/maintenance/registry.py @@ -3861,7 +3865,7 @@ files: owner: storage-root reason: storage-root cross-cutting helper - path: polylogue/storage/index_generation.py - loc: 773 + loc: 893 target: TBD owner: storage-domain - path: polylogue/storage/insights/__init__.py diff --git a/polylogue/cli/commands/maintenance/__init__.py b/polylogue/cli/commands/maintenance/__init__.py index 023ef80099..1b3878a553 100644 --- a/polylogue/cli/commands/maintenance/__init__.py +++ b/polylogue/cli/commands/maintenance/__init__.py @@ -59,6 +59,12 @@ "rebuild_index_command", "Inspect or execute an authority-safe source-to-index rebuild.", ), + ( + "rebuild-index-status", + "_rebuild_index_status", + "rebuild_index_status_command", + "Report consolidated raw-replay rebuild status (lease/generation/cursor/delta/recovery). Read-only.", + ), ( "raw-authority-frontier", "_raw_identity", diff --git a/polylogue/cli/commands/maintenance/_rebuild_index_status.py b/polylogue/cli/commands/maintenance/_rebuild_index_status.py new file mode 100644 index 0000000000..50e883aad6 --- /dev/null +++ b/polylogue/cli/commands/maintenance/_rebuild_index_status.py @@ -0,0 +1,101 @@ +"""``maintenance rebuild-index-status``: consolidated raw-replay rebuild status. + +polylogue-b5l.1 AC5: one command reports lease ownership, the active +generation, the resumable transaction's cursor/delta, and explicit +stale-lock/failed-transaction recovery guidance -- see +``polylogue.maintenance.rebuild_index.rebuild_status`` for the assembled +payload this command renders. Entirely read-only. +""" + +from __future__ import annotations + +import json + +import click + +from polylogue.logging import configure_logging +from polylogue.paths import archive_root + + +@click.command("rebuild-index-status") +@click.option( + "--operation-id", + "operation_id", + type=str, + default=None, + help=( + "Rebuild transaction to report. Omit to resolve the daemon's own well-known " + "bulk-rebuild operation id (the ops reset --index && polylogued run case)." + ), +) +@click.option( + "--no-daemon-fallback", + "no_daemon_fallback", + is_flag=True, + help="Do not fall back to the daemon's well-known bulk-rebuild operation id when --operation-id is omitted.", +) +@click.option( + "--output-format", + "output_format", + type=click.Choice(["plain", "json"]), + default="plain", + show_default=True, + help="Output format.", +) +def rebuild_index_status_command( + operation_id: str | None, + no_daemon_fallback: bool, + output_format: str, +) -> None: + """Report consolidated raw-replay rebuild status. Read-only.""" + from polylogue.maintenance.rebuild_index import rebuild_status + + configure_logging() + root = archive_root() + status = rebuild_status(root, operation_id=operation_id, include_daemon_bulk_rebuild=not no_daemon_fallback) + + if output_format == "json": + click.echo(json.dumps(status, indent=2, sort_keys=True)) + return + + click.echo(f"Archive root: {status['archive_root']}") + lease = status["lease"] + assert isinstance(lease, dict) + click.echo( + f"Lease: held={lease['held']} holder_pid={lease['holder_pid']} " + f"holder_host={lease['holder_host']} holder_alive={lease['holder_alive']} stale={lease['stale']}" + ) + generation = status["generation"] + if isinstance(generation, dict): + click.echo( + f"Generation: id={generation['generation_id']} state={generation['state']} " + f"created_at_ms={generation['created_at_ms']}" + ) + else: + click.echo("Generation: none") + click.echo(f"Schema: user_version={status['schema_version']}") + click.echo(f"Operation id: {status['operation_id']}") + transaction = status["transaction"] + if isinstance(transaction, dict): + click.echo( + f"Transaction: status={transaction['status']} " + f"processed_raw_count={transaction['processed_raw_count']:,} " + f"processed_blob_bytes={transaction['processed_blob_bytes']:,} " + f"last_raw_id={transaction['last_raw_id']} updated_at_ms={transaction['updated_at_ms']}" + ) + else: + click.echo("Transaction: none") + delta = status["delta"] + if isinstance(delta, dict): + click.echo(f"Delta: source_snapshot_matches={delta['source_snapshot_matches']}") + recovery = status["recovery"] + assert isinstance(recovery, list) + if recovery: + click.echo("Recovery:") + for message in recovery: + click.echo(f" - {message}") + else: + click.echo("Recovery: none") + + +__all__ = ["rebuild_index_status_command"] From ce5353b09d2d7331db122f0f235c6a81ea3f83ff Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 13:35:57 +0200 Subject: [PATCH 6/7] fix(test): satisfy mypy --strict on the new rebuild-status/lease tests Problem: devtools verify --quick's mypy step failed on the two new test files added for polylogue-b5l.1 -- indexing an object-typed dict value without a narrowing isinstance check, and a monkeypatched repair_session_insights replacement typed with a bare *args/**kwargs signature that mypy correctly rejected against the real function's concrete keyword-only signature. Solution: isinstance-narrow status["lease"] before indexing it (same pattern already used elsewhere in this file); give probing_repair_session_insights the exact same signature as repair_session_insights (config, dry_run, then the five keyword-only params) instead of *args/**kwargs. Verification: python3 -m mypy tests/unit/maintenance/test_rebuild_status.py tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py polylogue/maintenance/rebuild_index.py polylogue/storage/index_generation.py polylogue/cli/commands/maintenance/_rebuild_index_status.py -> Success: no issues found in 5 source files. devtools test tests/unit/maintenance/ test_rebuild_status.py tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py tests/unit/storage/test_index_generation.py -> 41 passed. Ref polylogue-b5l.1 Co-Authored-By: Claude --- .../test_rebuild_index_lease_lifecycle.py | 27 +++++++++++++++++-- tests/unit/maintenance/test_rebuild_status.py | 6 +++-- 2 files changed, 29 insertions(+), 4 deletions(-) diff --git a/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py b/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py index cd4f4fee62..75c8e50fb3 100644 --- a/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py +++ b/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py @@ -27,12 +27,18 @@ import json from pathlib import Path +from typing import TYPE_CHECKING import pytest from polylogue.core.enums import Provider from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync from polylogue.storage.index_generation import ActiveWriterLease, RebuildLeaseUnavailableError + +if TYPE_CHECKING: + from polylogue.config import Config + from polylogue.core.protocols import ProgressCallback + from polylogue.storage.repair import RepairResult from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root @@ -77,7 +83,16 @@ def test_rebuild_lease_blocks_a_concurrent_writer_deep_inside_the_pass( real_repair_session_insights = repair_module.repair_session_insights - def probing_repair_session_insights(*args: object, **kwargs: object) -> object: + def probing_repair_session_insights( + config: Config, + dry_run: bool = False, + *, + progress_callback: ProgressCallback | None = None, + progress_total: int | None = None, + session_ids: tuple[str, ...] | None = None, + archive_root_override: Path | None = None, + owned_inactive_generation: tuple[str, str] | None = None, + ) -> RepairResult: # This terminal stage runs strictly AFTER replay has already # committed rows into the owned inactive generation, and strictly # BEFORE FTS parity / readiness / promotion -- exactly the window @@ -90,7 +105,15 @@ def probing_repair_session_insights(*args: object, **kwargs: object) -> object: probe_result["blocked"] = True else: writer.close() - return real_repair_session_insights(*args, **kwargs) + return real_repair_session_insights( + config, + dry_run, + progress_callback=progress_callback, + progress_total=progress_total, + session_ids=session_ids, + archive_root_override=archive_root_override, + owned_inactive_generation=owned_inactive_generation, + ) monkeypatch.setattr(repair_module, "repair_session_insights", probing_repair_session_insights) diff --git a/tests/unit/maintenance/test_rebuild_status.py b/tests/unit/maintenance/test_rebuild_status.py index b64d229e3d..b6ed33a97e 100644 --- a/tests/unit/maintenance/test_rebuild_status.py +++ b/tests/unit/maintenance/test_rebuild_status.py @@ -93,8 +93,10 @@ def test_reports_stale_lease_recovery_guidance(tmp_path: Path) -> None: os.fsync(holder_fd) try: status = rebuild_status(root, operation_id="none", include_daemon_bulk_rebuild=False) - assert status["lease"]["held"] is True - assert status["lease"]["stale"] is True + lease = status["lease"] + assert isinstance(lease, dict) + assert lease["held"] is True + assert lease["stale"] is True recovery = status["recovery"] assert isinstance(recovery, list) assert any("dead pid" in message for message in recovery) From 1f46135505ee2435087354e28ec3067f739b4f54 Mon Sep 17 00:00:00 2001 From: Sinity Date: Mon, 3 Aug 2026 13:45:57 +0200 Subject: [PATCH 7/7] docs(maintenance): document the new rebuild-index-status CLI command Problem: devtools verify docs-coverage failed after adding `ops maintenance rebuild-index-status` -- every public CLI command must be reachable (named verbatim) from README.md or docs/**/*.md, and the new command had zero doc footprint. Solution: document it alongside the existing `rebuild-index` inventory entry in docs/design/convergence-simplification-inventory.md, since that is where operational tooling for this command family is already inventoried. Verification: devtools verify docs-coverage -> "every public CLI command, MCP tool, config key, and stable route is reachable". Ref polylogue-b5l.1 Co-Authored-By: Claude --- docs/design/convergence-simplification-inventory.md | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/docs/design/convergence-simplification-inventory.md b/docs/design/convergence-simplification-inventory.md index 321528da1d..bda64d647e 100644 --- a/docs/design/convergence-simplification-inventory.md +++ b/docs/design/convergence-simplification-inventory.md @@ -437,6 +437,19 @@ to run it by hand when the trickle conveyor's backlog is bulk-scale, and the only viable path once the daemon's own conveyor made a live backlog net-negative. +**Companion status surface (polylogue-b5l.1):** +`polylogue ops maintenance rebuild-index-status` +(`polylogue/cli/commands/maintenance/_rebuild_index_status.py`, +handler `rebuild_index_status_command`) reports the consolidated, +read-only view an operator needs while `rebuild-index` (or the daemon's own +bulk-rebuild loop) is running or paused: archive-root lease ownership +(held/holder pid/host/liveness/staleness), the active generation, the active +index's schema version, and the resumable transaction's cursor +(`processed_raw_count`/`last_raw_id`/`updated_at_ms`) alongside a +source-snapshot delta and explicit stale-lock/failed-transaction recovery +guidance (`polylogue.maintenance.rebuild_index.rebuild_status`). It never +acquires the rebuild lease itself. + **Why it exists today:** it is the one code path that already does the right thing for a bulk backlog — one resumable transaction, blue-green generation, full parse envelope, one census+replay sweep — because it does