feat: implement durable permission-aware search

README.md

@@ -16,3 +16,23 @@ development fallback. Other modules may:
 
 An optional OpenSearch adapter is a later provider, not a hard dependency.
 Source modules remain responsible for defining visibility and authorization.
+
+## Index lifecycle
+
+Source modules register a versioned `search_sources` provider. A provider
+declares its resource types and index version, returns bounded resumable
+backfill pages, and batch-rechecks current authorization for sensitive
+resources. Incremental writes use `SearchIndexChange` and
+`search.index_writer.enqueue_change()` so change IDs are durable and
+idempotent in the same database transaction as the caller.
+
+The built-in backend exposes opaque cursor pagination and does not return
+pre-authorization totals. PostgreSQL uses full-text search and will add
+trigram indexes when `pg_trgm` is already installed; SQLite remains a bounded
+development fallback.
+
+Tenant search administrators can inspect `/api/v1/search/admin/diagnostics`,
+reconcile disabled modules, process queued changes, and start or continue
+resumable provider rebuilds. Rows requiring a source authorization recheck
+are omitted when their provider is unavailable, stale, or fails to return an
+explicit allow decision.

src/govoplan_search/backend/db/models.py

@@ -5,9 +5,11 @@ from datetime import datetime
 from typing import Any
 
 from sqlalchemy import (
+    Boolean,
     DateTime,
     ForeignKey,
     Index,
+    Integer,
     JSON,
     String,
     Text,
@@ -15,7 +17,7 @@ from sqlalchemy import (
 )
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
-from govoplan_core.db.base import Base, TimestampMixin
+from govoplan_core.db.base import Base, TimestampMixin, utcnow
 
 
 def new_uuid() -> str:
@@ -39,11 +41,23 @@ class SearchIndexDocument(Base, TimestampMixin):
             "resource_type",
         ),
         Index("ix_search_document_visibility", "tenant_id", "visibility"),
+        Index(
+            "ix_search_document_provider",
+            "tenant_id",
+            "provider_id",
+            "resource_type",
+            "active",
+        ),
     )
 
     id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
     tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
     module_id: Mapped[str] = mapped_column(String(100), nullable=False, index=True)
+    provider_id: Mapped[str] = mapped_column(
+        String(200),
+        nullable=False,
+        index=True,
+    )
     resource_type: Mapped[str] = mapped_column(
         String(100), nullable=False, index=True
     )
@@ -66,6 +80,44 @@ class SearchIndexDocument(Base, TimestampMixin):
         "metadata", JSON, default=dict, nullable=False
     )
     content_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+    source_revision: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+    )
+    change_cursor: Mapped[str | None] = mapped_column(
+        String(500),
+        nullable=True,
+    )
+    source_updated_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    language: Mapped[str] = mapped_column(
+        String(32),
+        default="simple",
+        nullable=False,
+    )
+    index_version: Mapped[int] = mapped_column(
+        Integer,
+        default=1,
+        nullable=False,
+    )
+    requires_authorization_recheck: Mapped[bool] = mapped_column(
+        Boolean,
+        default=False,
+        nullable=False,
+    )
+    rebuild_id: Mapped[str | None] = mapped_column(
+        String(36),
+        nullable=True,
+        index=True,
+    )
+    active: Mapped[bool] = mapped_column(
+        Boolean,
+        default=True,
+        nullable=False,
+        index=True,
+    )
     indexed_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False
     )
@@ -100,4 +152,204 @@ class SearchIndexAclToken(Base):
     )
 
 
