fix(storage): resumable, exclusive raw-replay rebuild + status surface

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

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",

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"]

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",
 ]