diff --git a/docs/openapi/search.yaml b/docs/openapi/search.yaml index 691c8c2f62..140445277d 100644 --- a/docs/openapi/search.yaml +++ b/docs/openapi/search.yaml @@ -3233,6 +3233,13 @@ x-polylogue-route-contracts: auth_policy: credential_if_configured response_contract: SearchEnvelope / SessionListResponse with route_state notes: Local UDS-only root-request parameter envelope; daemon owns query compilation. +- method: POST + pattern: /api/maintenance/rebuild-index + kind: maintenance + stability: operational + auth_policy: bearer_if_configured_and_same_origin + response_contract: RebuildIndexReceipt + notes: Runs exactly one source snapshot replay through the daemon write coordinator. - method: GET pattern: /api/facets kind: read_query diff --git a/docs/plans/topology-target.yaml b/docs/plans/topology-target.yaml index b511c951f1..4879238463 100644 --- a/docs/plans/topology-target.yaml +++ b/docs/plans/topology-target.yaml @@ -952,7 +952,7 @@ files: target: polylogue/cli/commands/maintenance/_raw_identity.py owner: stable - path: polylogue/cli/commands/maintenance/_rebuild_index.py - loc: 663 + loc: 511 target: polylogue/cli/commands/maintenance/_rebuild_index.py owner: stable - path: polylogue/cli/commands/maintenance/_run.py @@ -1560,7 +1560,7 @@ files: target: polylogue/daemon/healthz.py owner: stable - path: polylogue/daemon/http.py - loc: 4924 + loc: 5017 target: polylogue/daemon/http.py owner: stable - path: polylogue/daemon/lifecycle.py @@ -1629,7 +1629,7 @@ files: target: polylogue/daemon/provenance.py owner: stable - path: polylogue/daemon/route_contracts.py - loc: 625 + loc: 634 target: polylogue/daemon/route_contracts.py owner: stable - path: polylogue/daemon/similarity.py @@ -1721,7 +1721,7 @@ files: target: polylogue/daemon/workspace_routes.py owner: stable - path: polylogue/daemon/write_coordinator.py - loc: 439 + loc: 451 target: polylogue/daemon/write_coordinator.py owner: stable - path: polylogue/declarations/__init__.py @@ -2050,6 +2050,10 @@ files: loc: 505 target: polylogue/maintenance/preview.py owner: stable + - path: polylogue/maintenance/rebuild_index.py + loc: 378 + target: polylogue/maintenance/rebuild_index.py + owner: stable - path: polylogue/maintenance/registry.py loc: 277 target: polylogue/maintenance/registry.py diff --git a/docs/topology-status.md b/docs/topology-status.md index 005b74656a..446e2e1e63 100644 --- a/docs/topology-status.md +++ b/docs/topology-status.md @@ -28,12 +28,12 @@ Generated by `devtools render topology-status`. Reads `docs/plans/topology-targe ### Summary -- **Stable** (no move scoped): 873 +- **Stable** (no move scoped): 874 - **Kernel** (polylogue/ root): 8 - **Primitives** (storage-root): 19 - **TBD** (cell needs explicit assignment): 9 -- **Total declared**: 1044 -- **Realized polylogue/**/*.py**: 1044 files declared +- **Total declared**: 1045 +- **Realized polylogue/**/*.py**: 1045 files declared ### TBD cells (require explicit routing) diff --git a/polylogue/cli/commands/maintenance/_rebuild_index.py b/polylogue/cli/commands/maintenance/_rebuild_index.py index c860401925..89c8a38792 100644 --- a/polylogue/cli/commands/maintenance/_rebuild_index.py +++ b/polylogue/cli/commands/maintenance/_rebuild_index.py @@ -2,87 +2,92 @@ from __future__ import annotations -import asyncio import contextlib import json import sqlite3 -import time -from dataclasses import asdict from pathlib import Path from typing import Any, cast +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen import click -from polylogue.config import Config from polylogue.logging import configure_logging -from polylogue.paths import archive_root, render_root +from polylogue.paths import archive_root from polylogue.storage.archive_identity import ArchiveLocation +def _run_daemon_rebuild( + daemon_url: str, + *, + only_missing: bool, + raw_ids: tuple[str, ...], + max_blob_mb: float | None, + no_promote: bool, + operation_id: str | None, + raw_batch_size: int, + pass_byte_budget_mb: float | None, + pass_deadline_seconds: float | None, +) -> dict[str, object]: + """Execute one rebuild pass through the daemon-owned writer.""" + from polylogue.config import load_polylogue_config + + body = json.dumps( + { + "only_missing": only_missing, + "raw_ids": list(raw_ids), + "max_blob_mb": max_blob_mb, + "promote": not no_promote, + "operation_id": operation_id, + "raw_batch_size": raw_batch_size, + "pass_byte_budget_mb": pass_byte_budget_mb, + "pass_deadline_seconds": pass_deadline_seconds, + } + ).encode("utf-8") + headers = {"Content-Type": "application/json"} + if auth_token := load_polylogue_config().api_auth_token: + headers["Authorization"] = f"Bearer {auth_token}" + request = Request( + f"{daemon_url.rstrip('/')}/api/maintenance/rebuild-index", + data=body, + headers=headers, + method="POST", + ) + try: + with urlopen(request, timeout=600) as response: + payload = json.loads(response.read()) + except HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise click.ClickException(f"daemon rebuild rejected by {daemon_url}: HTTP {exc.code}: {detail}") from exc + except (URLError, OSError, ValueError) as exc: + raise click.ClickException(f"could not reach daemon at {daemon_url}: {exc}") from exc + if not isinstance(payload, dict): + raise click.ClickException(f"daemon at {daemon_url} returned an invalid rebuild receipt") + return cast(dict[str, object], payload) + + def _count_source_raw_sessions(root: Path) -> int: - source_db = root / "source.db" - if not source_db.exists(): - return 0 - with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: - row = conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() - return int(row[0]) if row is not None else 0 + from polylogue.maintenance.rebuild_index import count_source_raw_sessions + + return count_source_raw_sessions(root) def _missing_index_raw_ids(root: Path) -> list[str]: - source_db = root / "source.db" - index_db = ArchiveLocation.resolve(root).active_index_path - if not source_db.exists() or not index_db.exists(): - return [] - with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: - conn.execute("ATTACH DATABASE ? AS idx", (str(index_db),)) - rows = conn.execute( - """ - SELECT r.raw_id - FROM raw_sessions r - WHERE NOT EXISTS ( - SELECT 1 - FROM idx.sessions s - WHERE s.raw_id = r.raw_id - ) - ORDER BY r.acquired_at_ms, r.raw_id - """ - ).fetchall() - return [str(row[0]) for row in rows] + from polylogue.maintenance.rebuild_index import missing_index_raw_ids + + return missing_index_raw_ids(root) def _all_index_rebuild_raw_ids(root: Path) -> list[str]: - source_db = root / "source.db" - if not source_db.exists(): - return [] - with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: - rows = conn.execute( - """ - SELECT raw_id - FROM raw_sessions - ORDER BY acquired_at_ms, raw_id - """ - ).fetchall() - return [str(row[0]) for row in rows] + from polylogue.maintenance.rebuild_index import all_index_rebuild_raw_ids + + return all_index_rebuild_raw_ids(root) def _filter_raw_ids_by_max_blob_size(root: Path, raw_ids: list[str], max_blob_mb: float | None) -> list[str]: - if max_blob_mb is None or not raw_ids: - return raw_ids - max_bytes = int(max_blob_mb * 1024 * 1024) - source_db = root / "source.db" - placeholders = ",".join("?" for _ in raw_ids) - with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: - rows = conn.execute( - f""" - SELECT raw_id - FROM raw_sessions - WHERE raw_id IN ({placeholders}) - AND blob_size <= ? - ORDER BY acquired_at_ms, raw_id - """, - (*raw_ids, max_bytes), - ).fetchall() - return [str(row[0]) for row in rows] + from polylogue.maintenance.rebuild_index import filter_raw_ids_by_max_blob_size + + return filter_raw_ids_by_max_blob_size(root, raw_ids, max_blob_mb) def _rebuild_index_selection_plan( @@ -332,6 +337,15 @@ def _rebuild_index_selection_plan( help="Output format.", ) @click.option("--no-promote", is_flag=True, help="Leave an exact-ready generation inactive after rebuilding it.") +@click.option( + "--daemon", "use_daemon", is_flag=True, help="Run the bounded rebuild through the live polylogued daemon." +) +@click.option( + "--daemon-url", + default=lambda: __import__("os").environ.get("POLYLOGUE_DAEMON_URL", "http://127.0.0.1:8766"), + show_default="POLYLOGUE_DAEMON_URL or http://127.0.0.1:8766", + help="Daemon HTTP base URL used with --daemon.", +) def rebuild_index_command( only_missing: bool, raw_ids: tuple[str, ...], @@ -344,14 +358,14 @@ def rebuild_index_command( pass_deadline_seconds: float | None, output_format: str, no_promote: bool, + use_daemon: bool, + daemon_url: str, ) -> None: """Inspect or execute an authority-safe source-to-index rebuild. Execution expands the requested rows to complete logical revision cohorts; selection order and batch boundaries never participate in authority. """ - from polylogue.maintenance.replay import rebuild_index_from_source - configure_logging() if raw_ids and only_missing: raise click.UsageError("--raw-id cannot be combined with --only-missing") @@ -363,6 +377,8 @@ def rebuild_index_command( raise click.UsageError("--max-blob-mb requires --only-missing or --raw-id") if plan_limit <= 0: raise click.BadParameter("plan limit must be positive", param_hint="--plan-limit") + if use_daemon and plan_only: + raise click.UsageError("--daemon executes a rebuild; --plan is always a local read-only preview") if raw_batch_size <= 0: raise click.BadParameter("raw batch size must be positive", param_hint="--raw-batch-size") if pass_byte_budget_mb is not None and pass_byte_budget_mb <= 0: @@ -371,9 +387,30 @@ def rebuild_index_command( raise click.BadParameter("pass deadline must be positive", param_hint="--pass-deadline-seconds") if operation_id is not None and (raw_ids or only_missing or max_blob_mb is not None or plan_only): raise click.UsageError("--operation-id only resumes an unfiltered full-source rebuild") + if operation_id is not None and (pass_byte_budget_mb is not None or pass_deadline_seconds is not None): + raise click.UsageError("resumed rebuild budgets are durable; omit pass budget options with --operation-id") root = archive_root() - location = ArchiveLocation.resolve(root) + if use_daemon: + payload = _run_daemon_rebuild( + daemon_url, + only_missing=only_missing, + raw_ids=raw_ids, + max_blob_mb=max_blob_mb, + no_promote=no_promote, + operation_id=operation_id, + raw_batch_size=raw_batch_size, + pass_byte_budget_mb=pass_byte_budget_mb, + pass_deadline_seconds=pass_deadline_seconds, + ) + if output_format == "json": + click.echo(json.dumps(payload, indent=2, sort_keys=True)) + return + click.echo(f"Archive root: {payload.get('archive_root', root)}") + click.echo(f"Classified: {int(cast(Any, payload['classified_full_count'])):,} full revision(s)") + click.echo(f"Replayed: {int(cast(Any, payload['replayed_logical_source_count'])):,} logical source(s)") + click.echo(f"Quarantined: {int(cast(Any, payload['quarantined_raw_count'])):,} raw row(s)") + return raw_count = _count_source_raw_sessions(root) if raw_count == 0: payload = { @@ -445,215 +482,26 @@ def rebuild_index_command( f"blob={int(group['blob_bytes']):,} source={group['source_path']}" ) return - from polylogue.cli.commands.status import _archive_readiness_status - from polylogue.maintenance.offline_guard import running_daemon_pid - from polylogue.storage.index_generation import ( - IndexGenerationStore, - RebuildLease, - source_revision_snapshot, - ) - - generation_store = IndexGenerationStore(root) - with RebuildLease(root): - active_config = Config( - archive_root=root, - render_root=render_root(), - sources=[], - db_path=location.active_index_path, - ) - daemon_pid = running_daemon_pid(active_config) - if daemon_pid is not None: - raise click.ClickException(f"offline rebuild refused while polylogued PID {daemon_pid} is running") - raw_count = _count_source_raw_sessions(root) - resumable_full_source = not raw_ids and not only_missing and max_blob_mb is None - transaction = None - page = None - pass_started_at_ms = int(time.time() * 1000) - if resumable_full_source: - transaction = ( - generation_store.load_transaction(operation_id) - if operation_id is not None - else generation_store.create_transaction( - source_snapshot=source_revision_snapshot(root), - pass_byte_budget=( - int(pass_byte_budget_mb * 1024 * 1024) if pass_byte_budget_mb is not None else None - ), - pass_deadline_ms=(int(pass_deadline_seconds * 1000) if pass_deadline_seconds is not None else None), - ) - ) - if operation_id is not None and (pass_byte_budget_mb is not None or pass_deadline_seconds is not None): - raise click.UsageError( - "resumed rebuild budgets are durable; omit pass budget options with --operation-id" - ) - if transaction.status in {"promoted", "stale"}: - raise click.ClickException( - f"rebuild operation {transaction.operation_id} is {transaction.status}; start a new operation" - ) - if source_revision_snapshot(root) != transaction.source_snapshot: - generation_store.checkpoint_transaction( - transaction, - status="stale", - error="source evidence changed since this rebuild was planned", - ) - raise click.ClickException( - f"rebuild operation {transaction.operation_id} is stale because source evidence changed" - ) - generation = generation_store.load(transaction.generation_id) - if generation.owner_id != transaction.generation_owner_id or generation.state != "inactive": - raise click.ClickException(f"rebuild operation {transaction.operation_id} lost its inactive candidate") - page = generation_store.next_raw_page(transaction, limit=raw_batch_size) - selected_raw_ids = [raw_id for raw_id, _acquired_at_ms, _blob_size in page.rows] - selected_raw_count = len(selected_raw_ids) - skipped_by_blob_limit_count = 0 - else: - selected_raw_ids = ( - list(dict.fromkeys(raw_ids)) - if raw_ids - else _missing_index_raw_ids(root) - if only_missing - else _all_index_rebuild_raw_ids(root) - ) - unfiltered_selected_raw_count = len(selected_raw_ids) - selected_raw_ids = _filter_raw_ids_by_max_blob_size(root, selected_raw_ids, max_blob_mb) - selected_raw_count = len(selected_raw_ids) - skipped_by_blob_limit_count = unfiltered_selected_raw_count - selected_raw_count - generation = generation_store.create(source_snapshot=source_revision_snapshot(root)) - source_drifted = False - try: - generation_root = Path(generation.index_path).parent - config = Config( - archive_root=generation_root, - render_root=render_root(), - sources=[], - db_path=Path(generation.index_path), - ) - result = asyncio.run( - rebuild_index_from_source( - config, - raw_ids=selected_raw_ids, - raw_batch_size=raw_batch_size, - ingest_workers=None, - materialize=True, - progress_callback=None, - owned_inactive_generation=(generation.generation_id, generation.owner_id), - ) - ) - if transaction is not None and selected_raw_ids: - if source_revision_snapshot(root) != transaction.source_snapshot: - transaction = generation_store.checkpoint_transaction( - transaction, - status="stale", - error="source evidence changed during this bounded rebuild pass", - ) - source_drifted = True - raise click.ClickException( - f"rebuild operation {transaction.operation_id} is stale because source evidence changed" - ) - assert page is not None - last_raw_id, last_acquired_at_ms, _blob_size = page.rows[-1] - elapsed_ms = int(time.time() * 1000) - pass_started_at_ms - deadline_expired = ( - transaction.pass_deadline_ms is not None and elapsed_ms >= transaction.pass_deadline_ms - ) - status = "deferred" if page.deferred_reason == "byte-budget" or deadline_expired else "paused" - if deadline_expired: - status = "deferred" - transaction = generation_store.checkpoint_transaction( - transaction, - status=status, - last_acquired_at_ms=last_acquired_at_ms, - last_raw_id=last_raw_id, - processed_raw_count=transaction.processed_raw_count + len(selected_raw_ids), - processed_blob_bytes=transaction.processed_blob_bytes + sum(row[2] for row in page.rows), - ) - if page.has_more or deadline_expired: - payload = { - "archive_root": str(root), - "raw_session_count": raw_count, - "selected_raw_count": selected_raw_count, - "skipped_by_blob_limit_count": 0, - "status": status, - "materialized": False, - "generation": asdict(generation), - "transaction": asdict(transaction), - **result, - } - if output_format == "json": - click.echo(json.dumps(payload, indent=2, sort_keys=True)) - else: - click.echo(f"Rebuild operation: {transaction.operation_id}") - click.echo(f"Scheduled: {selected_raw_count:,} raw row(s)") - click.echo(f"Paused cursor: {transaction.cursor or 'start'}") - click.echo("Resume with --operation-id after this bounded pass.") - return - from polylogue.storage.repair import repair_session_insights + from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync - insight_result = repair_session_insights( - config, - dry_run=False, - archive_root_override=generation_root, - owned_inactive_generation=(generation.generation_id, generation.owner_id), + try: + receipt = rebuild_index_from_source_sync( + RebuildIndexRequest( + archive_root=root, + only_missing=only_missing, + raw_ids=raw_ids, + max_blob_mb=max_blob_mb, + promote=not no_promote, + operation_id=operation_id, + raw_batch_size=raw_batch_size, + pass_byte_budget_mb=pass_byte_budget_mb, + pass_deadline_seconds=pass_deadline_seconds, ) - if not insight_result.success: - raise click.ClickException(f"session insight materialization failed: {insight_result.detail}") - if source_revision_snapshot(root) != generation.source_snapshot: - if transaction is not None: - transaction = generation_store.checkpoint_transaction( - transaction, - status="stale", - error="source evidence changed before terminal readiness", - ) - source_drifted = True - raise click.ClickException(f"source evidence changed while rebuilding {generation.generation_id}") - readiness = _archive_readiness_status(generation_root) - if not readiness.get("checked") or int(readiness.get("blocked_surface_count", 1)) != 0: - blocked = [ - name - for name, info in cast(dict[str, dict[str, object]], readiness.get("surfaces", {})).items() - if info.get("ready") is not True - ] - detail = ( - f"reason: {readiness.get('reason')}" - if not readiness.get("checked") - else "blocked surfaces: " + ", ".join(blocked) - ) - raise click.ClickException( - f"inactive generation {generation.generation_id} is not exact-ready; {detail}" - ) - if transaction is not None: - transaction = generation_store.checkpoint_transaction( - transaction, - status="ready", - ) - if not no_promote: - generation = generation_store.promote(generation) - if transaction is not None: - transaction = generation_store.checkpoint_transaction(transaction, status="promoted") - except Exception: - if transaction is not None and not source_drifted: - with contextlib.suppress(Exception): - generation_store.checkpoint_transaction( - transaction, - status="failed", - error="bounded rebuild pass failed; candidate retained for diagnosis or explicit recovery", - ) - else: - with contextlib.suppress(Exception): - generation_store.discard_if_inactive(generation) - raise - payload = { - "archive_root": str(root), - "raw_session_count": raw_count, - "selected_raw_count": selected_raw_count, - "skipped_by_blob_limit_count": skipped_by_blob_limit_count, - "status": "replayed", - "materialized": True, - "materialization": insight_result.to_dict(), - "generation": asdict(generation), - "readiness": readiness, - "transaction": asdict(transaction) if transaction is not None else None, - **result, - } + ) + except (RuntimeError, ValueError) as exc: + raise click.ClickException(str(exc)) from exc + payload = receipt.to_dict() + result = payload if output_format == "json": click.echo(json.dumps(payload, indent=2, sort_keys=True)) return diff --git a/polylogue/daemon/http.py b/polylogue/daemon/http.py index 727a54b199..d9e4726a2c 100644 --- a/polylogue/daemon/http.py +++ b/polylogue/daemon/http.py @@ -403,6 +403,11 @@ def _authenticated_post_routes() -> tuple[_StaticPostRoute, ...]: _StaticPostRoute("/api/ingest", ("api", "ingest"), "_handle_ingest"), _StaticPostRoute("/api/maintenance/plan", ("api", "maintenance", "plan"), "_handle_maintenance_plan"), _StaticPostRoute("/api/maintenance/run", ("api", "maintenance", "run"), "_handle_maintenance_run"), + _StaticPostRoute( + "/api/maintenance/rebuild-index", + ("api", "maintenance", "rebuild-index"), + "_handle_rebuild_index", + ), ) @@ -4787,6 +4792,94 @@ def _handle_maintenance_run(self) -> None: envelope = envelope_from_operation(result, origin="daemon", mode="execute") self._send_json(HTTPStatus.OK, envelope.to_dict()) + @daemon_safe_handler + def _handle_rebuild_index(self) -> None: + """POST /api/maintenance/rebuild-index — one coordinator-owned replay pass.""" + content_length = int(self.headers.get("Content-Length", 0)) + body_raw = self.rfile.read(content_length) if content_length > 0 else b"{}" + try: + body = json.loads(body_raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + if not isinstance(body, dict): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + raw_ids_value = body.get("raw_ids", []) + if not isinstance(raw_ids_value, list) or not all( + isinstance(raw_id, str) and raw_id for raw_id in raw_ids_value + ): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + only_missing = body.get("only_missing", False) + promote = body.get("promote", True) + max_blob_mb = body.get("max_blob_mb") + operation_id = body.get("operation_id") + raw_batch_size = body.get("raw_batch_size", 500) + pass_byte_budget_mb = body.get("pass_byte_budget_mb") + pass_deadline_seconds = body.get("pass_deadline_seconds") + if not isinstance(only_missing, bool) or not isinstance(promote, bool): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + if max_blob_mb is not None and ( + isinstance(max_blob_mb, bool) or not isinstance(max_blob_mb, int | float) or max_blob_mb <= 0 + ): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + if operation_id is not None and (not isinstance(operation_id, str) or not operation_id): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + if isinstance(raw_batch_size, bool) or not isinstance(raw_batch_size, int): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + if pass_byte_budget_mb is not None and ( + isinstance(pass_byte_budget_mb, bool) or not isinstance(pass_byte_budget_mb, int | float) + ): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + if pass_deadline_seconds is not None and ( + isinstance(pass_deadline_seconds, bool) or not isinstance(pass_deadline_seconds, int | float) + ): + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + + from polylogue.maintenance.rebuild_index import ( + RebuildIndexRequest, + rebuild_index_from_source_sync, + validate_rebuild_index_request, + ) + from polylogue.paths import archive_root + + request = RebuildIndexRequest( + archive_root=archive_root(), + only_missing=only_missing, + raw_ids=tuple(raw_ids_value), + max_blob_mb=float(max_blob_mb) if max_blob_mb is not None else None, + promote=promote, + operation_id=operation_id, + raw_batch_size=raw_batch_size, + pass_byte_budget_mb=float(pass_byte_budget_mb) if pass_byte_budget_mb is not None else None, + pass_deadline_seconds=(float(pass_deadline_seconds) if pass_deadline_seconds is not None else None), + ) + try: + validate_rebuild_index_request(request) + except ValueError: + self._send_error(HTTPStatus.BAD_REQUEST, "invalid_request") + return + + bridge = getattr(self.server, "write_bridge", None) + if bridge is None: + # Direct handler unit tests predate the server-owned bridge; real + # daemon servers always install it and therefore use run_sync. + receipt = rebuild_index_from_source_sync(request) + else: + receipt = cast(DaemonWriteThreadBridge, bridge).run_sync( + "http.maintenance.rebuild-index", + rebuild_index_from_source_sync, + request, + ) + self._send_json(HTTPStatus.OK, receipt.to_dict()) + @daemon_safe_handler def _handle_maintenance_status(self, operation_id: str) -> None: """GET /api/maintenance/status/ — delegate to maintenance_registry_http.""" diff --git a/polylogue/daemon/route_contracts.py b/polylogue/daemon/route_contracts.py index a9eb7930d3..edbd90c869 100644 --- a/polylogue/daemon/route_contracts.py +++ b/polylogue/daemon/route_contracts.py @@ -228,6 +228,15 @@ class RouteContract: "SearchEnvelope / SessionListResponse with route_state", "Local UDS-only root-request parameter envelope; daemon owns query compilation.", ), + RouteContract( + "POST", + "/api/maintenance/rebuild-index", + "maintenance", + "operational", + "bearer_if_configured_and_same_origin", + "RebuildIndexReceipt", + "Runs exactly one source snapshot replay through the daemon write coordinator.", + ), RouteContract( "GET", "/api/facets", diff --git a/polylogue/daemon/write_coordinator.py b/polylogue/daemon/write_coordinator.py index feeb19fad5..0f6ed63fd9 100644 --- a/polylogue/daemon/write_coordinator.py +++ b/polylogue/daemon/write_coordinator.py @@ -400,6 +400,18 @@ async def wait_for_release() -> None: future.cancel() logger.warning("timed out releasing daemon write gate actor=%s", actor) + def run_sync(self, actor: str, function: Callable[P, T], /, *args: P.args, **kwargs: P.kwargs) -> T: + """Run a blocking request operation through the daemon's sole writer. + + Unlike :meth:`hold`, this is for a complete bounded request operation: + the coordinator owns the worker thread until the function has really + returned, so a timed-out HTTP caller never admits a second writer. + """ + future = asyncio.run_coroutine_threadsafe( + self._coordinator.run_sync(actor, function, *args, **kwargs), self._loop + ) + return future.result(timeout=self._timeout) + def daemon_write_telemetry_payload() -> dict[str, object]: """Return the bounded process-global writer state for status surfaces.""" diff --git a/polylogue/maintenance/rebuild_index.py b/polylogue/maintenance/rebuild_index.py new file mode 100644 index 0000000000..42fa1fd6a3 --- /dev/null +++ b/polylogue/maintenance/rebuild_index.py @@ -0,0 +1,378 @@ +"""Daemon-safe source-to-index rebuild execution. + +The operation owns the write-side rebuild protocol; CLI and HTTP are adapters. +Callers must hold the daemon writer coordinator for an online rebuild. The +offline guard rejects every other live-daemon caller, preserving break-glass +operation after the daemon has stopped. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import sqlite3 +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import cast + +from polylogue.config import Config +from polylogue.maintenance.offline_guard import offline_maintenance_block_reason +from polylogue.paths import render_root +from polylogue.storage.archive_identity import ArchiveLocation + + +@dataclass(frozen=True, slots=True) +class RebuildIndexRequest: + """One bounded source snapshot replay request.""" + + archive_root: Path + only_missing: bool = False + raw_ids: tuple[str, ...] = () + max_blob_mb: float | None = None + promote: bool = True + operation_id: str | None = None + raw_batch_size: int = 500 + pass_byte_budget_mb: float | None = None + pass_deadline_seconds: float | None = None + + +@dataclass(frozen=True, slots=True) +class RebuildIndexReceipt: + """Typed evidence emitted after one source-to-index rebuild pass.""" + + archive_root: str + raw_session_count: int + selected_raw_count: int + skipped_by_blob_limit_count: int + status: str + materialized: bool + materialization: dict[str, object] + generation: dict[str, object] + readiness: dict[str, object] + replay: dict[str, object] + transaction: dict[str, object] | None = None + + def to_dict(self) -> dict[str, object]: + return { + "archive_root": self.archive_root, + "raw_session_count": self.raw_session_count, + "selected_raw_count": self.selected_raw_count, + "skipped_by_blob_limit_count": self.skipped_by_blob_limit_count, + "status": self.status, + "materialized": self.materialized, + "materialization": self.materialization, + "generation": self.generation, + "readiness": self.readiness, + "transaction": self.transaction, + **self.replay, + } + + +def validate_rebuild_index_request(request: RebuildIndexRequest) -> None: + """Reject selection and transaction combinations that cannot be promoted safely.""" + if request.raw_ids and request.only_missing: + raise ValueError("--raw-id cannot be combined with --only-missing") + if (request.raw_ids or request.only_missing) and request.promote: + raise ValueError("partial rebuild selections require --no-promote and can never replace the active index") + if request.max_blob_mb is not None and request.max_blob_mb <= 0: + raise ValueError("max blob size must be positive") + if request.max_blob_mb is not None and not request.raw_ids and not request.only_missing: + raise ValueError("--max-blob-mb requires --only-missing or --raw-id") + if request.raw_batch_size <= 0: + raise ValueError("raw batch size must be positive") + if request.pass_byte_budget_mb is not None and request.pass_byte_budget_mb <= 0: + raise ValueError("pass byte budget must be positive") + if request.pass_deadline_seconds is not None and request.pass_deadline_seconds <= 0: + raise ValueError("pass deadline must be positive") + if request.operation_id is not None and ( + request.raw_ids or request.only_missing or request.max_blob_mb is not None + ): + raise ValueError("--operation-id only resumes an unfiltered full-source rebuild") + if request.operation_id is not None and ( + request.pass_byte_budget_mb is not None or request.pass_deadline_seconds is not None + ): + raise ValueError("resumed rebuild budgets are durable; omit pass budget options with --operation-id") + + +def count_source_raw_sessions(root: Path) -> int: + source_db = root / "source.db" + if not source_db.exists(): + return 0 + with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: + row = conn.execute("SELECT COUNT(*) FROM raw_sessions").fetchone() + return int(row[0]) if row is not None else 0 + + +def missing_index_raw_ids(root: Path) -> list[str]: + source_db = root / "source.db" + index_db = ArchiveLocation.resolve(root).active_index_path + if not source_db.exists() or not index_db.exists(): + return [] + with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: + conn.execute("ATTACH DATABASE ? AS idx", (str(index_db),)) + rows = conn.execute( + """ + SELECT r.raw_id FROM raw_sessions r + WHERE NOT EXISTS (SELECT 1 FROM idx.sessions s WHERE s.raw_id = r.raw_id) + ORDER BY r.acquired_at_ms, r.raw_id + """ + ).fetchall() + return [str(row[0]) for row in rows] + + +def all_index_rebuild_raw_ids(root: Path) -> list[str]: + source_db = root / "source.db" + if not source_db.exists(): + return [] + with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: + rows = conn.execute("SELECT raw_id FROM raw_sessions ORDER BY acquired_at_ms, raw_id").fetchall() + return [str(row[0]) for row in rows] + + +def filter_raw_ids_by_max_blob_size(root: Path, raw_ids: list[str], max_blob_mb: float | None) -> list[str]: + if max_blob_mb is None or not raw_ids: + return raw_ids + source_db = root / "source.db" + placeholders = ",".join("?" for _ in raw_ids) + with contextlib.closing(sqlite3.connect(f"file:{source_db}?mode=ro", uri=True, timeout=10.0)) as conn: + rows = conn.execute( + f"SELECT raw_id FROM raw_sessions WHERE raw_id IN ({placeholders}) AND blob_size <= ? " + "ORDER BY acquired_at_ms, raw_id", + (*raw_ids, int(max_blob_mb * 1024 * 1024)), + ).fetchall() + return [str(row[0]) for row in rows] + + +def select_rebuild_raw_ids(request: RebuildIndexRequest) -> tuple[int, list[str], int]: + """Select source rows deterministically before the replay starts.""" + root = request.archive_root + raw_count = count_source_raw_sessions(root) + raw_ids = ( + list(dict.fromkeys(request.raw_ids)) + if request.raw_ids + else missing_index_raw_ids(root) + if request.only_missing + else all_index_rebuild_raw_ids(root) + ) + unfiltered_count = len(raw_ids) + selected = filter_raw_ids_by_max_blob_size(root, raw_ids, request.max_blob_mb) + return raw_count, selected, unfiltered_count - len(selected) + + +async def rebuild_index_from_source(request: RebuildIndexRequest) -> RebuildIndexReceipt: + """Replay one source snapshot into an owned generation and optionally promote it.""" + from polylogue.cli.commands.status import _archive_readiness_status + from polylogue.maintenance.replay import rebuild_index_from_source as replay_source + from polylogue.storage.index_generation import IndexGenerationStore, RebuildLease, source_revision_snapshot + from polylogue.storage.repair import repair_session_insights + + validate_rebuild_index_request(request) + root = request.archive_root + active_config = Config( + archive_root=root, + render_root=render_root(), + sources=[], + db_path=ArchiveLocation.resolve(root).active_index_path, + ) + if reason := offline_maintenance_block_reason(active_config, active=True, dry_run=False): + raise RuntimeError(reason) + + generation_store = IndexGenerationStore(root) + with RebuildLease(root): + raw_count = count_source_raw_sessions(root) + if raw_count == 0: + return RebuildIndexReceipt( + archive_root=str(root), + raw_session_count=0, + selected_raw_count=0, + skipped_by_blob_limit_count=0, + status="empty-source", + materialized=False, + materialization={}, + generation={}, + readiness={}, + replay={}, + ) + resumable_full_source = not request.raw_ids and not request.only_missing and request.max_blob_mb is None + transaction = None + page = None + pass_started_at_ms = int(time.time() * 1000) + if resumable_full_source: + transaction = ( + generation_store.load_transaction(request.operation_id) + if request.operation_id is not None + else generation_store.create_transaction( + source_snapshot=source_revision_snapshot(root), + pass_byte_budget=( + int(request.pass_byte_budget_mb * 1024 * 1024) + if request.pass_byte_budget_mb is not None + else None + ), + pass_deadline_ms=( + int(request.pass_deadline_seconds * 1000) if request.pass_deadline_seconds is not None else None + ), + ) + ) + if transaction.status in {"promoted", "stale"}: + raise RuntimeError( + f"rebuild operation {transaction.operation_id} is {transaction.status}; start a new operation" + ) + if source_revision_snapshot(root) != transaction.source_snapshot: + generation_store.checkpoint_transaction( + transaction, + status="stale", + error="source evidence changed since this rebuild was planned", + ) + raise RuntimeError( + f"rebuild operation {transaction.operation_id} is stale because source evidence changed" + ) + generation = generation_store.load(transaction.generation_id) + if generation.owner_id != transaction.generation_owner_id or generation.state != "inactive": + raise RuntimeError(f"rebuild operation {transaction.operation_id} lost its inactive candidate") + page = generation_store.next_raw_page(transaction, limit=request.raw_batch_size) + selected_raw_ids = [raw_id for raw_id, _acquired_at_ms, _blob_size in page.rows] + selected_raw_count = len(selected_raw_ids) + skipped_by_blob_limit_count = 0 + else: + raw_count, selected_raw_ids, skipped_by_blob_limit_count = select_rebuild_raw_ids(request) + selected_raw_count = len(selected_raw_ids) + generation = generation_store.create(source_snapshot=source_revision_snapshot(root)) + source_drifted = False + try: + generation_root = Path(generation.index_path).parent + config = Config( + archive_root=generation_root, + render_root=render_root(), + sources=[], + db_path=Path(generation.index_path), + ) + replay = await replay_source( + config, + raw_ids=selected_raw_ids, + raw_batch_size=request.raw_batch_size, + ingest_workers=None, + materialize=True, + progress_callback=None, + owned_inactive_generation=(generation.generation_id, generation.owner_id), + ) + if transaction is not None and selected_raw_ids: + if source_revision_snapshot(root) != transaction.source_snapshot: + transaction = generation_store.checkpoint_transaction( + transaction, + status="stale", + error="source evidence changed during this bounded rebuild pass", + ) + source_drifted = True + raise RuntimeError( + f"rebuild operation {transaction.operation_id} is stale because source evidence changed" + ) + assert page is not None + last_raw_id, last_acquired_at_ms, _blob_size = page.rows[-1] + elapsed_ms = int(time.time() * 1000) - pass_started_at_ms + deadline_expired = ( + transaction.pass_deadline_ms is not None and elapsed_ms >= transaction.pass_deadline_ms + ) + status = "deferred" if page.deferred_reason == "byte-budget" or deadline_expired else "paused" + transaction = generation_store.checkpoint_transaction( + transaction, + status=status, + last_acquired_at_ms=last_acquired_at_ms, + last_raw_id=last_raw_id, + processed_raw_count=transaction.processed_raw_count + len(selected_raw_ids), + processed_blob_bytes=transaction.processed_blob_bytes + sum(row[2] for row in page.rows), + ) + if page.has_more or deadline_expired: + return RebuildIndexReceipt( + archive_root=str(root), + raw_session_count=raw_count, + selected_raw_count=selected_raw_count, + skipped_by_blob_limit_count=0, + status=status, + materialized=False, + materialization={}, + generation=cast(dict[str, object], asdict(generation)), + readiness={}, + replay=replay, + transaction=cast(dict[str, object], asdict(transaction)), + ) + insight_result = repair_session_insights( + config, + dry_run=False, + archive_root_override=generation_root, + owned_inactive_generation=(generation.generation_id, generation.owner_id), + ) + if not insight_result.success: + raise RuntimeError(f"session insight materialization failed: {insight_result.detail}") + if source_revision_snapshot(root) != generation.source_snapshot: + if transaction is not None: + transaction = generation_store.checkpoint_transaction( + transaction, + status="stale", + error="source evidence changed before terminal readiness", + ) + source_drifted = True + raise RuntimeError(f"source evidence changed while rebuilding {generation.generation_id}") + readiness = _archive_readiness_status(generation_root) + if not readiness.get("checked") or int(readiness.get("blocked_surface_count", 1)) != 0: + blocked = [ + name + for name, info in cast(dict[str, dict[str, object]], readiness.get("surfaces", {})).items() + if info.get("ready") is not True + ] + detail = ( + f"reason: {readiness.get('reason')}" + if not readiness.get("checked") + else "blocked surfaces: " + ", ".join(blocked) + ) + raise RuntimeError(f"inactive generation {generation.generation_id} is not exact-ready; {detail}") + if transaction is not None: + transaction = generation_store.checkpoint_transaction(transaction, status="ready") + if request.promote: + generation = generation_store.promote(generation) + if transaction is not None: + transaction = generation_store.checkpoint_transaction(transaction, status="promoted") + except Exception: + if transaction is not None and not source_drifted: + with contextlib.suppress(Exception): + generation_store.checkpoint_transaction( + transaction, + status="failed", + error="bounded rebuild pass failed; candidate retained for diagnosis or explicit recovery", + ) + else: + with contextlib.suppress(Exception): + generation_store.discard_if_inactive(generation) + raise + return RebuildIndexReceipt( + archive_root=str(root), + raw_session_count=raw_count, + selected_raw_count=selected_raw_count, + skipped_by_blob_limit_count=skipped_by_blob_limit_count, + status="replayed", + materialized=True, + materialization=cast(dict[str, object], insight_result.to_dict()), + generation=cast(dict[str, object], asdict(generation)), + readiness=cast(dict[str, object], readiness), + replay=replay, + transaction=cast(dict[str, object], asdict(transaction)) if transaction is not None else None, + ) + + +def rebuild_index_from_source_sync(request: RebuildIndexRequest) -> RebuildIndexReceipt: + """Synchronous adapter for offline CLI callers.""" + return asyncio.run(rebuild_index_from_source(request)) + + +__all__ = [ + "RebuildIndexReceipt", + "RebuildIndexRequest", + "all_index_rebuild_raw_ids", + "count_source_raw_sessions", + "filter_raw_ids_by_max_blob_size", + "missing_index_raw_ids", + "rebuild_index_from_source", + "rebuild_index_from_source_sync", + "select_rebuild_raw_ids", + "validate_rebuild_index_request", +] diff --git a/tests/unit/cli/test_archive_maintenance_cli.py b/tests/unit/cli/test_archive_maintenance_cli.py index c4e328c867..95e30a896f 100644 --- a/tests/unit/cli/test_archive_maintenance_cli.py +++ b/tests/unit/cli/test_archive_maintenance_cli.py @@ -1634,13 +1634,13 @@ def test_rebuild_index_source_replay_expands_every_execution_selection_to_author monkeypatch: pytest.MonkeyPatch, selection_args: list[str], ) -> None: - monkeypatch.setattr("polylogue.cli.commands.maintenance._rebuild_index._count_source_raw_sessions", lambda _root: 4) + monkeypatch.setattr("polylogue.maintenance.rebuild_index.count_source_raw_sessions", lambda _root: 4) monkeypatch.setattr( - "polylogue.cli.commands.maintenance._rebuild_index._all_index_rebuild_raw_ids", + "polylogue.maintenance.rebuild_index.all_index_rebuild_raw_ids", lambda _root: ["raw-parent", "raw-child"], ) monkeypatch.setattr( - "polylogue.cli.commands.maintenance._rebuild_index._missing_index_raw_ids", + "polylogue.maintenance.rebuild_index.missing_index_raw_ids", lambda _root: ["raw-parent", "raw-child"], ) @@ -1675,6 +1675,73 @@ def test_rebuild_index_force_write_option_is_retired(cli_runner: CliRunner) -> N assert "--force-write" in result.output +def test_rebuild_index_daemon_path_posts_the_real_selection_request( + cli_workspace: dict[str, Path], cli_runner: CliRunner, monkeypatch: pytest.MonkeyPatch +) -> None: + captured: dict[str, object] = {} + + class Response: + def read(self) -> bytes: + return json.dumps( + { + "archive_root": str(cli_workspace["archive_root"]), + "classified_full_count": 2, + "replayed_logical_source_count": 1, + "quarantined_raw_count": 0, + } + ).encode() + + def __enter__(self) -> Response: + return self + + def __exit__(self, *_args: object) -> None: + return None + + def fake_urlopen(request: object, *, timeout: int) -> Response: + captured["url"] = request.full_url # type: ignore[attr-defined] + captured["body"] = json.loads(request.data) # type: ignore[attr-defined] + captured["timeout"] = timeout + return Response() + + monkeypatch.setattr(maintenance_rebuild_index, "urlopen", fake_urlopen) + result = cli_runner.invoke( + cli, + [ + "--plain", + "ops", + "maintenance", + "rebuild-index", + "--daemon", + "--daemon-url", + "http://127.0.0.1:9876", + "--raw-batch-size", + "17", + "--pass-byte-budget-mb", + "12.5", + "--pass-deadline-seconds", + "45", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert captured == { + "url": "http://127.0.0.1:9876/api/maintenance/rebuild-index", + "body": { + "only_missing": False, + "raw_ids": [], + "max_blob_mb": None, + "promote": True, + "operation_id": None, + "raw_batch_size": 17, + "pass_byte_budget_mb": 12.5, + "pass_deadline_seconds": 45.0, + }, + "timeout": 600, + } + assert "Classified:" in result.output + + @pytest.mark.parametrize("selection_args", [["--only-missing"], ["--raw-id", "raw-a"]]) def test_partial_rebuild_requires_no_promote_before_archive_mutation( cli_workspace: dict[str, Path], cli_runner: CliRunner, selection_args: list[str] @@ -1894,7 +1961,7 @@ def test_rebuild_index_deadline_defers_postflight_until_resume( ) clock = [100.0, 102.0] monkeypatch.setattr( - "polylogue.cli.commands.maintenance._rebuild_index.time.time", + "polylogue.maintenance.rebuild_index.time.time", lambda: clock.pop(0) if clock else 102.0, ) first = cli_runner.invoke( diff --git a/tests/unit/daemon/test_http_write_coordination.py b/tests/unit/daemon/test_http_write_coordination.py index 16630e5927..24b26edf95 100644 --- a/tests/unit/daemon/test_http_write_coordination.py +++ b/tests/unit/daemon/test_http_write_coordination.py @@ -25,6 +25,10 @@ def hold(self, actor: str) -> Iterator[None]: finally: self.timeline.append(f"exit:{actor}") + def run_sync(self, actor: str, function: Callable[..., object], *args: object) -> object: + self.timeline.append(f"run_sync:{actor}") + return function(*args) + def _handler(path: list[str], timeline: list[str]) -> DaemonAPIHandler: def allow_auth(required_scope: WebCredentialScope = "read", *, allow_web: bool = True) -> bool: @@ -92,6 +96,20 @@ def dispatch_delete(*_args: object) -> bool: ] +def test_rebuild_index_route_uses_the_bridge_run_sync_writer_path() -> None: + timeline: list[str] = [] + handler = _handler(["api", "maintenance", "rebuild-index"], timeline) + + def body() -> None: + bridge = handler.server.write_bridge + bridge.run_sync("http.maintenance.rebuild-index", lambda: timeline.append("body")) + + handler._handle_rebuild_index = body # type: ignore[method-assign] + handler._do_post_impl() + + assert timeline == ["run_sync:http.maintenance.rebuild-index", "body"] + + @pytest.mark.parametrize("signal", ["traces", "metrics", "logs"]) def test_otlp_persistence_route_holds_gate_around_receiver(signal: str) -> None: timeline: list[str] = [] diff --git a/tests/unit/daemon/test_maintenance_endpoints.py b/tests/unit/daemon/test_maintenance_endpoints.py index 5d813dbe15..3943ee8b34 100644 --- a/tests/unit/daemon/test_maintenance_endpoints.py +++ b/tests/unit/daemon/test_maintenance_endpoints.py @@ -76,6 +76,12 @@ def test_run_route_dispatched(self) -> None: handler.do_POST() mock.assert_called_once() + def test_rebuild_index_route_dispatched(self) -> None: + handler = _make_handler("/api/maintenance/rebuild-index", body={}) + with patch.object(handler, "_handle_rebuild_index") as mock: + handler.do_POST() + mock.assert_called_once() + def test_unknown_maintenance_post_route_404(self) -> None: """POST /api/maintenance/status returns 404 — status is GET-only.""" handler = _make_handler("/api/maintenance/status/x") @@ -150,6 +156,38 @@ def test_run_invalid_json_400(self) -> None: handler._handle_maintenance_run() mock.assert_called_once_with(HTTPStatus.BAD_REQUEST, "invalid_request") + def test_rebuild_index_runs_the_typed_service_inside_the_route_executor(self, tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def] + from polylogue.maintenance.rebuild_index import RebuildIndexReceipt + + monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(tmp_path)) + handler = _make_handler("/api/maintenance/rebuild-index", body={"promote": False, "raw_ids": ["raw-1"]}) + receipt = RebuildIndexReceipt( + archive_root=str(tmp_path), + raw_session_count=1, + selected_raw_count=1, + skipped_by_blob_limit_count=0, + status="replayed", + materialized=True, + materialization={}, + generation={"generation_id": "candidate-1", "active": False}, + readiness={"checked": True, "blocked_surface_count": 0}, + replay={"classified_full_count": 1, "replayed_logical_source_count": 1, "quarantined_raw_count": 0}, + ) + handler.server.write_bridge = type( + "Bridge", + (), + {"run_sync": lambda _self, _actor, function, *args: function(*args)}, + )() + with patch( + "polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync", return_value=receipt + ) as rebuild: + with patch.object(handler, "_send_json") as send: + handler._handle_rebuild_index() + request = rebuild.call_args.args[0] + assert request.raw_ids == ("raw-1",) + assert request.promote is False + assert send.call_args.args == (HTTPStatus.OK, receipt.to_dict()) + class TestMaintenanceRegistryEndpoints: """GET /api/maintenance/status/ and /api/maintenance/operations (#1197).""" diff --git a/tests/unit/daemon/test_route_contracts.py b/tests/unit/daemon/test_route_contracts.py index 27bc1a9dd3..02564dfb23 100644 --- a/tests/unit/daemon/test_route_contracts.py +++ b/tests/unit/daemon/test_route_contracts.py @@ -2,7 +2,10 @@ from __future__ import annotations +import json from http import HTTPStatus +from types import SimpleNamespace +from unittest.mock import MagicMock import pytest @@ -62,6 +65,57 @@ def test_get_dispatch_tables_are_bound_to_route_contracts() -> None: assert parameterized_route.pattern == parameterized_route.contract.pattern +def test_rebuild_index_handler_forwards_resumable_pass_options_through_writer_bridge() -> None: + """The live HTTP handler reaches the daemon bridge with the canonical request model. + + This fails if the handler drops any bounded-pass field before handing the + request to ``rebuild_index_from_source_sync``. + """ + from polylogue.maintenance.rebuild_index import RebuildIndexReceipt + + receipt = RebuildIndexReceipt( + archive_root="/archive", + raw_session_count=2, + selected_raw_count=1, + skipped_by_blob_limit_count=0, + status="deferred", + materialized=False, + materialization={}, + generation={"generation_id": "candidate"}, + readiness={}, + replay={"scheduled_raw_count": 1}, + transaction={"operation_id": "resume-1", "status": "deferred"}, + ) + bridge = SimpleNamespace(write_bridge=MagicMock()) + bridge.write_bridge.run_sync.return_value = receipt + handler = _make_handler( + "POST", + "/api/maintenance/rebuild-index", + body=json.dumps( + { + "raw_batch_size": 17, + "pass_byte_budget_mb": 12.5, + "pass_deadline_seconds": 45, + "promote": False, + } + ).encode(), + server=bridge, + ) + _send_error, send_json = _capture_responses(handler) + + handler._handle_rebuild_index() + + actor, function, request = bridge.write_bridge.run_sync.call_args.args + assert actor == "http.maintenance.rebuild-index" + assert function.__name__ == "rebuild_index_from_source_sync" + assert request.operation_id is None + assert request.raw_batch_size == 17 + assert request.pass_byte_budget_mb == 12.5 + assert request.pass_deadline_seconds == 45.0 + assert request.promote is False + send_json.assert_called_once_with(HTTPStatus.OK, receipt.to_dict()) + + def test_stable_routes_have_explicit_auth_and_response_contracts() -> None: """Stable routes must declare the security posture and response shape.""" diff --git a/tests/unit/daemon/test_web_auth.py b/tests/unit/daemon/test_web_auth.py index 38ffcd9a23..03ab9c3521 100644 --- a/tests/unit/daemon/test_web_auth.py +++ b/tests/unit/daemon/test_web_auth.py @@ -150,6 +150,7 @@ def test_bootstrap_rotates_http_only_cookie_and_authenticates_read_route() -> No ("/api/ingest", "_handle_ingest"), ("/api/maintenance/plan", "_handle_maintenance_plan"), ("/api/maintenance/run", "_handle_maintenance_run"), + ("/api/maintenance/rebuild-index", "_handle_rebuild_index"), ], ) def test_web_credential_cannot_execute_archive_control_routes(path: str, handler_name: str) -> None: