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 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"] 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/polylogue/storage/index_generation.py b/polylogue/storage/index_generation.py index 56031f5f2f..208bb71193 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 @@ -24,6 +25,19 @@ 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 +#: ``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 @@ -159,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: @@ -273,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. @@ -495,28 +587,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: @@ -766,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/maintenance/test_rebuild_index_lease_lifecycle.py b/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py new file mode 100644 index 0000000000..75c8e50fb3 --- /dev/null +++ b/tests/unit/maintenance/test_rebuild_index_lease_lifecycle.py @@ -0,0 +1,134 @@ +"""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 +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 + + +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( + 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 + # 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( + 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) + + 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() diff --git a/tests/unit/maintenance/test_rebuild_status.py b/tests/unit/maintenance/test_rebuild_status.py new file mode 100644 index 0000000000..b6ed33a97e --- /dev/null +++ b/tests/unit/maintenance/test_rebuild_status.py @@ -0,0 +1,194 @@ +"""``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) + 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) + 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 diff --git a/tests/unit/storage/test_index_generation.py b/tests/unit/storage/test_index_generation.py index 51a9a440fe..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, ) @@ -505,3 +506,210 @@ 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"] + + +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