-__all__ = ["SearchIndexAclToken", "SearchIndexDocument", "new_uuid"]
+class SearchIndexState(Base, TimestampMixin):
+    __tablename__ = "search_index_states"
+    __table_args__ = (
+        UniqueConstraint(
+            "tenant_id",
+            "provider_id",
+            "resource_type",
+            name="uq_search_index_state_source",
+        ),
+        Index(
+            "ix_search_index_state_status",
+            "tenant_id",
+            "status",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(
+        String(36),
+        primary_key=True,
+        default=new_uuid,
+    )
+    tenant_id: Mapped[str] = mapped_column(
+        String(36),
+        nullable=False,
+        index=True,
+    )
+    provider_id: Mapped[str] = mapped_column(
+        String(200),
+        nullable=False,
+        index=True,
+    )
+    module_id: Mapped[str] = mapped_column(
+        String(100),
+        nullable=False,
+        index=True,
+    )
+    resource_type: Mapped[str] = mapped_column(
+        String(100),
+        nullable=False,
+        index=True,
+    )
+    index_version: Mapped[int] = mapped_column(
+        Integer,
+        default=1,
+        nullable=False,
+    )
+    status: Mapped[str] = mapped_column(
+        String(30),
+        default="idle",
+        nullable=False,
+        index=True,
+    )
+    rebuild_id: Mapped[str | None] = mapped_column(
+        String(36),
+        nullable=True,
+    )
+    checkpoint_cursor: Mapped[str | None] = mapped_column(
+        String(500),
+        nullable=True,
+    )
+    high_watermark: Mapped[str | None] = mapped_column(
+        String(500),
+        nullable=True,
+    )
+    last_change_cursor: Mapped[str | None] = mapped_column(
+        String(500),
+        nullable=True,
+    )
+    indexed_documents: Mapped[int] = mapped_column(
+        Integer,
+        default=0,
+        nullable=False,
+    )
+    rejected_documents: Mapped[int] = mapped_column(
+        Integer,
+        default=0,
+        nullable=False,
+    )
+    rebuild_started_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    rebuild_completed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    last_success_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+
+class SearchIndexChangeQueue(Base, TimestampMixin):
+    __tablename__ = "search_index_change_queue"
+    __table_args__ = (
+        UniqueConstraint(
+            "change_id",
+            name="uq_search_index_change_queue_change",
+        ),
+        Index(
+            "ix_search_index_change_queue_pending",
+            "status",
+            "available_at",
+            "created_at",
+        ),
+        Index(
+            "ix_search_index_change_queue_source",
+            "tenant_id",
+            "provider_id",
+            "resource_type",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(
+        String(36),
+        primary_key=True,
+        default=new_uuid,
+    )
+    change_id: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+        index=True,
+    )
+    tenant_id: Mapped[str] = mapped_column(
+        String(36),
+        nullable=False,
+        index=True,
+    )
+    provider_id: Mapped[str] = mapped_column(
+        String(200),
+        nullable=False,
+        index=True,
+    )
+    module_id: Mapped[str] = mapped_column(
+        String(100),
+        nullable=False,
+        index=True,
+    )
+    resource_type: Mapped[str] = mapped_column(
+        String(100),
+        nullable=False,
+        index=True,
+    )
+    resource_id: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+        index=True,
+    )
+    kind: Mapped[str] = mapped_column(
+        String(20),
+        nullable=False,
+    )
+    source_revision: Mapped[str] = mapped_column(
+        String(255),
+        nullable=False,
+    )
+    source_cursor: Mapped[str] = mapped_column(
+        String(500),
+        nullable=False,
+    )
+    document_: Mapped[dict[str, Any] | None] = mapped_column(
+        "document",
+        JSON,
+        nullable=True,
+    )
+    occurred_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    status: Mapped[str] = mapped_column(
+        String(30),
+        default="queued",
+        nullable=False,
+        index=True,
+    )
+    attempts: Mapped[int] = mapped_column(
+        Integer,
+        default=0,
+        nullable=False,
+    )
+    available_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        default=utcnow,
+        nullable=False,
+        index=True,
+    )
+    processed_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True),
+        nullable=True,
+    )
+    error: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+
+__all__ = [
+    "SearchIndexAclToken",
+    "SearchIndexChangeQueue",
+    "SearchIndexDocument",
+    "SearchIndexState",
+    "new_uuid",
+]

src/govoplan_search/backend/manifest.py

@@ -118,7 +118,8 @@ manifest = ModuleManifest(
     ),
     provides_interfaces=(
         ModuleInterfaceProvider(name="search.provider", version="1.0.0"),
-        ModuleInterfaceProvider(name="search.index_writer", version="1.0.0"),
+        ModuleInterfaceProvider(name="search.index_writer", version="1.1.0"),
+        ModuleInterfaceProvider(name="search.source", version="1.0.0"),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -158,8 +159,10 @@ manifest = ModuleManifest(
         script_location=str(Path(__file__).with_name("migrations") / "versions"),
         retirement_supported=True,
         retirement_provider=drop_table_retirement_provider(
+            search_models.SearchIndexChangeQueue,
             search_models.SearchIndexAclToken,
             search_models.SearchIndexDocument,
+            search_models.SearchIndexState,
             label="Search",
         ),
         retirement_notes=(
@@ -169,7 +172,9 @@ manifest = ModuleManifest(
     ),
     uninstall_guard_providers=(
         persistent_table_uninstall_guard(
+            search_models.SearchIndexChangeQueue,
             search_models.SearchIndexDocument,
+            search_models.SearchIndexState,
             label="Search index",
         ),
     ),

src/govoplan_search/backend/migrations/versions/b2c3d4e5f607_v0114_search_lifecycle.py

@@ -0,0 +1,311 @@
+"""Add versioned search indexing lifecycle.
+
+Revision ID: b2c3d4e5f607
+Revises: a1b2c3d4e5f6
+Create Date: 2026-07-29
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "b2c3d4e5f607"
+down_revision = "a1b2c3d4e5f6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    for column in (
+        sa.Column(
+            "provider_id",
+            sa.String(length=200),
+            nullable=False,
+            server_default="legacy.index",
+        ),
+        sa.Column(
+            "source_revision",
+            sa.String(length=255),
+            nullable=False,
+            server_default="1",
+        ),
+        sa.Column(
+            "change_cursor",
+            sa.String(length=500),
+            nullable=True,
+        ),
+        sa.Column(
+            "source_updated_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+        sa.Column(
+            "language",
+            sa.String(length=32),
+            nullable=False,
+            server_default="simple",
+        ),
+        sa.Column(
+            "index_version",
+            sa.Integer(),
+            nullable=False,
+            server_default="1",
+        ),
+        sa.Column(
+            "requires_authorization_recheck",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.false(),
+        ),
+        sa.Column(
+            "rebuild_id",
+            sa.String(length=36),
+            nullable=True,
+        ),
+        sa.Column(
+            "active",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.true(),
+        ),
+    ):
+        op.add_column("search_index_documents", column)
+    for column in ("provider_id", "rebuild_id", "active"):
+        op.create_index(
+            op.f(f"ix_search_index_documents_{column}"),
+            "search_index_documents",
+            [column],
+        )
+    op.create_index(
+        "ix_search_document_provider",
+        "search_index_documents",
+        [
+            "tenant_id",
+            "provider_id",
+            "resource_type",
+            "active",
+        ],
+    )
+    bind = op.get_bind()
+    if (
+        bind.dialect.name == "postgresql"
+        and bind.scalar(
+            sa.text(
+                "SELECT EXISTS ("
+                "SELECT 1 FROM pg_extension "
+                "WHERE extname = 'pg_trgm'"
+                ")"
+            )
+        )
+    ):
+        op.execute(
+            "CREATE INDEX ix_search_document_title_trgm "
+            "ON search_index_documents USING gin "
+            "(lower(title) gin_trgm_ops)"
+        )
+        op.execute(
+            "CREATE INDEX ix_search_document_text_trgm "
+            "ON search_index_documents USING gin "
+            "(lower(search_text) gin_trgm_ops)"
+        )
+
+    op.create_table(
+        "search_index_states",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("provider_id", sa.String(length=200), nullable=False),
+        sa.Column("module_id", sa.String(length=100), nullable=False),
+        sa.Column("resource_type", sa.String(length=100), nullable=False),
+        sa.Column("index_version", sa.Integer(), nullable=False),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("rebuild_id", sa.String(length=36), nullable=True),
+        sa.Column(
+            "checkpoint_cursor",
+            sa.String(length=500),
+            nullable=True,
+        ),
+        sa.Column(
+            "high_watermark",
+            sa.String(length=500),
+            nullable=True,
+        ),
+        sa.Column(
+            "last_change_cursor",
+            sa.String(length=500),
+            nullable=True,
+        ),
+        sa.Column("indexed_documents", sa.Integer(), nullable=False),
+        sa.Column("rejected_documents", sa.Integer(), nullable=False),
+        sa.Column(
+            "rebuild_started_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+        sa.Column(
+            "rebuild_completed_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+        sa.Column(
+            "last_success_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+        sa.Column("last_error", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint(
+            "id",
+            name=op.f("pk_search_index_states"),
+        ),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "provider_id",
+            "resource_type",
+            name="uq_search_index_state_source",
+        ),
+    )
+    for column in (
+        "tenant_id",
+        "provider_id",
+        "module_id",
+        "resource_type",
+        "status",
+    ):
+        op.create_index(
+            op.f(f"ix_search_index_states_{column}"),
+            "search_index_states",
+            [column],
+        )
+    op.create_index(
+        "ix_search_index_state_status",
+        "search_index_states",
+        ["tenant_id", "status"],
+    )
+
+    op.create_table(
+        "search_index_change_queue",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("change_id", sa.String(length=255), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("provider_id", sa.String(length=200), nullable=False),
+        sa.Column("module_id", sa.String(length=100), nullable=False),
+        sa.Column("resource_type", sa.String(length=100), nullable=False),
+        sa.Column("resource_id", sa.String(length=255), nullable=False),
+        sa.Column("kind", sa.String(length=20), nullable=False),
+        sa.Column(
+            "source_revision",
+            sa.String(length=255),
+            nullable=False,
+        ),
+        sa.Column(
+            "source_cursor",
+            sa.String(length=500),
+            nullable=False,
+        ),
+        sa.Column("document", sa.JSON(), nullable=True),
+        sa.Column(
+            "occurred_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("attempts", sa.Integer(), nullable=False),
+        sa.Column(
+            "available_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+        ),
+        sa.Column(
+            "processed_at",
+            sa.DateTime(timezone=True),
+            nullable=True,
+        ),
+        sa.Column("error", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+        ),
+        sa.PrimaryKeyConstraint(
+            "id",
+            name=op.f("pk_search_index_change_queue"),
+        ),
+        sa.UniqueConstraint(
+            "change_id",
+            name="uq_search_index_change_queue_change",
+        ),
+    )
+    for column in (
+        "change_id",
+        "tenant_id",
+        "provider_id",
+        "module_id",
+        "resource_type",
+        "resource_id",
+        "status",
+        "available_at",
+    ):
+        op.create_index(
+            op.f(f"ix_search_index_change_queue_{column}"),
+            "search_index_change_queue",
+            [column],
+        )
+    op.create_index(
+        "ix_search_index_change_queue_pending",
+        "search_index_change_queue",
+        ["status", "available_at", "created_at"],
+    )
+    op.create_index(
+        "ix_search_index_change_queue_source",
+        "search_index_change_queue",
+        ["tenant_id", "provider_id", "resource_type"],
+    )
+
+
+def downgrade() -> None:
+    if op.get_bind().dialect.name == "postgresql":
+        op.execute(
+            "DROP INDEX IF EXISTS ix_search_document_text_trgm"
+        )
+        op.execute(
+            "DROP INDEX IF EXISTS ix_search_document_title_trgm"
+        )
+    op.drop_table("search_index_change_queue")
+    op.drop_table("search_index_states")
+    op.drop_index(
+        "ix_search_document_provider",
+        table_name="search_index_documents",
+    )
+    for column in ("active", "rebuild_id", "provider_id"):
+        op.drop_index(
+            op.f(f"ix_search_index_documents_{column}"),
+            table_name="search_index_documents",
+        )
+    for column in (
+        "active",
+        "rebuild_id",
+        "requires_authorization_recheck",
+        "index_version",
+        "language",
+        "source_updated_at",
+        "change_cursor",
+        "source_revision",
+        "provider_id",
+    ):
+        op.drop_column("search_index_documents", column)

src/govoplan_search/backend/router.py

@@ -4,17 +4,27 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from sqlalchemy.orm import Session
 
 from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
+from govoplan_core.audit.logging import audit_from_principal
+from govoplan_core.core.search import CAPABILITY_SEARCH_INDEX_WRITER
 from govoplan_core.core.registry import PlatformRegistry
 from govoplan_core.core.search import SearchQuery
 from govoplan_core.db.session import get_session
-from govoplan_search.backend.manifest import READ_SCOPE
+from govoplan_search.backend.manifest import ADMIN_SCOPE, READ_SCOPE
 from govoplan_search.backend.schemas import (
+    SearchChangeDispatchResponse,
+    SearchDiagnosticsResponse,
+    SearchIndexStateResponse,
+    SearchModuleReconcileResponse,
     SearchProviderListResponse,
     SearchProviderResponse,
+    SearchRebuildResponse,
     SearchResponse,
     SearchResultResponse,
 )
-from govoplan_search.backend.service import aggregate_search
+from govoplan_search.backend.service import (
+    SearchIndexService,
+    aggregate_search_page,
+)
 
 
 router = APIRouter(prefix="/search", tags=["search"])
@@ -38,6 +48,26 @@ def _require_read(principal: ApiPrincipal) -> None:
         )
 
 
+def _require_admin(principal: ApiPrincipal) -> None:
+    if not has_scope(principal, ADMIN_SCOPE):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=f"Missing scope: {ADMIN_SCOPE}",
+        )
+
+
+def _service(registry: PlatformRegistry) -> SearchIndexService:
+    capability = registry.capability(
+        CAPABILITY_SEARCH_INDEX_WRITER
+    )
+    if not isinstance(capability, SearchIndexService):
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Search index service is not available.",
+        )
+    return capability
+
+
 @router.get("", response_model=SearchResponse)
 def api_search(
     request: Request,
@@ -48,11 +78,18 @@ def api_search(
     context_id: str | None = Query(default=None, max_length=255),
     limit: int = Query(default=25, ge=1, le=100),
     offset: int = Query(default=0, ge=0, le=10_000),
+    cursor: str | None = Query(default=None, max_length=2000),
+    language: str = Query(
+        default="simple",
+        max_length=32,
+        pattern=r"^[A-Za-z_]+$",
+    ),
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(get_api_principal),
 ) -> SearchResponse:
     _require_read(principal)
     registry = _registry(request)
+    try:
         query = SearchQuery(
             text=q,
             tenant_id=principal.tenant_id,
@@ -62,13 +99,20 @@ def api_search(
             context_id=context_id,
             limit=limit,
             offset=offset,
+            cursor=cursor,
+            language=language.casefold(),
         )
-    results, diagnostics = aggregate_search(
+        page = aggregate_search_page(
             registry,
             session,
             principal,
             query=query,
         )
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
     return SearchResponse(
         query=query.text,
         results=[
@@ -90,11 +134,15 @@ def api_search(
                         else None
                     ),
                     "metadata": dict(result.metadata),
+                    "source_revision": result.source_revision,
+                    "provenance": dict(result.provenance),
                 }
             )
-            for result in results
+            for result in page.results
         ],
-        diagnostics=list(diagnostics),
+        diagnostics=list(page.diagnostics),
+        next_cursor=page.next_cursor,
+        has_more=page.next_cursor is not None,
     )
 
 
@@ -118,4 +166,190 @@ def api_search_providers(
     )
 
 
+@router.get(
+    "/admin/diagnostics",
+    response_model=SearchDiagnosticsResponse,
+)
+def api_search_diagnostics(
+    request: Request,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> SearchDiagnosticsResponse:
+    _require_admin(principal)
+    registry = _registry(request)
+    return SearchDiagnosticsResponse.model_validate(
+        _service(registry).diagnostics(session, principal)
+    )
+
+
+@router.post(
+    "/admin/reconcile-modules",
+    response_model=SearchModuleReconcileResponse,
+)
+def api_reconcile_search_modules(
+    request: Request,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> SearchModuleReconcileResponse:
+    _require_admin(principal)
+    result = _service(_registry(request)).reconcile_active_modules(
+        session,
+        tenant_id=principal.tenant_id,
+    )
+    audit_from_principal(
+        session,
+        principal,
+        action="search.modules.reconciled",
+        scope="tenant",
+        object_type="search_index",
+        details=result,
+    )
+    session.commit()
+    return SearchModuleReconcileResponse(**result)
+
+
+@router.post(
+    "/admin/changes/process",
+    response_model=SearchChangeDispatchResponse,
+)
+def api_process_search_changes(
+    request: Request,
+    limit: int = Query(default=100, ge=1, le=500),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> SearchChangeDispatchResponse:
+    _require_admin(principal)
+    result = _service(_registry(request)).process_changes(
+        session,
+        limit=limit,
+        tenant_id=principal.tenant_id,
+    )
+    audit_from_principal(
+        session,
+        principal,
+        action="search.changes.processed",
+        scope="tenant",
+        object_type="search_index",
+        details=result,
+    )
+    session.commit()
+    return SearchChangeDispatchResponse(**result)
+
+
+@router.post(
+    "/admin/rebuilds/{provider_id}/{resource_type}/start",
+    response_model=SearchRebuildResponse,
+)
+def api_start_search_rebuild(
+    provider_id: str,
+    resource_type: str,
+    request: Request,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> SearchRebuildResponse:
+    _require_admin(principal)
+    try:
+        state = _service(_registry(request)).start_rebuild(
+            session,
+            principal,
+            provider_id=provider_id,
+            resource_type=resource_type,
+        )
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+    audit_from_principal(
+        session,
+        principal,
+        action="search.rebuild.started",
+        scope="tenant",
+        object_type="search_index",
+        object_id=state.id,
+        details={
+            "provider_id": provider_id,
+            "resource_type": resource_type,
+            "rebuild_id": state.rebuild_id,
+        },
+    )
+    session.commit()
+    return SearchRebuildResponse(
+        state=_search_state_response(state)
+    )
+
+
+@router.post(
+    "/admin/rebuilds/{provider_id}/{resource_type}/continue",
+    response_model=SearchRebuildResponse,
+)
+def api_continue_search_rebuild(
+    provider_id: str,
+    resource_type: str,
+    request: Request,
+    limit: int = Query(default=100, ge=1, le=500),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> SearchRebuildResponse:
+    _require_admin(principal)
+    service = _service(_registry(request))
+    try:
+        state = service.continue_rebuild(
+            session,
+            principal,
+            provider_id=provider_id,
+            resource_type=resource_type,
+            limit=limit,
+        )
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=str(exc),
+        ) from exc
+    audit_from_principal(
+        session,
+        principal,
+        action="search.rebuild.continued",
+        scope="tenant",
+        object_type="search_index",
+        object_id=state.id,
+        details={
+            "provider_id": provider_id,
+            "resource_type": resource_type,
+            "status": state.status,
+            "indexed_documents": state.indexed_documents,
+        },
+    )
+    session.commit()
+    return SearchRebuildResponse(
+        state=_search_state_response(state)
+    )
+
+
+def _search_state_response(state: object) -> SearchIndexStateResponse:
+    return SearchIndexStateResponse(
+        provider_id=str(getattr(state, "provider_id")),
+        module_id=str(getattr(state, "module_id")),
+        resource_type=str(getattr(state, "resource_type")),
+        index_version=int(getattr(state, "index_version")),
+        status=str(getattr(state, "status")),
+        checkpoint_cursor=getattr(state, "checkpoint_cursor"),
+        high_watermark=getattr(state, "high_watermark"),
+        last_change_cursor=getattr(state, "last_change_cursor"),
+        indexed_documents=int(
+            getattr(state, "indexed_documents")
+        ),
+        rejected_documents=int(
+            getattr(state, "rejected_documents")
+        ),
+        rebuild_started_at=getattr(state, "rebuild_started_at"),
+        rebuild_completed_at=getattr(
+            state,
+            "rebuild_completed_at",
+        ),
+        last_success_at=getattr(state, "last_success_at"),
+        last_error=getattr(state, "last_error"),
+    )
+
+
 __all__ = ["router"]

src/govoplan_search/backend/schemas.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from datetime import datetime
 from typing import Any, Literal
 
 from pydantic import BaseModel, Field
@@ -31,6 +32,8 @@ class SearchResultResponse(BaseModel):
     breadcrumbs: list[str] = Field(default_factory=list)
     external_reference: SearchExternalReferenceResponse | None = None
     metadata: dict[str, Any] = Field(default_factory=dict)
+    source_revision: str | None = None
+    provenance: dict[str, Any] = Field(default_factory=dict)
 
 
 class SearchProviderDiagnosticResponse(BaseModel):
@@ -43,6 +46,8 @@ class SearchResponse(BaseModel):
     query: str
     results: list[SearchResultResponse]
     diagnostics: list[SearchProviderDiagnosticResponse] = Field(default_factory=list)
+    next_cursor: str | None = None
+    has_more: bool = False
 
 
 class SearchProviderResponse(BaseModel):
@@ -56,10 +61,58 @@ class SearchProviderListResponse(BaseModel):
     providers: list[SearchProviderResponse]
 
 
+class SearchIndexStateResponse(BaseModel):
+    provider_id: str
+    module_id: str
+    resource_type: str
+    index_version: int
+    status: str
+    checkpoint_cursor: str | None = None
+    high_watermark: str | None = None
+    last_change_cursor: str | None = None
+    indexed_documents: int = 0
+    rejected_documents: int = 0
+    rebuild_started_at: datetime | None = None
+    rebuild_completed_at: datetime | None = None
+    last_success_at: datetime | None = None
+    last_error: str | None = None
+
+
+class SearchDiagnosticsResponse(BaseModel):
+    backend: str
+    trigram_available: bool
+    queue: dict[str, int] = Field(default_factory=dict)
+    queue_oldest_age_seconds: float | None = None
+    states: list[SearchIndexStateResponse] = Field(
+        default_factory=list
+    )
+
+
+class SearchRebuildResponse(BaseModel):
+    state: SearchIndexStateResponse
+
+
+class SearchChangeDispatchResponse(BaseModel):
+    selected: int
+    applied: int
+    retrying: int
+    quarantined: int
+
+
+class SearchModuleReconcileResponse(BaseModel):
+    disabled_documents: int
+    enabled_documents: int
+
+
 __all__ = [
     "SearchProviderDiagnosticResponse",
     "SearchProviderListResponse",
     "SearchProviderResponse",
+    "SearchChangeDispatchResponse",
+    "SearchDiagnosticsResponse",
+    "SearchIndexStateResponse",
+    "SearchModuleReconcileResponse",
+    "SearchRebuildResponse",
     "SearchResponse",
     "SearchResultResponse",
 ]

tests/test_migrations.py

@@ -26,13 +26,15 @@ class SearchMigrationTests(unittest.TestCase):
             try:
                 with engine.connect() as connection:
                     self.assertIn(
-                        "a1b2c3d4e5f6",
+                        "b2c3d4e5f607",
                         set(MigrationContext.configure(connection).get_current_heads()),
                     )
                     self.assertEqual(
                         {
                             "search_index_acl_tokens",
+                            "search_index_change_queue",
                             "search_index_documents",
+                            "search_index_states",
                         },
                         {
                             name

tests/test_search_service.py

@@ -6,20 +6,108 @@ from types import SimpleNamespace
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 
-from govoplan_core.core.search import SearchDocument, SearchQuery
+from govoplan_core.core.search import (
+    SearchBackfillPage,
+    SearchDocument,
+    SearchIndexChange,
+    SearchQuery,
+    SearchResourceType,
+    SearchResult,
+)
 from govoplan_core.db.base import Base
 from govoplan_search.backend.db.models import (
     SearchIndexAclToken,
+    SearchIndexChangeQueue,
     SearchIndexDocument,
+    SearchIndexState,
+)
+from govoplan_search.backend.service import (
+    SearchIndexService,
+    aggregate_search_page,
 )
-from govoplan_search.backend.service import SearchIndexService
 
 
 class _Registry:
+    def __init__(self, source=None, active_modules=("search", "cases")):
+        self.source = source
+        self.active_modules = active_modules
+
     def manifests(self):
+        return tuple(
+            SimpleNamespace(id=module_id)
+            for module_id in self.active_modules
+        )
+
+    def search_sources(self):
+        if self.source is None:
+            return ()
         return (
-            SimpleNamespace(id="search"),
+            (
-            SimpleNamespace(id="cases"),
+                SimpleNamespace(
+                    registration=SimpleNamespace(id="cases.records")
+                ),
+                self.source,
+            ),
+        )
+
+
+class _Source:
+    def __init__(self, pages=()):
+        self.pages = list(pages)
+        self.authorized_ids = {"case-1"}
+
+    def resource_types(self):
+        return (
+            SearchResourceType(
+                provider_id="cases.records",
+                module_id="cases",
+                resource_type="case",
+                label="Cases",
+                requires_authorization_recheck=True,
+            ),
+        )
+
+    def backfill(self, session, *, request):
+        del session, request
+        return self.pages.pop(0)
+
+    def authorize(self, session, principal, *, requests):
+        del session, principal
+        return {
+            request.reference.key: (
+                request.reference.resource_id
+                in self.authorized_ids
+            )
+            for request in requests
+        }
+
+
+class _ResultProvider:
+    def search(self, session, principal, *, query):
+        del session, principal
+        return tuple(
+            SearchResult(
+                provider_id="ignored",
+                module_id="cases",
+                resource_type="case",
+                resource_id=f"case-{number}",
+                title=f"Permit {number}",
+                url=f"/cases/case-{number}",
+                score=float(10 - number),
+            )
+            for number in range(1, 7)
+        )[: query.limit]
+
+
+class _AggregateRegistry:
+    def search_providers(self):
+        return (
+            (
+                SimpleNamespace(
+                    registration=SimpleNamespace(id="cases.live")
+                ),
+                _ResultProvider(),
+            ),
         )
 
 
@@ -31,6 +119,8 @@ class SearchServiceTests(unittest.TestCase):
             tables=(
                 SearchIndexDocument.__table__,
                 SearchIndexAclToken.__table__,
+                SearchIndexState.__table__,
+                SearchIndexChangeQueue.__table__,
             ),
         )
         self.session = Session(self.engine)
@@ -164,6 +254,267 @@ class SearchServiceTests(unittest.TestCase):
                 ),
             )
 
+        with self.assertRaisesRegex(ValueError, "secret field"):
+            self.service.upsert_document(
+                self.session,
+                self.principal,
+                document=SearchDocument(
+                    tenant_id="tenant-1",
+                    module_id="cases",
+                    resource_type="case",
+                    resource_id="case-secret",
+                    title="Permit",
+                    url="/cases/case-secret",
+                    acl_tokens=("account:account-1",),
+                    metadata={"access_token": "must-not-index"},
+                ),
+            )
+
+    def test_query_rejects_cross_tenant_principal(self) -> None:
+        with self.assertRaises(PermissionError):
+            self.service.search(
+                self.session,
+                self.principal,
+                query=SearchQuery(
+                    text="permit",
+                    tenant_id="tenant-2",
+                ),
+            )
+
+    def test_source_authorization_recheck_is_fail_closed(self) -> None:
+        source = _Source()
+        service = SearchIndexService(_Registry(source))
+        for resource_id in ("case-1", "case-2"):
+            service.upsert_document(
+                self.session,
+                self.principal,
+                document=_source_document(resource_id),
+            )
+        self.session.flush()
+
+        results = service.search(
+            self.session,
+            self.principal,
+            query=SearchQuery(
+                text="permit",
+                tenant_id="tenant-1",
+            ),
+        )
+        self.assertEqual(
+            ["case-1"],
+            [item.resource_id for item in results],
+        )
+        unavailable_results = SearchIndexService(
+            _Registry()
+        ).search(
+            self.session,
+            self.principal,
+            query=SearchQuery(
+                text="permit",
+                tenant_id="tenant-1",
+            ),
+        )
+        self.assertEqual((), unavailable_results)
+
+    def test_durable_change_queue_is_idempotent(self) -> None:
+        document = _source_document("case-1")
+        change = SearchIndexChange(
+            change_id="change-1",
+            provider_id="cases.records",
+            kind="upsert",
+            reference=document.reference,
+            source_revision=document.source_revision,
+            cursor=document.change_cursor or "",
+            document=document,
+        )
+
+        self.assertTrue(
+            self.service.enqueue_change(
+                self.session,
+                change=change,
+            )
+        )
+        self.assertFalse(
+            self.service.enqueue_change(
+                self.session,
+                change=change,
+            )
+        )
+        result = self.service.process_changes(self.session)
+        self.assertEqual(1, result["applied"])
+        self.assertEqual(
+            "case-1",
+            self.session.query(SearchIndexDocument).one().resource_id,
+        )
+        self.assertEqual(
+            "applied",
+            self.session.query(SearchIndexChangeQueue).one().status,
+        )
+
+        delete_change = SearchIndexChange(
+            change_id="change-2",
+            provider_id="cases.records",
+            kind="delete",
+            reference=document.reference,
+            source_revision="3",
+            cursor="cursor-2",
+        )
+        self.service.enqueue_change(
+            self.session,
+            change=delete_change,
+        )
+        self.service.process_changes(self.session)
+        self.assertEqual(
+            0,
+            self.session.query(SearchIndexDocument).count(),
+        )
+
+    def test_rebuild_resumes_and_removes_stale_documents(self) -> None:
+        stale = _source_document("stale-case")
+        self.service.upsert_document(
+            self.session,
+            self.principal,
+            document=stale,
+        )
+        source = _Source(
+            pages=(
+                SearchBackfillPage(
+                    documents=(_source_document("case-1"),),
+                    next_cursor="page-2",
+                    complete=False,
+                    high_watermark="changes-9",
+                ),
+                SearchBackfillPage(
+                    documents=(_source_document("case-2"),),
+                    next_cursor=None,
+                    complete=True,
+                    high_watermark="changes-9",
+                ),
+            )
+        )
+        service = SearchIndexService(_Registry(source))
+
+        started = service.start_rebuild(
+            self.session,
+            self.principal,
+            provider_id="cases.records",
+            resource_type="case",
+        )
+        first_page = service.continue_rebuild(
+            self.session,
+            self.principal,
+            provider_id="cases.records",
+            resource_type="case",
+        )
+        self.assertEqual(started.id, first_page.id)
+        self.assertEqual("backfilling", first_page.status)
+        self.assertEqual("page-2", first_page.checkpoint_cursor)
+
+        complete = service.continue_rebuild(
+            self.session,
+            self.principal,
+            provider_id="cases.records",
+            resource_type="case",
+        )
+        self.assertEqual("ready", complete.status)
+        self.assertEqual("changes-9", complete.high_watermark)
+        self.assertEqual(
+            ["case-1", "case-2"],
+            sorted(
+                item.resource_id
+                for item in self.session.query(
+                    SearchIndexDocument
+                ).all()
+            ),
+        )
+
+    def test_module_reconciliation_disables_derived_rows(self) -> None:
+        self.service.upsert_document(
+            self.session,
+            self.principal,
+            document=SearchDocument(
+                tenant_id="tenant-1",
+                module_id="cases",
+                resource_type="case",
+                resource_id="case-1",
+                title="Permit",
+                url="/cases/case-1",
+                acl_tokens=("account:account-1",),
+            ),
+        )
+        disabled_service = SearchIndexService(
+            _Registry(active_modules=("search",))
+        )
+        result = disabled_service.reconcile_active_modules(
+            self.session
+        )
+        self.assertEqual(1, result["disabled_documents"])
+        self.assertFalse(
+            self.session.query(SearchIndexDocument).one().active
+        )
+
+    def test_aggregate_cursor_is_stable_and_non_overlapping(self) -> None:
+        registry = _AggregateRegistry()
+        first = aggregate_search_page(
+            registry,
+            self.session,
+            self.principal,
+            query=SearchQuery(
+                text="permit",
+                tenant_id="tenant-1",
+                limit=2,
+            ),
+        )
+        second = aggregate_search_page(
+            registry,
+            self.session,
+            self.principal,
+            query=SearchQuery(
+                text="permit",
+                tenant_id="tenant-1",
+                limit=2,
+                cursor=first.next_cursor,
+            ),
+        )
+
+        self.assertIsNotNone(first.next_cursor)
+        self.assertEqual(
+            {"case-1", "case-2"},
+            {item.resource_id for item in first.results},
+        )
+        self.assertEqual(
+            {"case-3", "case-4"},
+            {item.resource_id for item in second.results},
+        )
+        with self.assertRaisesRegex(ValueError, "cursor"):
+            aggregate_search_page(
+                registry,
+                self.session,
+                self.principal,
+                query=SearchQuery(
+                    text="different",
+                    tenant_id="tenant-1",
+                    limit=2,
+                    cursor=first.next_cursor,
+                ),
+            )
+
+
+def _source_document(resource_id: str) -> SearchDocument:
+    return SearchDocument(
+        tenant_id="tenant-1",
+        module_id="cases",
+        resource_type="case",
+        resource_id=resource_id,
+        title=f"Permit {resource_id}",
+        url=f"/cases/{resource_id}",
+        acl_tokens=("account:account-1",),
+        provider_id="cases.records",
+        source_revision="2",
+        change_cursor=f"cursor-{resource_id}",
+        requires_authorization_recheck=True,
+    )
+
 
 if __name__ == "__main__":
     unittest.main()

webui/src/api/search.ts

@@ -27,6 +27,8 @@ export type SearchResult = {
   breadcrumbs: string[];
   external_reference?: SearchExternalReference | null;
   metadata: Record<string, unknown>;
+  source_revision?: string | null;
+  provenance: Record<string, unknown>;
 };
 
 export type SearchResponse = {
@@ -37,6 +39,8 @@ export type SearchResponse = {
     status: "unavailable";
     message: string;
   }>;
+  next_cursor?: string | null;
+  has_more: boolean;
 };
 
 export type SearchRequest = {
@@ -47,6 +51,8 @@ export type SearchRequest = {
   contextId?: string;
   limit?: number;
   offset?: number;
+  cursor?: string;
+  language?: string;
 };
 
 export function search(
@@ -63,7 +69,9 @@ export function search(
       context_kind: request.contextKind,
       context_id: request.contextId,
       limit: request.limit,
-      offset: request.offset
+      offset: request.offset,
+      cursor: request.cursor,
+      language: request.language
     }),
     { signal }
   );

webui/src/features/search/SearchPage.tsx

@@ -1,5 +1,5 @@
-import { ExternalLink, Search } from "lucide-react";
+import { ChevronDown, ExternalLink, Search } from "lucide-react";
-import { useEffect, useMemo, useState, type FormEvent } from "react";
+import { useCallback, useEffect, useMemo, useState, type FormEvent } from "react";
 import { useSearchParams } from "react-router-dom";
 import {
   DismissibleAlert,
@@ -15,26 +15,31 @@ export default function SearchPage({ settings }: PlatformRouteContext) {
   const navigate = useGuardedNavigate();
   const [params, setParams] = useSearchParams();
   const query = params.get("q") ?? "";
-  const modules = params.getAll("module");
+  const moduleKey = params.getAll("module").join("\u001f");
-  const resourceTypes = params.getAll("resource_type");
+  const resourceTypeKey = params.getAll("resource_type").join("\u001f");
+  const contextId = params.get("context") ?? undefined;
+  const modules = useMemo(
+    () => moduleKey ? moduleKey.split("\u001f") : [],
+    [moduleKey]
+  );
+  const resourceTypes = useMemo(
+    () => resourceTypeKey ? resourceTypeKey.split("\u001f") : [],
+    [resourceTypeKey]
+  );
   const [draft, setDraft] = useState(query);
   const [response, setResponse] = useState<SearchResponse | null>(null);
   const [loading, setLoading] = useState(false);
   const [error, setError] = useState("");
   const requestKey = useMemo(
-    () => JSON.stringify([query, modules, resourceTypes]),
+    () => JSON.stringify([query, moduleKey, resourceTypeKey, contextId]),
-    [query, modules, resourceTypes]
+    [contextId, moduleKey, query, resourceTypeKey]
   );
 
   useEffect(() => {
     setDraft(query);
   }, [query]);
 
-  useEffect(() => {
+  const loadResults = useCallback((cursor?: string) => {
-    if (!query.trim()) {
-      setResponse(null);
-      return;
-    }
     const controller = new AbortController();
     setLoading(true);
     setError("");
@@ -45,20 +50,48 @@ export default function SearchPage({ settings }: PlatformRouteContext) {
         modules,
         resourceTypes,
         contextKind: modules.length ? "module" : "global",
-        contextId: params.get("context") ?? undefined,
+        contextId,
-        limit: 100
+        limit: 50,
+        cursor
       },
       controller.signal
     ).
-      then(setResponse).
+      then((next) => {
+        setResponse((current) =>
+          cursor && current ?
+            {
+              ...next,
+              results: [...current.results, ...next.results],
+              diagnostics: [
+                ...current.diagnostics,
+                ...next.diagnostics.filter((item) =>
+                  !current.diagnostics.some(
+                    (currentItem) => currentItem.provider_id === item.provider_id
+                  )
+                )
+              ]
+            } :
+            next
+        );
+      }).
       catch((reason) => {
         if ((reason as Error).name !== "AbortError") {
           setError(reason instanceof Error ? reason.message : "Search failed.");
         }
       }).
       finally(() => setLoading(false));
+    return controller;
+  }, [contextId, modules, query, resourceTypes, settings]);
+
+  useEffect(() => {
+    if (!query.trim()) {
+      setResponse(null);
+      return;
+    }
+    setResponse(null);
+    const controller = loadResults();
     return () => controller.abort();
-  }, [requestKey, settings]);
+  }, [loadResults, requestKey]);
 
   function submit(event: FormEvent) {
     event.preventDefault();
@@ -123,6 +156,16 @@ export default function SearchPage({ settings }: PlatformRouteContext) {
             </button>
           )}
         </div>
+        {response?.next_cursor &&
+        <button
+          type="button"
+          className="search-load-more"
+          disabled={loading}
+          onClick={() => loadResults(response.next_cursor ?? undefined)}>
+            <ChevronDown size={16} />
+            <span>Load more</span>
+          </button>
+        }
       </PageScrollViewport>
     </main>
   );

webui/src/styles/search.css

@@ -222,6 +222,36 @@
   font-size: 12px;
 }
 
+.search-load-more {
+  display: inline-flex;
+  align-items: center;
+  gap: 7px;
+  margin-top: 14px;
+  border: 1px solid var(--control-border);
+  border-radius: 4px;
+  background: linear-gradient(
+    var(--control-gradient-start),
+    var(--control-gradient-end)
+  );
+  color: var(--control-text);
+  cursor: pointer;
+  padding: 7px 12px;
+  font: inherit;
+  font-weight: 700;
+}
+
+.search-load-more:hover:not(:disabled) {
+  background: linear-gradient(
+    var(--control-gradient-start),
+    var(--control-gradient-end-hover)
+  );
+}
+
+.search-load-more:disabled {
+  cursor: default;
+  opacity: 0.55;
+}
+
 @media (max-width: 900px) {
   .global-search {
     width: 34px;